1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906
|
"""
Brian's logging module.
Preferences
-----------
.. document_brian_prefs:: logging
"""
import atexit
import logging
import logging.handlers
import os
import shutil
import sys
import tempfile
import time
from logging.handlers import RotatingFileHandler
from warnings import warn
import numpy
try:
import scipy
except ImportError:
scipy = None
import sympy
import brian2
from brian2.core.preferences import BrianPreference, prefs
from .environment import running_from_ipython
__all__ = ["get_logger", "BrianLogger", "std_silent"]
# ===============================================================================
# Logging preferences
# ===============================================================================
def log_level_validator(log_level):
log_levels = ("CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "DIAGNOSTIC")
return log_level.upper() in log_levels
#: Our new log level for more detailed debug output (mostly useful for debugging
#: Brian itself, not for user scripts)
DIAGNOSTIC = 5
#: Translation from string representation to number
LOG_LEVELS = {
"CRITICAL": logging.CRITICAL,
"ERROR": logging.ERROR,
"WARNING": logging.WARNING,
"INFO": logging.INFO,
"DEBUG": logging.DEBUG,
"DIAGNOSTIC": DIAGNOSTIC,
}
logging.addLevelName(DIAGNOSTIC, "DIAGNOSTIC")
if "logging" not in prefs.pref_register:
# Duplicate import of this module can happen when the documentation is built
prefs.register_preferences(
"logging",
"Logging system preferences",
delete_log_on_exit=BrianPreference(
default=True,
docs="""
Whether to delete the log and script file on exit.
If set to ``True`` (the default), log files (and the copy of the main
script) will be deleted after the brian process has exited, unless an
uncaught exception occurred. If set to ``False``, all log files will be
kept.
""",
),
file_log_level=BrianPreference(
default="DEBUG",
docs="""
What log level to use for the log written to the log file.
In case file logging is activated (see `logging.file_log`), which log
level should be used for logging. Has to be one of CRITICAL, ERROR,
WARNING, INFO, DEBUG or DIAGNOSTIC.
""",
validator=log_level_validator,
),
console_log_level=BrianPreference(
default="INFO",
docs="""
What log level to use for the log written to the console.
Has to be one of CRITICAL, ERROR, WARNING, INFO, DEBUG or DIAGNOSTIC.
""",
validator=log_level_validator,
),
file_log=BrianPreference(
default=True,
docs="""
Whether to log to a file or not.
If set to ``True`` (the default), logging information will be written
to a file. The log level can be set via the `logging.file_log_level`
preference.
""",
),
file_log_max_size=BrianPreference(
default=10000000,
docs="""
The maximum size for the debug log before it will be rotated.
If set to any value ``> 0``, the debug log will be rotated once
this size is reached. Rotating the log means that the old debug log
will be moved into a file in the same directory but with suffix ``".1"``
and the a new log file will be created with the same pathname as the
original file. Only one backup is kept; if a file with suffix ``".1"``
already exists when rotating, it will be overwritten.
If set to ``0``, no log rotation will be applied.
The default setting rotates the log file after 10MB.
""",
),
save_script=BrianPreference(
default=True,
docs="""
Whether to save a copy of the script that is run.
If set to ``True`` (the default), a copy of the currently run script
is saved to a temporary location. It is deleted after a successful
run (unless `logging.delete_log_on_exit` is ``False``) but is kept after
an uncaught exception occured. This can be helpful for debugging,
in particular when several simulations are running in parallel.
""",
),
std_redirection=BrianPreference(
default=True,
docs="""
Whether or not to redirect stdout/stderr to null at certain places.
This silences a lot of annoying compiler output, but will also hide
error messages making it harder to debug problems. You can always
temporarily switch it off when debugging. If
`logging.std_redirection_to_file` is set to ``True`` as well, then the
output is saved to a file and if an error occurs the name of this file
will be printed.
""",
),
std_redirection_to_file=BrianPreference(
default=True,
docs="""
Whether to redirect stdout/stderr to a file.
If both ``logging.std_redirection`` and this preference are set to
``True``, all standard output/error (most importantly output from
the compiler) will be stored in files and if an error occurs the name
of this file will be printed. If `logging.std_redirection` is ``True``
and this preference is ``False``, then all standard output/error will
be completely suppressed, i.e. neither be displayed nor stored in a
file.
The value of this preference is ignore if `logging.std_redirection` is
set to ``False``.
""",
),
display_brian_error_message=BrianPreference(
default=True,
docs="""
Whether to display a text for uncaught errors, mentioning the location
of the log file, the mailing list and the github issues.
Defaults to ``True``.""",
),
)
# ===============================================================================
# Initial setup
# ===============================================================================
def _encode(text):
"""Small helper function to encode unicode strings as UTF-8."""
return text.encode("UTF-8")
UNHANDLED_ERROR_MESSAGE = (
"Brian 2 encountered an unexpected error. "
"If you think this is a bug in Brian 2, please report this issue either to the "
"discourse forum at <http://brian.discourse.group/>, "
"or to the issue tracker at <https://github.com/brian-team/brian2/issues>."
)
def brian_excepthook(exc_type, exc_obj, exc_tb):
"""
Display a message mentioning the debug log in case of an uncaught
exception.
"""
# Do not catch Ctrl+C
if exc_type == KeyboardInterrupt:
return
logger = logging.getLogger("brian2")
BrianLogger.exception_occured = True
if not prefs["logging.display_brian_error_message"]:
# Put the exception message in the log file, but do not log to the
# console
if BrianLogger.console_handler is not None:
logger.removeHandler(BrianLogger.console_handler)
logger.exception("An exception occured", exc_info=(exc_type, exc_obj, exc_tb))
if BrianLogger.console_handler is not None:
logger.addHandler(BrianLogger.console_handler)
# Run the default except hook
return sys.__excepthook__(exc_type, exc_obj, exc_tb)
message = UNHANDLED_ERROR_MESSAGE
if BrianLogger.tmp_log is not None:
message += (
" Please include this file with debug information in your "
f"report: {BrianLogger.tmp_log} "
)
if BrianLogger.tmp_script is not None:
message += (
" Additionally, you can also include a copy "
"of the script that was run, available "
f"at: {BrianLogger.tmp_script}"
)
if hasattr(std_silent, "dest_fname_stdout"):
stdout = std_silent.dest_fname_stdout
stderr = std_silent.dest_fname_stderr
message += (
" You can also include a copy of the "
"redirected std stream outputs, available at "
f"'{stdout}' and '{stderr}'."
)
message += " Thanks!" # very important :)
logger.error(message, exc_info=(exc_type, exc_obj, exc_tb))
def clean_up_logging():
"""
Shutdown the logging system and delete the debug log file if no error
occured.
"""
BrianLogger.initialized = False
logging.shutdown()
if not BrianLogger.exception_occured and prefs["logging.delete_log_on_exit"]:
if BrianLogger.tmp_log is not None:
try:
os.remove(BrianLogger.tmp_log)
except OSError as exc:
warn(f"Could not delete log file: {exc}")
# Remove log files that have been rotated (currently only one)
rotated_log = f"{BrianLogger.tmp_log}.1"
if os.path.exists(rotated_log):
try:
os.remove(rotated_log)
except OSError as exc:
warn(f"Could not delete log file: {exc}")
if BrianLogger.tmp_script is not None:
try:
os.remove(BrianLogger.tmp_script)
except OSError as exc:
warn(f"Could not delete copy of script file: {exc}")
std_silent.close()
atexit.register(clean_up_logging)
class HierarchyFilter:
"""
A class for suppressing all log messages in a subtree of the name hierarchy.
Does exactly the opposite as the `logging.Filter` class, which allows
messages in a certain name hierarchy to *pass*.
Parameters
----------
name : str
The name hiearchy to suppress. See `BrianLogger.suppress_hierarchy` for
details.
"""
def __init__(self, name):
self.orig_filter = logging.Filter(name)
def filter(self, record):
"""
Filter out all messages in a subtree of the name hierarchy.
"""
# do the opposite of what the standard filter class would do
return not self.orig_filter.filter(record)
class NameFilter:
"""
A class for suppressing log messages ending with a certain name.
Parameters
----------
name : str
The name to suppress. See `BrianLogger.suppress_name` for details.
"""
def __init__(self, name):
self.name = name
def filter(self, record):
"""
Filter out all messages ending with a certain name.
"""
# The last part of the name
record_name = record.name.split(".")[-1]
return self.name != record_name
class RemoveBrian2Filter(logging.Filter):
"""
A class for removing the ``brian2`` prefix from log messages.
Will be used for extension packages that use the Brian logging system.
"""
def filter(self, record):
assert record.name.startswith("brian2."), record.name
record.name = record.name[7:]
return True
class BrianLogger:
"""
Convenience object for logging. Call `get_logger` to get an instance of
this class.
Parameters
----------
name : str
The name used for logging, normally the name of the module. If the name
does not start with ``brian2.``, it will be prepended automatically when
interacting with the `logging` module. This means that from the logging's
system point of view, it will use the configuration for the logger in the
``brian2`` hierachy. However, when displaying the name in the log messages,
the ``brian2.`` prefix will be removed. This is useful for extension
modules, which can use the Brian logging system, but will be displayed as
``myextension`` instead of ``brian2.myextension``. This is also used in
Brian's test suite, which only considers log messages starting with
``brian2``.
"""
#: Global flag to know whether the logging system is in a usable state
initialized = False
#: Class attribute to remember whether any exception occured
exception_occured = False
#: Class attribute for remembering log messages that should only be
#: displayed once
_log_messages = set()
#: The name of the temporary log file (by default deleted after the run if
#: no exception occurred), if any
tmp_log = None
#: The `logging.FileHandler` responsible for logging to the temporary log
#: file
file_handler = None
#: The `logging.StreamHandler` responsible for logging to the console
console_handler = None
#: The name of the temporary copy of the main script file (by default
#: deleted after the run if no exception occurred), if any
tmp_script = None
#: The pid of the process that initialized the logger – used to switch off file logging in
#: multiprocessing contexts
_pid = None
def __init__(self, name):
if not name.startswith("brian2."):
self.name = "brian2." + name
self.filter_name = True
else:
self.name = name
self.filter_name = False
def _log(self, log_level, msg, name_suffix, once):
"""
Log an entry.
Parameters
----------
log_level : {'debug', 'info', 'warn', 'error'}
The level with which to log the message.
msg : str
The log message.
name_suffix : str
A suffix that will be added to the logger name.
once : bool
Whether to suppress identical messages if they are logged again.
"""
if not BrianLogger.initialized:
# Prevent logging errors on exit
return
name = self.name
if name_suffix:
name += f".{name_suffix}"
# Switch off file logging when using multiprocessing
if BrianLogger.tmp_log is not None and BrianLogger._pid != os.getpid():
BrianLogger.tmp_log = None
logging.getLogger("brian2").removeHandler(BrianLogger.file_handler)
BrianLogger.file_handler = None
if once:
# Check whether this exact message has already been displayed
log_tuple = (name, log_level, msg)
if log_tuple in BrianLogger._log_messages:
return
else:
BrianLogger._log_messages.add(log_tuple)
the_logger = logging.getLogger(name)
if self.filter_name:
filter = RemoveBrian2Filter()
the_logger.addFilter(filter)
the_logger.log(LOG_LEVELS[log_level], msg)
if self.filter_name:
the_logger.removeFilter(filter)
def diagnostic(self, msg, name_suffix=None, once=False):
"""
Log a diagnostic message.
Parameters
----------
msg : str
The message to log.
name_suffix : str, optional
A suffix to add to the name, e.g. a class or function name.
once : bool, optional
Whether this message should be logged only once and not repeated
if sent another time.
"""
self._log("DIAGNOSTIC", msg, name_suffix, once)
def debug(self, msg, name_suffix=None, once=False):
"""
Log a debug message.
Parameters
----------
msg : str
The message to log.
name_suffix : str, optional
A suffix to add to the name, e.g. a class or function name.
once : bool, optional
Whether this message should be logged only once and not repeated
if sent another time.
"""
self._log("DEBUG", msg, name_suffix, once)
def info(self, msg, name_suffix=None, once=False):
"""
Log an info message.
Parameters
----------
msg : str
The message to log.
name_suffix : str, optional
A suffix to add to the name, e.g. a class or function name.
once : bool, optional
Whether this message should be logged only once and not repeated
if sent another time.
"""
self._log("INFO", msg, name_suffix, once)
def warn(self, msg, name_suffix=None, once=False):
"""
Log a warn message.
Parameters
----------
msg : str
The message to log.
name_suffix : str, optional
A suffix to add to the name, e.g. a class or function name.
once : bool, optional
Whether this message should be logged only once and not repeated
if sent another time.
"""
self._log("WARNING", msg, name_suffix, once)
warning = warn
def error(self, msg, name_suffix=None, once=False):
"""
Log an error message.
Parameters
----------
msg : str
The message to log.
name_suffix : str, optional
A suffix to add to the name, e.g. a class or function name.
once : bool, optional
Whether this message should be logged only once and not repeated
if sent another time.
"""
self._log("ERROR", msg, name_suffix, once)
@staticmethod
def _suppress(filterobj, filter_log_file):
"""
Apply a filter object to log messages.
Parameters
----------
filterobj : `logging.Filter`
A filter object to apply to log messages.
filter_log_file : bool
Whether the filter also applies to log messages in the log file.
"""
BrianLogger.console_handler.addFilter(filterobj)
if filter_log_file:
BrianLogger.file_handler.addFilter(filterobj)
@staticmethod
def suppress_hierarchy(name, filter_log_file=False):
"""
Suppress all log messages in a given hiearchy.
Parameters
----------
name : str
Suppress all log messages in the given `name` hierarchy. For
example, specifying ``'brian2'`` suppresses all messages logged
by Brian, specifying ``'brian2.codegen'`` suppresses all messages
generated by the code generation modules.
filter_log_file : bool, optional
Whether to suppress the messages also in the log file. Defaults to
``False`` meaning that suppressed messages are not displayed on
the console but are still saved to the log file.
"""
suppress_filter = HierarchyFilter(name)
BrianLogger._suppress(suppress_filter, filter_log_file)
@staticmethod
def suppress_name(name, filter_log_file=False):
"""
Suppress all log messages with a given name.
Parameters
----------
name : str
Suppress all log messages ending in the given `name`. For
example, specifying ``'resolution_conflict'`` would suppress
messages with names such as
``brian2.equations.codestrings.CodeString.resolution_conflict`` or
``brian2.equations.equations.Equations.resolution_conflict``.
filter_log_file : bool, optional
Whether to suppress the messages also in the log file. Defaults to
``False`` meaning that suppressed messages are not displayed on
the console but are still saved to the log file.
"""
suppress_filter = NameFilter(name)
BrianLogger._suppress(suppress_filter, filter_log_file)
@staticmethod
def log_level_diagnostic():
"""
Set the log level to "diagnostic".
"""
BrianLogger.console_handler.setLevel(DIAGNOSTIC)
@staticmethod
def log_level_debug():
"""
Set the log level to "debug".
"""
BrianLogger.console_handler.setLevel(logging.DEBUG)
@staticmethod
def log_level_info():
"""
Set the log level to "info".
"""
BrianLogger.console_handler.setLevel(logging.INFO)
@staticmethod
def log_level_warn():
"""
Set the log level to "warn".
"""
BrianLogger.console_handler.setLevel(logging.WARN)
@staticmethod
def log_level_error():
"""
Set the log level to "error".
"""
BrianLogger.console_handler.setLevel(logging.ERROR)
@staticmethod
def initialize():
"""
Initialize Brian's logging system. This function will be called
automatically when Brian is imported.
"""
# get the main logger
logger = logging.getLogger("brian2")
logger.propagate = False
logger.setLevel(LOG_LEVELS["DIAGNOSTIC"])
# Log to a file
if prefs["logging.file_log"]:
try:
# Temporary filename used for logging
with tempfile.NamedTemporaryFile(
prefix="brian_debug_", suffix=".log", delete=False
) as tmp_f:
BrianLogger.tmp_log = tmp_f.name
# Remove any previously existing file handler
if BrianLogger.file_handler is not None:
BrianLogger.file_handler.close()
logger.removeHandler(BrianLogger.file_handler)
# Rotate log file after prefs['logging.file_log_max_size'] bytes and keep one copy
BrianLogger.file_handler = RotatingFileHandler(
BrianLogger.tmp_log,
mode="a",
maxBytes=prefs["logging.file_log_max_size"],
backupCount=1,
encoding="utf-8",
)
BrianLogger.file_handler.setLevel(
LOG_LEVELS[prefs["logging.file_log_level"].upper()]
)
BrianLogger.file_handler.setFormatter(
logging.Formatter(
"%(asctime)s %(levelname)-10s %(name)s: %(message)s"
)
)
logger.addHandler(BrianLogger.file_handler)
BrianLogger._pid = os.getpid()
except OSError as ex:
warn(f"Could not create log file: {ex}")
# Save a copy of the script
BrianLogger.tmp_script = None
if prefs["logging.save_script"]:
if (
len(sys.argv[0])
and not running_from_ipython()
and os.path.isfile(sys.argv[0])
):
try:
tmp_file = tempfile.NamedTemporaryFile(
prefix="brian_script_", suffix=".py", delete=False
)
with tmp_file:
# Timestamp
tmp_file.write(_encode(f"# {time.asctime()}\n"))
# Command line arguments
tmp_file.write(_encode(f"# Run as: {' '.join(sys.argv)}\n\n"))
# The actual script file
# TODO: We are copying the script file as it is, this might clash
# with the encoding we used for the comments added above
with open(os.path.abspath(sys.argv[0]), "rb") as script_file:
shutil.copyfileobj(script_file, tmp_file)
BrianLogger.tmp_script = tmp_file.name
except OSError as ex:
warn(f"Could not copy script file to temp directory: {ex}")
if BrianLogger.console_handler is not None:
logger.removeHandler(BrianLogger.console_handler)
# create console handler with a higher log level
BrianLogger.console_handler = logging.StreamHandler()
BrianLogger.console_handler.setLevel(
LOG_LEVELS[prefs["logging.console_log_level"]]
)
BrianLogger.console_handler.setFormatter(
logging.Formatter("%(levelname)-10s %(message)s [%(name)s]")
)
# add the handler to the logger
logger.addHandler(BrianLogger.console_handler)
# We want to log all warnings
logging.captureWarnings(True) # pylint: disable=E1101
# Manually connect to the warnings logger so that the warnings end up in
# the log file. Note that connecting to the console handler here means
# duplicated warning messages in the ipython notebook, but not doing so
# would mean that they are not displayed at all in the standard ipython
# interface...
warn_logger = logging.getLogger("py.warnings")
warn_logger.addHandler(BrianLogger.console_handler)
# Put some standard info into the log file
logger.log(
logging.DEBUG,
f"Logging to file: {BrianLogger.tmp_log}, copy of main script saved as:"
f" {BrianLogger.tmp_script}",
)
logger.log(logging.DEBUG, f"Python interpreter: {sys.executable}")
logger.log(logging.DEBUG, f"Platform: {sys.platform}")
version_infos = {
"brian": brian2.__version__,
"numpy": numpy.__version__,
"scipy": scipy.__version__ if scipy else "not installed",
"sympy": sympy.__version__,
"python": sys.version,
}
for _name, _version in version_infos.items():
logger.log(logging.DEBUG, f"{_name} version is: {str(_version)}")
# Handle uncaught exceptions
sys.excepthook = brian_excepthook
BrianLogger.initialized = True
def get_logger(module_name="brian2"):
"""
Get an object that can be used for logging.
Parameters
----------
module_name : str
The name used for logging, should normally be the module name as
returned by ``__name__``.
Returns
-------
logger : `BrianLogger`
"""
return BrianLogger(module_name)
class catch_logs:
"""
A context manager for catching log messages. Use this for testing the
messages that are logged. Defaults to catching warning/error messages and
this is probably the only real use case for testing. Note that while this
context manager is active, *all* log messages are suppressed. Using this
context manager returns a list of (log level, name, message) tuples.
Parameters
----------
log_level : int or str, optional
The log level above which messages are caught.
only_from : list, optional
A list of module names from which messages are caught. Defaults to
the ``brian2`` module.
Examples
--------
>>> logger = get_logger('brian2.logtest')
>>> logger.warn('An uncaught warning') # doctest: +SKIP
WARNING brian2.logtest: An uncaught warning
>>> with catch_logs() as l:
... logger.warn('a caught warning')
... print('l contains: %s' % l)
...
l contains: [('WARNING', 'brian2.logtest', 'a caught warning')]
"""
_entered = False
def __init__(self, log_level=logging.WARN, only_from=("brian2",)):
self.log_list = []
self.handler = LogCapture(self.log_list, log_level, only_from=only_from)
self._entered = False
def __enter__(self):
if self._entered:
raise RuntimeError(f"Cannot enter {self!r} twice")
self._entered = True
return self.log_list
def __exit__(self, *exc_info):
if not self._entered:
raise RuntimeError(f"Cannot exit {self!r} without entering first")
self.handler.uninstall()
class LogCapture(logging.Handler):
"""
A class for capturing log warnings. This class is used by
`~brian2.utils.logger.catch_logs` to allow testing in a similar
way as with `warnings.catch_warnings`.
"""
def __init__(self, log_list, log_level=logging.WARN, only_from=("brian2",)):
logging.Handler.__init__(self, level=log_level)
self.log_list = log_list
self.only_from = only_from
# make a copy of the previous handlers
self.handlers = list(logging.getLogger("brian2").handlers)
self.install()
def emit(self, record):
# Append a tuple consisting of (level, name, msg) to the list of
# log messages
if any(
record.name == name or record.name.startswith(name + ".")
for name in self.only_from
):
self.log_list.append((record.levelname, record.name, record.msg))
def install(self):
"""
Install this handler to catch all warnings. Temporarily disconnect all
other handlers.
"""
the_logger = logging.getLogger("brian2")
for handler in self.handlers:
the_logger.removeHandler(handler)
the_logger.addHandler(self)
def uninstall(self):
"""
Uninstall this handler and re-connect the previously installed
handlers.
"""
the_logger = logging.getLogger("brian2")
for handler in self.handlers:
the_logger.addHandler(handler)
# See http://stackoverflow.com/questions/26126160/redirecting-standard-out-in-err-back-after-os-dup2
# for an explanation of how this function works. Note that 1 and 2 are the file
# numbers for stdout and stderr
class std_silent:
"""
Context manager that temporarily silences stdout and stderr but keeps the
output saved in a temporary file and writes it if an exception is raised.
"""
dest_stdout = None
dest_stderr = None
def __init__(self, alwaysprint=False):
self.alwaysprint = alwaysprint or not prefs["logging.std_redirection"]
self.redirect_to_file = prefs["logging.std_redirection_to_file"]
if (
not self.alwaysprint
and self.redirect_to_file
and std_silent.dest_stdout is None
):
std_silent.dest_fname_stdout = tempfile.NamedTemporaryFile(
prefix="brian_stdout_", suffix=".log", delete=False
).name
std_silent.dest_fname_stderr = tempfile.NamedTemporaryFile(
prefix="brian_stderr_", suffix=".log", delete=False
).name
std_silent.dest_stdout = open(std_silent.dest_fname_stdout, "w")
std_silent.dest_stderr = open(std_silent.dest_fname_stderr, "w")
def __enter__(self):
if not self.alwaysprint and self.redirect_to_file:
sys.stdout.flush()
sys.stderr.flush()
self.orig_out_fd = os.dup(1)
self.orig_err_fd = os.dup(2)
os.dup2(std_silent.dest_stdout.fileno(), 1)
os.dup2(std_silent.dest_stderr.fileno(), 2)
def __exit__(self, exc_type, exc_value, traceback):
if not self.alwaysprint and self.redirect_to_file:
std_silent.dest_stdout.flush()
std_silent.dest_stderr.flush()
if exc_type is not None:
with open(std_silent.dest_fname_stdout) as f:
out = f.read()
with open(std_silent.dest_fname_stderr) as f:
err = f.read()
os.dup2(self.orig_out_fd, 1)
os.dup2(self.orig_err_fd, 2)
os.close(self.orig_out_fd)
os.close(self.orig_err_fd)
if exc_type is not None:
sys.stdout.write(out)
sys.stderr.write(err)
@classmethod
def close(cls):
if std_silent.dest_stdout is not None:
std_silent.dest_stdout.close()
if prefs["logging.delete_log_on_exit"]:
try:
os.remove(std_silent.dest_fname_stdout)
except OSError:
# TODO: this happens quite frequently - why?
# The file objects are closed as far as Python is concerned,
# but maybe Windows is still hanging on to them?
pass
if std_silent.dest_stderr is not None:
std_silent.dest_stderr.close()
if prefs["logging.delete_log_on_exit"]:
try:
os.remove(std_silent.dest_fname_stderr)
except OSError:
pass
|