auto-pi-lot / autopilot / 745

pending completion

Build # 745

Build Type

Pull #177

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/auto-pi-lot/autopilot/commit/<a class=hub.com/auto-pi-lot/autopilot/commit/b5de754f5df31cafb3589501d638309d6bf64b16">b5de754f5<a href="https://github.com/auto-pi-lot/autopilot/commit/b5de754f5df31cafb3589501d638309d6bf64b16">">Merge </a><a class="double-link" href="https://github.com/auto-pi-lot/autopilot/commit/<a class="double-link" href="https://github.com/auto-pi-lot/autopilot/commit/211be0e5f26d0ada2a1227d8eef144a1c2b35001">211be0e5f</a>">211be0e5f</a><a href="https://github.com/auto-pi-lot/autopilot/commit/b5de754f5df31cafb3589501d638309d6bf64b16"> into 90956187d">90956187d</a>

Pull Request Pull Request #177: v0.5.0a0 - The Data Modeling Edition

Run Details

3920 of 3920 new or added lines in 64 files covered. (100.0%)

3904 of 11502 relevant lines covered (33.94%)

0.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.3

/autopilot/utils/loggers.py

import os
import logging
import re
import multiprocessing as mp
import inspect
from pathlib import Path
from logging.handlers import RotatingFileHandler
from threading import Lock
import warnings
from rich.logging import RichHandler
from dataclasses import dataclass
from datetime import datetime
from parse import parse
import typing
from typing import Literal


from autopilot.root import Autopilot_Type

from autopilot import prefs

_LOGGERS = [] # type: list
"""
List of instantiated loggers, used in :func:`.init_logger` to return existing loggers without modification
"""

_INIT_LOCK = Lock() # type: Lock

LOGLEVELS = Literal['DEBUG', 'INFO', 'WARNING', 'ERROR']

