Source code for labgrid.consoleloggingreporter

import os
import sys
from datetime import datetime

from .step import steps

[docs]class ConsoleLoggingReporter: """ConsoleLoggingReporter - Reporter that writes console log files Args: logpath (str): path to store the logfiles in """ instance = None
[docs] @classmethod def start(cls, path): """starts the ConsoleLoggingReporter""" assert cls.instance is None cls.instance = cls(path)
[docs] @classmethod def stop(cls): """stops the ConsoleLoggingReporter""" assert cls.instance is not None cls.instance._stop() steps.unsubscribe(cls.instance.notify) cls.instance = None
[docs] def __init__(self, logpath): self._logcache = {} self.logpath = logpath steps.subscribe(self.notify)
def _stop(self): while self._logcache: _, log = self._logcache.popitem() # ignore cache entries for errors if log is None: continue log.close()
[docs] def get_logfile(self, event): """Returns the correct file handle from cache or creates a new file handle""" source = event.step.source try: log = self._logcache[source] except KeyError: if name = 'console_{}_{}'.format(, else: name = 'console_{}'.format( name = os.path.join(self.logpath, name) try: log = self._logcache[source] = open(name, mode='ab', buffering=0) except PermissionError: print("failed to open log file {}".format(name), file=sys.stderr) log = self._logcache[source] = None if not log: return None if log.write( "Labgrid Console Logfile for {} {}\n". format(,"utf-8") ) else: log.write( "Labgrid Console Logfile for {}\n".format( .encode("utf-8") ) log.write( "Logfile started at {}\n".format("utf-8") ) log.write("=== Log starts here ===\n".encode('utf-8')) return log
[docs] def notify(self, event): """This is the callback function for steps""" step = event.step if step.tag == 'console': if str(step) == 'read': if'state') == 'stop': if step.result and step.source: log = self.get_logfile(event) if not log: return log.write(step.result)