Source code for parsing.library.tracker

# Copyright (C) 2017 Technologies, LLC
# is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.

import datetime

from timeit import default_timer as timer

from parsing.library.exceptions import PipelineError

[docs]class TrackerError(PipelineError): """Tracker error class."""
[docs]class Tracker: """Tracks specified attributes and broadcasts to viewers. @property attributes are defined for all BROADCAST_TYPES """ BROADCAST_TYPES = { "SCHOOL", "YEAR", "TERM", "DEPARTMENT", "STATS", "INSTRUCTOR", "TIME", "MODE", } def __init__(self): """Initialize tracker object.""" self.viewers = [] self._create_tracking_properties() def _create_tracking_properties(self): """Create @property attributes for all BROADCAST_TYPES. This is equivalent to enumerating for each broadcast type:: @property def year(self): return self._year @year.setter def year(self, value): self._year = value self.broadcast('YEAR') Will not override an attribute if an @property is already defined within the class for a broadcast type. """ for btype in Tracker.BROADCAST_TYPES: name = btype.lower() storage_name = "_{}".format(name) # If attribute is already part of class, do not override it. if ( hasattr(self, storage_name) or hasattr(self.__class__, name) or hasattr(self, name) ): continue # NOTE: closure methods are used to capture the variables # in their current context of the loop. def closure_getter(name, storage_name): def getter(self): return getattr(self, storage_name) return getter def closure_setter(btype, name, storage_name): def setter(self, value): setattr(self, storage_name, value) self.broadcast(btype) return setter setattr( self.__class__, name, property( closure_getter(name, storage_name), closure_setter(btype, name, storage_name), ), )
[docs] def start(self): """Start timer of tracker object.""" self.timestamp = datetime.datetime.utcnow().strftime("%Y/%m/%d-%H:%M:%S") self.start_time = timer()
[docs] def end(self): """End tracker and report to viewers.""" self.end_time = timer()
[docs] def add_viewer(self, viewer, name=None): """Add viewer to broadcast queue. Args: viewer (Viewer): Viewer to add. name (None, str, optional): Name the viewer. """ if name is None: name = "viewer{}".format(len(self.viewers)) self.viewers.append((name, viewer))
[docs] def remove_viewer(self, name): """Remove all viewers that match name. Args: name (str): Viewer name to remove. """ self.viewers = [v for v in self.viewers if v[0] != name]
[docs] def has_viewer(self, name): """Determine if name exists in viewers. Args: name (str): The name to check against. Returns: bool: True if name in viewers else False """ return name in dict(self.viewers)
[docs] def get_viewer(self, name): """Get viewer by name. Will return arbitrary match if multiple viewers with same name exist. Args: name (str): Viewer name to get. Returns: Viewer: Viewer instance if found, else None """ return dict(self.viewers).get(name)
[docs] def broadcast(self, broadcast_type): """Broadcast tracker update to viewers. Args: broadcast_type (str): message to go along broadcast bus. Raises: TrackerError: if broadcast_type is not in BROADCAST_TYPE. """ if broadcast_type not in Tracker.BROADCAST_TYPES: raise TrackerError("unsupported broadcast type {}".format(broadcast_type)) # TODO - broadcast based on optional name argument for name, viewer in self.viewers: viewer.receive(self, broadcast_type)
[docs] def report(self): """Notify viewers that tracker has ended.""" for name, viewer in self.viewers:
[docs]class NullTracker(Tracker): """Dummy tracker used as an interface placeholder.""" def __init__(self, *args, **kwargs): """Construct null tracker.""" super(NullTracker, self).__init__(*args, **kwargs)
[docs] def broadcast(self, broadcast_type): """Do nothing."""
[docs] def report(self): """Do nothing."""