def init_logger(instance=None, module_name=None, class_name=None, object_name=None) -> logging.Logger:
    """
    Initialize a logger

    Loggers are created such that...

    * There is one logger per module (eg. all gpio objects will log to hardware.gpio)
    * If the passed object has a ``name`` attribute, that name will be prefixed to its log messages in the file
    * The loglevel for the file handler and the stdout is determined by ``prefs.get('LOGLEVEL')``, and if none is provided ``WARNING`` is used by default
    * logs are rotated according to ``prefs.get('LOGSIZE')`` (in bytes) and ``prefs.get('LOGNUM')`` (number of backups of ``prefs.get('LOGSIZE')`` to cycle through)

    Logs are stored in ``prefs.get('LOGDIR')``, and are formatted like::

        "%(asctime)s - %(name)s - %(levelname)s : %(message)s"

    Loggers can be initialized either by passing an object to the first ``instance`` argument, or
    by specifying any of ``module_name`` , ``class_name`` , or ``object_name`` (at least one must be specified)
    which are combined with periods like ``module.class_name.object_name``

    Args:
        instance: The object that we are creating a logger for! if None, at least one of ``module, class_name, or object_name`` must be passed
        module_name (None, str): If no ``instance`` passed, the module name to create a logger for
        class_name (None, str): If no ``instance`` passed, the class name to create a logger for
        object_name (None, str): If no ``instance`` passed, the object name/id to create a logger for

    Returns:
        :class:`logging.logger`
    """

    # --------------------------------------------------
    # gather variables
    # --------------------------------------------------

    if instance is not None:
        # get name of module_name without prefixed autopilot
        # eg passed autopilot.hardware.gpio.Digital_In -> hardware.gpio
        # filtering leading 'autopilot' from string

        module_name = instance.__module__
        if "__main__" in module_name:
            # awkward workaround to get module name of __main__ run objects
            mod_obj = inspect.getmodule(instance)
            try:
                mod_suffix  = inspect.getmodulename(inspect.getmodule(instance).__file__)
                module_name = '.'.join([mod_obj.__package__, mod_suffix])
            except AttributeError:
                # when running interactively or from a plugin, __main__ does not have __file__
                module_name = "__main__"


        module_name = re.sub('^autopilot.', '', module_name)

        class_name = instance.__class__.__name__

        # if object is running in separate process, give it its own file
        if issubclass(instance.__class__, mp.Process):
            # it should be at least 2 (in case its first spawned in its module)
            # but otherwise nocollide
            p_num = 2
            _module_name = module_name
            module_name = f"{_module_name}_{str(p_num).zfill(2)}"
            if module_name in globals()['_LOGGERS']:
                for existing_mod in globals()['_LOGGERS']:
                    if module_name in existing_mod and re.match(r'\d$', existing_mod):
                        p_num += 1

                module_name = f"{_module_name}_{str(p_num).zfill(2)}"

        # get name of object if it has one
        if hasattr(instance, 'id'):
            object_name = str(instance.id)
        elif hasattr(instance, 'name'):
            object_name = str(instance.name)
        else:
            object_name = None

        # --------------------------------------------------
        # check if logger needs to be made, or exists already
        # --------------------------------------------------
    elif not any((module_name, class_name, object_name)):
        raise ValueError('Need to either give an object to create a logger for, or one of module_name, class_name, or object_name')


    # get name of logger to get
    logger_name_pieces = [v for v in (module_name, class_name, object_name) if v is not None]
    logger_name = '.'.join(logger_name_pieces)

    # trim __ from logger names, linux don't like to make things like that
    # re.sub(r"^\_\_")

    # --------------------------------------------------
    # if new logger must be made, make it, otherwise just return existing logger
    # --------------------------------------------------

    # use a lock to prevent loggers from being double-created, just to be extra careful
    with globals()['_INIT_LOCK']:

        # check if something starting with module_name already exists in loggers
        MAKE_NEW = False
        if not any([test_logger == module_name for test_logger in globals()['_LOGGERS']]):
            MAKE_NEW = True

        if MAKE_NEW:
            parent_logger = logging.getLogger(module_name)
            loglevel = getattr(logging, prefs.get('LOGLEVEL'))
            parent_logger.setLevel(loglevel)

            # make formatter that includes name
            log_formatter = logging.Formatter("[%(asctime)s] %(levelname)s [%(name)s]: %(message)s")

            ## file handler
            # base filename is the module_name + '.log
            base_filename = Path(prefs.get('LOGDIR')) / (module_name + '.log')

            fh = _file_handler(base_filename)
            fh.setLevel(loglevel)
            fh.setFormatter(log_formatter)
            parent_logger.addHandler(fh)

            # rich logging handler for stdout
            parent_logger.addHandler(_rich_handler())

            # if our parent is the rootlogger, disable propagation to avoid printing to stdout
            if isinstance(parent_logger.parent, logging.RootLogger):
                parent_logger.propagate = False

            ## log creation
            globals()['_LOGGERS'].append(module_name)
            parent_logger.debug(f'parent, module-level logger created: {module_name}')

        logger = logging.getLogger(logger_name)
        if logger_name not in globals()['_LOGGERS']:
        # logger.addHandler(_rich_handler())
            logger.debug(f"Logger created: {logger_name}")
            globals()['_LOGGERS'].append(logger_name)

    return logger


def _rich_handler() -> RichHandler:
    rich_handler = RichHandler(rich_tracebacks=True, markup=True)
    rich_formatter = logging.Formatter(
        "[bold green]\[%(name)s][/bold green] %(message)s",
        datefmt='[%y-%m-%dT%H:%M:%S]'
    )
    rich_handler.setFormatter(rich_formatter)
    return rich_handler

