Index: leahi_dialin/utils/base.py =================================================================== diff -u -r41de945f9c773e54e965e80d9e46def828beb732 -r6d104d3185ac3ed7c18c97ecdc13fd59bf53a8d1 --- leahi_dialin/utils/base.py (.../base.py) (revision 41de945f9c773e54e965e80d9e46def828beb732) +++ leahi_dialin/utils/base.py (.../base.py) (revision 6d104d3185ac3ed7c18c97ecdc13fd59bf53a8d1) @@ -15,31 +15,17 @@ ############################################################################ import logging import os -from abc import ABC, abstractmethod from datetime import datetime -from enum import Enum from threading import Timer + INTERVAL_1s = 1 INTERVAL_5s = 5 INTERVAL_10s = 10 INTERVAL_60s = 60 -class AbstractObserver(ABC): - """ - Publicly accessible parent class for all observers. - The update method will receive data when data is made available - """ - @abstractmethod - def update(self): - """ - Attach an observer - """ - pass - - class _FauxLogger: def __init__(self, printing_enabled=False): @@ -206,31 +192,6 @@ return "{0}{1}{2}".format(path, i, ext) -class AbstractSubSystem: - - @abstractmethod - def __init__(self): - """ - Initialization function for the sub system - # The abstract base class requires all abstract methods are overridden by children classes - """ - self._observers = [] - self._datetime_fmt = "%m.%d.%Y_%I.%M.%S.%f" - pass - - def attach(self, observer: AbstractObserver): - """ - Attach an observer so it is updated upon published events - """ - self._observers.append(observer) - - def detach(self, observer: AbstractObserver): - """ - Detach an observer - """ - self._observers.remove(observer) - - def publish(keys): """ Decorator that accepts a list of variable names to publish @@ -267,50 +228,6 @@ return _decorator -class DialinEnum(Enum): - - @classmethod - def has_value(cls, value): - return value in cls._value2member_map_ - - @classmethod - def from_str(cls, label: str): - for enum_member in cls.__members__.values(): - # If the string matching with the enum's name - if label.lower() == enum_member.name.lower(): - return enum_member - # Replace _ with ' ' and check again - elif label.lower().replace('_', ' ') == enum_member.name.lower().replace('_', ' '): - return enum_member - # If the string matching with the enum's state without the 'mode_' text - elif label.lower() == enum_member.name.lower().replace('mode_', ''): - return enum_member - # If it's in the string list provided for the enum - elif enum_member.name in cls._str_list and label.lower() in cls._str_list[enum_member.name]: - return enum_member - # Replace _ with ' ' and check the list again - elif enum_member.name in cls._str_list and label.lower().replace('_', ' ') in cls._str_list[enum_member.name]: - return enum_member - # Replace _ with ' ' and check the list again with _ with ' ' - elif enum_member.name in cls._str_list and label.lower().replace('_', ' ') in [enum_mem.lower().replace('_', ' ') for enum_mem in cls._str_list[enum_member.name] ]: - return enum_member - # If the enum is the NUM_ collector, then return stop as after that only aliases are present - elif enum_member.name.lower().startswith('num_'): - return None - return None - - -class AlarmEnum(Enum): - def __init__(self, *args): - cls = self.__class__ - if any(self.value == member.value for member in cls): - raise ValueError("aliases not allowed: %r --> %r" % (self.name, cls(self.value).name)) - - @classmethod - def has_value(cls, value): - return value in cls._value2member_map_ - - class IntervalTimer(object): """ A class object that is used to execute a function on a timed interval. @@ -341,16 +258,3 @@ def stop(self): self._timer.cancel() self.is_running = False - - -class InternalEvent: - def __init__(self): - self._listeners = [] - - def add_listener(self, func): - self._listeners.append(func) - return func - - def trigger(self, *args, **kwargs): - for func in self._listeners: - func(*args, **kwargs)