def _file_handler(base_filename: Path) -> RotatingFileHandler:
    # if directory doesn't exist, try to make it
    if not base_filename.parent.exists():
        base_filename.parent.mkdir(parents=True, exist_ok=True)

    try:
        fh = RotatingFileHandler(
            str(base_filename),
            mode='a',
            maxBytes=int(prefs.get('LOGSIZE')),
            backupCount=int(prefs.get('LOGNUM'))
        )
    except PermissionError as e:
        # catch permissions errors, try to chmod our way out of it
        try:
            for mod_file in Path(base_filename).parent.glob(f"{Path(base_filename).stem}*"):
                os.chmod(mod_file, 0o777)
                warnings.warn(f'Couldnt access {mod_file}, changed permissions to 0o777')

            fh = RotatingFileHandler(
                base_filename,
                mode='a',
                maxBytes=int(prefs.get('LOGSIZE')),
                backupCount=int(prefs.get('LOGNUM'))
            )
        except Exception as f:
            raise PermissionError(
                f'Couldnt open logfile {base_filename}, and couldnt chmod our way out of it.\n' + '-' * 20 + f'\ngot errors:\n{e}\n\n{f}\n' + '-' * 20)

    return fh

# --------------------------------------------------
# Parsers and in-memory representation of logs
# --------------------------------------------------

class ParseError(RuntimeError):
    """
    Error parsing a logfile
    """

def _convert_asc_timestamp(timestamp:str) -> datetime:
    hunk, ms = timestamp.split(',')
    ms += '000'
    timestamp = '.'.join([hunk, ms])
    return datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S.%f')

@dataclass
class Log_Format:
    format: str
    """A format string parseable by ``parse``"""
    example: str
    """An example string (that allows for testing)"""
    conversions: typing.Optional[typing.Dict[str, typing.Callable]] = None
    """A dictionary matching keys in the ``format`` string to callables for post-parsing coercion"""

    def parse(self, log_entry:str) -> dict:
        if self.conversions is None:
            conversions = {}
        else:
            conversions = self.conversions

        res = parse(self.format, log_entry, conversions)
        if res is None:
            raise ParseError(f'Could not parse {log_entry} with {self.format}')
        else:
            return res.named



LOG_FORMATS = (
    Log_Format(
        format = '{timestamp:Timestamp} - {name} - {level} : {message}',
        conversions = {
            'Timestamp': _convert_asc_timestamp
        },
        example = "2022-03-07 16:56:48,954 - networking.node.Net_Node._T - DEBUG : RECEIVED: ID: _testpi_9879; TO: T; SENDER: _testpi; KEY: DATA; FLAGS: {'NOREPEAT': True}; VALUE: {'trial_num': 1197, 'timestamp': '2022-03-01T23:52:16.995387', 'frequency': 45255.0, 'amplitude': 0.1, 'ramp': 5.0, 'pilot': 'testpi', 'subject': '0895'}"
    ),
    Log_Format(
        format = '[{timestamp:Timestamp}] {level} [{name}]: {message}',
        conversions = {
            'Timestamp': _convert_asc_timestamp
        },
        example = '[2022-03-09 16:13:43,224] INFO [networking.node]: parent, module-level logger created: networking.node'
    )
)
"""
Possible formats of logging messages (to allow change over versions) as a `parse string <https://github.com/r1chardj0n3s/parse>`_
"""

MESSAGE_FORMATS = {
    'node_msg_recv': '{action}: ID: {message_id}; TO: {to}; SENDER: {sender}; KEY: {key}; FLAGS: {flags}; VALUE: {value}',
    'node_msg_sent': '{action} - ID: {message_id}; TO: {to}; SENDER: {sender}; KEY: {key}; FLAGS: {flags}; VALUE: {value}'
}
"""
Additional parsing patterns for logged messages

* ``node_msg``: Logging messages from :class:`.networking.node.Net_Node`

"""




class LogEntry(Autopilot_Type):
    """
    Single entry in a log
    """
    timestamp: datetime
    name: str
    level: LOGLEVELS
    message: typing.Union[str, dict]

    def parse_message(self, format:typing.List[str]):
        """
        Parse the message using a format string specified as a key in the :data:`.MESSAGE_FORMATS` dictionary (or a format string itself)

        replaces the :attr:`.message` attribute.

        If parsing unsuccessful, no exception is raised because there are often messages that are not parseable in the logs!

        Args:
            format (typing.List[str]): List of format strings to try!

        Returns:

        """
        if isinstance(format, str):
            format = [format]

        format = [MESSAGE_FORMATS[f]if f in MESSAGE_FORMATS.keys() else f for f in format]

        for f in format:
            result = parse(f, self.message)
            if result is not None:
                self.message = result.named
                return


    @classmethod
    def from_string(cls,
                    entry:str,
                    parse_message:typing.Optional[typing.List[str]]=None) -> 'LogEntry':
        """
        Create a LogEntry by parsing a string.

        Try to parse using any of the possible :ref:`.LOG_FORMATS`, raising a
        :class:`.ParseError` if none are successful

        Args:
            entry (str): single line of a logging file
            parse_message (Optional[str]): Parse messages with the :data:`.MESSAGE_FORMATS` key or format string

        Returns:
            :class:`.LogEntry`

        Raises:
            :class:`.ParseError` if no messages are parsed

        """
        for aformat in LOG_FORMATS:
            try:
                result = aformat.parse(entry)
            except ParseError:
                # fine, we're searching for one that works
                continue

            entry = cls(**result)
            if parse_message is not None:
                entry.parse_message(parse_message)
            return entry

        # if we haven't returned anything, raise a parse error
        raise ParseError(f'Couldnt parse entry with any known log formats: {entry}')

class Log(Autopilot_Type):
    """
    Representation of a logfile in memory
    """
    entries: typing.List[LogEntry]

    @classmethod
    def from_logfile(cls,
                     file: typing.Union[Path, str],
                     include_backups:bool = True,
                     parse_messages: typing.Optional[typing.List[str]]=None ):
        """
        Load a logfile (and maybe its backups) from a logfile location

        Args:
            file (:class:`pathlib.Path`, str): If string, converted to Path. If relative (and relative file is not found),
                then attempts to find relative to ``prefs.LOGDIR``
            include_backups (bool): if ``True`` (default), try and load all of the backup logfiles (that have .1, .2, etc appended)
            parse_messages (Optional[str]): Parse messages with the :data:`.MESSAGE_FORMATS` key or format string

        Returns:
            :class:`.Log`
        """
        file = Path(file)
        if not file.exists():
            if file.is_absolute():
                relfile = file.relative_to(prefs.get('LOGDIR'))
            else:
                relfile = Path(prefs.get('LOGDIR')) / file
            if not relfile.exists():
                raise ParseError(f"Could not find input file either as an absolute path, or path relative to LOGDIR, got path {str(file)}")
            file = relfile

        with open(file, 'r') as lfile:
            lines = lfile.readlines()
        entries = [LogEntry.from_string(e, parse_messages) for e in lines]

        if include_backups:
            subfile = file.with_suffix(file.suffix+'.1')
            while subfile.exists():
                with open(subfile, 'r') as sfile:
                    lines = sfile.readlines()
                entries.extend([LogEntry.from_string(e, parse_messages) for e in lines])
                subfile = subfile.with_suffix('.' + str(int(subfile.suffix[1:])+1))

        return cls(entries=entries)



1	import os	1✔
2	import logging	1✔
3	import re	1✔
4	import multiprocessing as mp	1✔
5	import inspect	1✔
6	from pathlib import Path	1✔
7	from logging.handlers import RotatingFileHandler	1✔
8	from threading import Lock	1✔
9	import warnings	1✔
10	from rich.logging import RichHandler	1✔
11	from dataclasses import dataclass	1✔
12	from datetime import datetime	1✔
13	from parse import parse	1✔
14	import typing	1✔
15	from typing import Literal	1✔
16
17
18	from autopilot.root import Autopilot_Type	1✔
19
20	from autopilot import prefs	1✔
21
22	_LOGGERS = [] # type: list	1✔
23	"""
24	List of instantiated loggers, used in :func:`.init_logger` to return existing loggers without modification
25	"""
26
27	_INIT_LOCK = Lock() # type: Lock	1✔
28
29	LOGLEVELS = Literal['DEBUG', 'INFO', 'WARNING', 'ERROR']	1✔
30
31	def init_logger(instance=None, module_name=None, class_name=None, object_name=None) -> logging.Logger:	1✔
32	"""
33	Initialize a logger
34
35	Loggers are created such that...
36
37	* There is one logger per module (eg. all gpio objects will log to hardware.gpio)
38	* If the passed object has a ``name`` attribute, that name will be prefixed to its log messages in the file
39	* The loglevel for the file handler and the stdout is determined by ``prefs.get('LOGLEVEL')``, and if none is provided ``WARNING`` is used by default
40	* logs are rotated according to ``prefs.get('LOGSIZE')`` (in bytes) and ``prefs.get('LOGNUM')`` (number of backups of ``prefs.get('LOGSIZE')`` to cycle through)
41
42	Logs are stored in ``prefs.get('LOGDIR')``, and are formatted like::
43
44	"%(asctime)s - %(name)s - %(levelname)s : %(message)s"
45
46	Loggers can be initialized either by passing an object to the first ``instance`` argument, or
47	by specifying any of ``module_name`` , ``class_name`` , or ``object_name`` (at least one must be specified)
48	which are combined with periods like ``module.class_name.object_name``
49
50	Args:
51	instance: The object that we are creating a logger for! if None, at least one of ``module, class_name, or object_name`` must be passed
52	module_name (None, str): If no ``instance`` passed, the module name to create a logger for
53	class_name (None, str): If no ``instance`` passed, the class name to create a logger for
54	object_name (None, str): If no ``instance`` passed, the object name/id to create a logger for
55
56	Returns:
57	:class:`logging.logger`
58	"""
59
60	# --------------------------------------------------
61	# gather variables
62	# --------------------------------------------------
63
64	if instance is not None:	1✔
65	# get name of module_name without prefixed autopilot
66	# eg passed autopilot.hardware.gpio.Digital_In -> hardware.gpio
67	# filtering leading 'autopilot' from string
68
69	module_name = instance.__module__	1✔
70	if "__main__" in module_name:	1✔
71	# awkward workaround to get module name of __main__ run objects
72	mod_obj = inspect.getmodule(instance)	×
73	try:	×
74	mod_suffix = inspect.getmodulename(inspect.getmodule(instance).__file__)	×
75	module_name = '.'.join([mod_obj.__package__, mod_suffix])	×
76	except AttributeError:	×
77	# when running interactively or from a plugin, __main__ does not have __file__
78	module_name = "__main__"	×
79
80
81	module_name = re.sub('^autopilot.', '', module_name)	1✔
82
83	class_name = instance.__class__.__name__	1✔
84
85	# if object is running in separate process, give it its own file
86	if issubclass(instance.__class__, mp.Process):	1✔
87	# it should be at least 2 (in case its first spawned in its module)
88	# but otherwise nocollide
89	p_num = 2	1✔
90	_module_name = module_name	1✔
91	module_name = f"{_module_name}_{str(p_num).zfill(2)}"	1✔
92	if module_name in globals()['_LOGGERS']:	1✔
93	for existing_mod in globals()['_LOGGERS']:	×
94	if module_name in existing_mod and re.match(r'\d$', existing_mod):	×
95	p_num += 1	×
96
97	module_name = f"{_module_name}_{str(p_num).zfill(2)}"	×
98
99	# get name of object if it has one
100	if hasattr(instance, 'id'):	1✔
101	object_name = str(instance.id)	1✔
102	elif hasattr(instance, 'name'):	1✔
103	object_name = str(instance.name)	1✔
104	else:
105	object_name = None	1✔
106
107	# --------------------------------------------------
108	# check if logger needs to be made, or exists already
109	# --------------------------------------------------
110	elif not any((module_name, class_name, object_name)):	1✔
111	raise ValueError('Need to either give an object to create a logger for, or one of module_name, class_name, or object_name')	×
112
113
114	# get name of logger to get
115	logger_name_pieces = [v for v in (module_name, class_name, object_name) if v is not None]	1✔
116	logger_name = '.'.join(logger_name_pieces)	1✔
117
118	# trim __ from logger names, linux don't like to make things like that
119	# re.sub(r"^\_\_")
120
121	# --------------------------------------------------
122	# if new logger must be made, make it, otherwise just return existing logger
123	# --------------------------------------------------
124
125	# use a lock to prevent loggers from being double-created, just to be extra careful
126	with globals()['_INIT_LOCK']:	1✔
127
128	# check if something starting with module_name already exists in loggers
129	MAKE_NEW = False	1✔
130	if not any([test_logger == module_name for test_logger in globals()['_LOGGERS']]):	1✔
131	MAKE_NEW = True	1✔
132
133	if MAKE_NEW:	1✔
134	parent_logger = logging.getLogger(module_name)	1✔
135	loglevel = getattr(logging, prefs.get('LOGLEVEL'))	1✔
136	parent_logger.setLevel(loglevel)	1✔
137
138	# make formatter that includes name
139	log_formatter = logging.Formatter("[%(asctime)s] %(levelname)s [%(name)s]: %(message)s")	1✔
140
141	## file handler
142	# base filename is the module_name + '.log
143	base_filename = Path(prefs.get('LOGDIR')) / (module_name + '.log')	1✔
144
145	fh = _file_handler(base_filename)	1✔
146	fh.setLevel(loglevel)	1✔
147	fh.setFormatter(log_formatter)	1✔
148	parent_logger.addHandler(fh)	1✔
149
150	# rich logging handler for stdout
151	parent_logger.addHandler(_rich_handler())	1✔
152
153	# if our parent is the rootlogger, disable propagation to avoid printing to stdout
154	if isinstance(parent_logger.parent, logging.RootLogger):	1✔
155	parent_logger.propagate = False	1✔
156
157	## log creation
158	globals()['_LOGGERS'].append(module_name)	1✔
159	parent_logger.debug(f'parent, module-level logger created: {module_name}')	1✔
160
161	logger = logging.getLogger(logger_name)	1✔
162	if logger_name not in globals()['_LOGGERS']:	1✔
163	# logger.addHandler(_rich_handler())
164	logger.debug(f"Logger created: {logger_name}")	1✔
165	globals()['_LOGGERS'].append(logger_name)	1✔
166
167	return logger	1✔
168
169
170	def _rich_handler() -> RichHandler:	1✔
171	rich_handler = RichHandler(rich_tracebacks=True, markup=True)	1✔
172	rich_formatter = logging.Formatter(	1✔
173	"[bold green]\[%(name)s][/bold green] %(message)s",
174	datefmt='[%y-%m-%dT%H:%M:%S]'
175	)
176	rich_handler.setFormatter(rich_formatter)	1✔
177	return rich_handler	1✔
178
179	def _file_handler(base_filename: Path) -> RotatingFileHandler:	1✔
180	# if directory doesn't exist, try to make it
181	if not base_filename.parent.exists():	1✔
182	base_filename.parent.mkdir(parents=True, exist_ok=True)	×
183
184	try:	1✔
185	fh = RotatingFileHandler(	1✔
186	str(base_filename),
187	mode='a',
188	maxBytes=int(prefs.get('LOGSIZE')),
189	backupCount=int(prefs.get('LOGNUM'))
190	)
191	except PermissionError as e:	×
192	# catch permissions errors, try to chmod our way out of it
193	try:	×
194	for mod_file in Path(base_filename).parent.glob(f"{Path(base_filename).stem}*"):	×
195	os.chmod(mod_file, 0o777)	×
196	warnings.warn(f'Couldnt access {mod_file}, changed permissions to 0o777')	×
197
198	fh = RotatingFileHandler(	×
199	base_filename,
200	mode='a',
201	maxBytes=int(prefs.get('LOGSIZE')),
202	backupCount=int(prefs.get('LOGNUM'))
203	)
204	except Exception as f:	×
205	raise PermissionError(	×
206	f'Couldnt open logfile {base_filename}, and couldnt chmod our way out of it.\n' + '-' * 20 + f'\ngot errors:\n{e}\n\n{f}\n' + '-' * 20)
207
208	return fh	1✔
209
210	# --------------------------------------------------
211	# Parsers and in-memory representation of logs
212	# --------------------------------------------------
213
214	class ParseError(RuntimeError):	1✔
215	"""
216	Error parsing a logfile
217	"""
218
219	def _convert_asc_timestamp(timestamp:str) -> datetime:	1✔
220	hunk, ms = timestamp.split(',')	1✔
221	ms += '000'	1✔
222	timestamp = '.'.join([hunk, ms])	1✔
223	return datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S.%f')	1✔
224
225	@dataclass	1✔
226	class Log_Format:	1✔
227	format: str	1✔
228	"""A format string parseable by ``parse``"""
229	example: str	1✔
230	"""An example string (that allows for testing)"""
231	conversions: typing.Optional[typing.Dict[str, typing.Callable]] = None	1✔
232	"""A dictionary matching keys in the ``format`` string to callables for post-parsing coercion"""
233
234	def parse(self, log_entry:str) -> dict:	1✔
235	if self.conversions is None:	1✔
236	conversions = {}	×
237	else:
238	conversions = self.conversions	1✔
239
240	res = parse(self.format, log_entry, conversions)	1✔
241	if res is None:	1✔
242	raise ParseError(f'Could not parse {log_entry} with {self.format}')	1✔
243	else:
244	return res.named	1✔
245
246
247
248	LOG_FORMATS = (	1✔
249	Log_Format(
250	format = '{timestamp:Timestamp} - {name} - {level} : {message}',
251	conversions = {
252	'Timestamp': _convert_asc_timestamp
253	},
254	example = "2022-03-07 16:56:48,954 - networking.node.Net_Node._T - DEBUG : RECEIVED: ID: _testpi_9879; TO: T; SENDER: _testpi; KEY: DATA; FLAGS: {'NOREPEAT': True}; VALUE: {'trial_num': 1197, 'timestamp': '2022-03-01T23:52:16.995387', 'frequency': 45255.0, 'amplitude': 0.1, 'ramp': 5.0, 'pilot': 'testpi', 'subject': '0895'}"
255	),
256	Log_Format(
257	format = '[{timestamp:Timestamp}] {level} [{name}]: {message}',
258	conversions = {
259	'Timestamp': _convert_asc_timestamp
260	},
261	example = '[2022-03-09 16:13:43,224] INFO [networking.node]: parent, module-level logger created: networking.node'
262	)
263	)
264	"""
265	Possible formats of logging messages (to allow change over versions) as a `parse string <https://github.com/r1chardj0n3s/parse>`_
266	"""
267
268	MESSAGE_FORMATS = {	1✔
269	'node_msg_recv': '{action}: ID: {message_id}; TO: {to}; SENDER: {sender}; KEY: {key}; FLAGS: {flags}; VALUE: {value}',
270	'node_msg_sent': '{action} - ID: {message_id}; TO: {to}; SENDER: {sender}; KEY: {key}; FLAGS: {flags}; VALUE: {value}'
271	}
272	"""
273	Additional parsing patterns for logged messages
274
275	* ``node_msg``: Logging messages from :class:`.networking.node.Net_Node`
276
277	"""
278
279
280
281
282	class LogEntry(Autopilot_Type):	1✔
283	"""
284	Single entry in a log
285	"""
286	timestamp: datetime	1✔
287	name: str	1✔
288	level: LOGLEVELS	1✔
289	message: typing.Union[str, dict]	1✔
290
291	def parse_message(self, format:typing.List[str]):	1✔
292	"""
293	Parse the message using a format string specified as a key in the :data:`.MESSAGE_FORMATS` dictionary (or a format string itself)
294
295	replaces the :attr:`.message` attribute.
296
297	If parsing unsuccessful, no exception is raised because there are often messages that are not parseable in the logs!
298
299	Args:
300	format (typing.List[str]): List of format strings to try!
301
302	Returns:
303
304	"""
305	if isinstance(format, str):	×
306	format = [format]	×
307
308	format = [MESSAGE_FORMATS[f]if f in MESSAGE_FORMATS.keys() else f for f in format]	×
309
310	for f in format:	×
311	result = parse(f, self.message)	×
312	if result is not None:	×
313	self.message = result.named	×
314	return	×
315
316
317	@classmethod	1✔
318	def from_string(cls,	1✔
319	entry:str,
320	parse_message:typing.Optional[typing.List[str]]=None) -> 'LogEntry':
321	"""
322	Create a LogEntry by parsing a string.
323
324	Try to parse using any of the possible :ref:`.LOG_FORMATS`, raising a
325	:class:`.ParseError` if none are successful
326
327	Args:
328	entry (str): single line of a logging file
329	parse_message (Optional[str]): Parse messages with the :data:`.MESSAGE_FORMATS` key or format string
330
331	Returns:
332	:class:`.LogEntry`
333
334	Raises:
335	:class:`.ParseError` if no messages are parsed
336
337	"""
338	for aformat in LOG_FORMATS:	1✔
339	try:	1✔
340	result = aformat.parse(entry)	1✔
341	except ParseError:	1✔
342	# fine, we're searching for one that works
343	continue	1✔
344
345	entry = cls(**result)	1✔
346	if parse_message is not None:	1✔
347	entry.parse_message(parse_message)	×
348	return entry	1✔
349
350	# if we haven't returned anything, raise a parse error
351	raise ParseError(f'Couldnt parse entry with any known log formats: {entry}')	×
352
353	class Log(Autopilot_Type):	1✔
354	"""
355	Representation of a logfile in memory
356	"""
357	entries: typing.List[LogEntry]	1✔
358
359	@classmethod	1✔
360	def from_logfile(cls,	1✔
361	file: typing.Union[Path, str],
362	include_backups:bool = True,
363	parse_messages: typing.Optional[typing.List[str]]=None ):
364	"""
365	Load a logfile (and maybe its backups) from a logfile location
366
367	Args:
368	file (:class:`pathlib.Path`, str): If string, converted to Path. If relative (and relative file is not found),
369	then attempts to find relative to ``prefs.LOGDIR``
370	include_backups (bool): if ``True`` (default), try and load all of the backup logfiles (that have .1, .2, etc appended)
371	parse_messages (Optional[str]): Parse messages with the :data:`.MESSAGE_FORMATS` key or format string
372
373	Returns:
374	:class:`.Log`
375	"""
376	file = Path(file)	1✔
377	if not file.exists():	1✔
378	if file.is_absolute():	×
379	relfile = file.relative_to(prefs.get('LOGDIR'))	×
380	else:
381	relfile = Path(prefs.get('LOGDIR')) / file	×
382	if not relfile.exists():	×
383	raise ParseError(f"Could not find input file either as an absolute path, or path relative to LOGDIR, got path {str(file)}")	×
384	file = relfile	×
385
386	with open(file, 'r') as lfile:	1✔
387	lines = lfile.readlines()	1✔
388	entries = [LogEntry.from_string(e, parse_messages) for e in lines]	1✔
389
390	if include_backups:	1✔
391	subfile = file.with_suffix(file.suffix+'.1')	1✔
392	while subfile.exists():	1✔
393	with open(subfile, 'r') as sfile:	1✔
394	lines = sfile.readlines()	1✔
395	entries.extend([LogEntry.from_string(e, parse_messages) for e in lines])	1✔
396	subfile = subfile.with_suffix('.' + str(int(subfile.suffix[1:])+1))	1✔
397
398	return cls(entries=entries)	1✔
399
400

auto-pi-lot / autopilot / 745

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous