Merge D5 work

.gitignore

@@ -1,2 +1,16 @@
-astra.db
+# databases
+*.db
+
+# pycache
 __pycache__
+
+# mypy
+.mypy_cache
+
+# pytest
+.pytest_cache
+
+# pyinstaller
+*.spec
+dist/
+build/

requirements.txt

@@ -2,5 +2,5 @@ PyYAML~=6.0.1
 SQLAlchemy~=2.0.23
 h5py~=3.10.0
 pandas~=2.1.3
-pytest
-matplotlib~=3.8.2
+pytest~=7.4.3
+matplotlib~=3.8.2

src/astra/data/alarm_container.py

@@ -1,22 +1,26 @@
+"""This module provides classes for storing and working with alarms."""
+
 from datetime import timedelta
 from queue import Queue
 from threading import Lock, Timer
 from typing import Callable, Mapping
 
 from astra.data.alarms import AlarmPriority, Alarm, AlarmCriticality
 
-NEW_QUEUE_KEY = 'n'
-
 
 class AlarmObserver:
     """
     Observes the state of the global alarms container and notifies interested
     parties whenever an update occurs
 
-    :param watchers: A list of functions to call on any update to the alarm container
-    :param mutex: Synchronization tool as many threads may notify watchers of updates
+    :param: watchers: A list of functions to call on any update to the alarm container
+    :type: list[Callable]
+
+    :param: mutex: Synchronization tool as many threads may notify watchers of updates
+    :type: Lock
     """
-    watchers = []
+
+    watchers: list[Callable] = []
     _mutex = Lock()
 
     @classmethod
@@ -43,17 +47,28 @@ class AlarmsContainer:
     A container for a global alarms dict that utilizes locking for multithreading
 
     :param alarms: The actual dictionary of alarms held
+    :type: dict[str, list[Alarm]]
+
     :param mutex: A lock used for mutating cls.alarms
+    :type: Lock
+
     :param observer: An Observer to monitor the state of the container
+    :type: AlarmObserver
     """
+
     observer = AlarmObserver()
-    alarms = {AlarmPriority.WARNING.name: [], AlarmPriority.LOW.name: [],
-              AlarmPriority.MEDIUM.name: [], AlarmPriority.HIGH.name: [],
-              AlarmPriority.CRITICAL.name: [], NEW_QUEUE_KEY: Queue()}
+    alarms: dict[str, list[Alarm]] = {
+        AlarmPriority.WARNING.name: [],
+        AlarmPriority.LOW.name: [],
+        AlarmPriority.MEDIUM.name: [],
+        AlarmPriority.HIGH.name: [],
+        AlarmPriority.CRITICAL.name: [],
+    }
+    new_alarms: Queue[Alarm] = Queue()
     mutex = Lock()
 
     @classmethod
-    def get_alarms(cls) -> dict[str, list[Alarm] | Queue]:
+    def get_alarms(cls) -> dict[str, list[Alarm]]:
         """
         Returns a shallow copy of <cls.alarms>
 
@@ -63,16 +78,16 @@ def get_alarms(cls) -> dict[str, list[Alarm] | Queue]:
             return cls.alarms.copy()
 
     @classmethod
-    def add_alarms(cls, alarms: list[Alarm],
-                   apm: Mapping[timedelta, Mapping[AlarmCriticality, AlarmPriority]]) -> None:
+    def add_alarms(
+        cls, alarms: list[Alarm], apm: Mapping[timedelta, Mapping[AlarmCriticality, AlarmPriority]]
+    ) -> None:
         """
         Updates the alarms global variable after acquiring the lock for it
 
-        :param dm: Holds information of data criticality and priority
         :param apm: Maps information on alarms to correct priority level
         :param alarms: The set of alarms to add to <cls.alarms>
         """
-        new_alarms = []
+        new_alarms: list[tuple[Alarm, AlarmCriticality]] = []
         times = [0, 5, 15, 30]
         timer_vals = []
         alarms.sort(reverse=True)
@@ -82,8 +97,8 @@ def add_alarms(cls, alarms: list[Alarm],
             with cls.mutex:
                 for alarm in alarms:
                     criticality = alarm.criticality
-                    alarm_timer_vals = []
-                    cls.alarms[NEW_QUEUE_KEY].put(alarm)
+                    alarm_timer_vals: list[timedelta] = []
+                    cls.new_alarms.put(alarm)
 
                     # Find the closest timeframe from 0, 5, 15, and 30 minutes from when the
                     # alarm was created to when it was actually confirmed
@@ -100,9 +115,9 @@ def add_alarms(cls, alarms: list[Alarm],
                         if alarm.event.creation_time < endpoint_time and not alarm_timer_vals:
                             priority_name = apm[timedelta(minutes=times[i - 1])][criticality]
                             priority = AlarmPriority(priority_name)
-                            cls.alarms[priority].append(alarm)
+                            cls.alarms[priority.name].append(alarm)
                             alarm.priority = priority
-                            new_alarms.append([alarm, priority])
+                            new_alarms.append((alarm, priority))
 
                             remaining_time = endpoint_time - alarm.event.creation_time
                             alarm_timer_vals.append(remaining_time)
@@ -112,46 +127,55 @@ def add_alarms(cls, alarms: list[Alarm],
                         priority = apm[timedelta(minutes=30)][criticality]
                         alarm.priority = priority
                         cls.alarms[priority.name].append(alarm)
-                        new_alarms.append([alarm, priority])
+                        new_alarms.append((alarm, priority))
                     timer_vals.append(alarm_timer_vals)
 
             # Now that the state of the alarms container has been update, notify watchers
             # cls.observer.notify_watchers()
 
             # Now, we need to create a timer thread for each alarm
             for i in range(len(new_alarms)):
-                alarm = new_alarms[i]
+                alarm_data: Alarm = new_alarms[i][0]
+                alarm_crit: AlarmCriticality = new_alarms[i][1]
+
                 associated_times = timer_vals[i]
 
                 for associated_time in associated_times:
                     # Because <associated_times> is a subset of <times>, we need to offset
                     # the index well later use into it
                     time_interval = len(times) - len(associated_times) + i
 
-                    new_timer = Timer(associated_time.seconds, cls._update_priority,
-                                      args=[alarm, timedelta(minutes=times[time_interval]), apm])
+                    new_timer = Timer(
+                        associated_time.seconds,
+                        cls._update_priority,
+                        args=[alarm_data, alarm_crit, timedelta(minutes=times[time_interval]), apm],
+                    )
                     new_timer.start()
 
     @classmethod
-    def _update_priority(cls, alarm_data: list[Alarm, AlarmPriority], time: timedelta,
-                         apm: Mapping[timedelta, Mapping[AlarmCriticality, AlarmPriority]]) -> None:
+    def _update_priority(
+        cls,
+        alarm: Alarm,
+        alarm_crit: AlarmCriticality,
+        time: timedelta,
+        apm: Mapping[timedelta, Mapping[AlarmCriticality, AlarmPriority]],
+    ) -> None:
         """
         Uses the alarm priority matrix with <time> to place <alarm> in the correct priority bin
         NOTE: we pass a list, and not a tuple, as we need to mutate said list
 
-        :param alarm_data: contains the current priority of the alarm and the alarm itself
+        :param alarm: The alarm to update priority of
+        :param alarm_crit: The base criticality of the alarm
         :param time: Time elapsed since the alarm came into effect
         :param apm: Maps information on alarms to correct priority level
         """
 
         with cls.mutex:
-            new_priority = apm[time][alarm_data[0].criticality]
-            cls.alarms[alarm_data[1]].remove(alarm_data[0])
-            cls.alarms[new_priority].append(alarm_data[0])
-            alarm_data[0].priority = new_priority
-            alarm_data[1] = new_priority
-        # Because of the nature of priorities, we need to notify both modification observers
-        # and addition observers
+            new_priority = apm[time][alarm.criticality]
+
+            cls.alarms[alarm_crit.name].remove(alarm)
+            cls.alarms[alarm_crit.name].append(alarm)
+            alarm.priority = new_priority
         cls.observer.notify_watchers()
 
     @classmethod
@@ -179,7 +203,7 @@ def remove_alarm(cls, alarm: Alarm) -> None:
         """
         with cls.mutex:
             for priority in cls.alarms:
-                if priority != NEW_QUEUE_KEY and alarm in cls.alarms[priority]:
+                if alarm in cls.alarms[priority]:
                     # Note: We don't consider the queue of new alarms, since by the time
                     # the alarm can be removed, it's already be taken out of the queue
                     cls.alarms[priority].remove(alarm)

src/astra/data/alarms.py

@@ -1,3 +1,9 @@
+"""
+This module defines classes and types relating to events and alarms.
+
+The classes defined here generally work the same way as they do in Panoptes.
+"""
+
 import functools
 from abc import ABC, abstractmethod
 from collections.abc import Iterable
@@ -12,18 +18,24 @@
 
 @dataclass(frozen=True)
 class EventBase(ABC):
-    type: ClassVar[str] = ''
+    type: ClassVar[str] = ''  # Used for display purposes
     persistence: float | None
     description: str
 
     @property
     @abstractmethod
     def tags(self) -> Iterable[Tag]:
+        """A unified way to access all tags associated with an event base, regardless of type."""
         raise NotImplementedError
 
 
 @dataclass(frozen=True)
 class SimpleEventBase(EventBase):
+    """
+    An event base revolving around a single tag.
+
+    Equivalent to SingleTagEventBase in Panoptes.
+    """
     tag: Tag
 
     @override
@@ -34,6 +46,13 @@ def tags(self) -> Iterable[Tag]:
 
 @dataclass(frozen=True)
 class CompoundEventBase(EventBase):
+    """
+    An event base that consists of several inner event bases.
+
+    Equivalent to MultiTagEventBase in Panoptes.
+
+    Tags are currently automatically determined based on the tags of the inner event bases.
+    """
     event_bases: list[EventBase]
 
     @override
@@ -75,12 +94,12 @@ class SOEEventBase(CompoundEventBase):
 
 
 @dataclass(frozen=True)
-class AllEventBase(CompoundEventBase):
+class AllEventBase(CompoundEventBase):  # LogicalAndEventBase in Panoptes
     type: ClassVar[str] = 'Logical AND'
 
 
 @dataclass(frozen=True)
-class AnyEventBase(CompoundEventBase):
+class AnyEventBase(CompoundEventBase):  # LogicalOrEventBase in Panoptes
     type: ClassVar[str] = 'Logical OR'
 
 
@@ -98,10 +117,11 @@ class Event:
 
     @property
     def type(self) -> str:
+        """A string representation of the general type (static, setpoint, etc) of this event."""
         return type(self.base).type
 
 
-@functools.total_ordering
+@functools.total_ordering  # Criticalities/priorities can be compared in the natural way
 class AlarmCriticality(Enum):
     WARNING = 'WARNING'
     LOW = 'LOW'
@@ -141,6 +161,7 @@ class AlarmBase:
 @dataclass(frozen=False)
 class Alarm:
     event: Event
+    # As of now, no separate ID for alarms -- the id of the underlying event is used instead.
     criticality: AlarmCriticality
     priority: AlarmPriority
     acknowledged: bool

src/astra/data/config_manager.py

@@ -1,3 +1,5 @@
+"""This module provides functionality for reading config files."""
+
 from yaml import safe_load
 
 # import parameters
@@ -11,12 +13,9 @@
 
 def _read_config_yaml(filename: str) -> None:
     """
-    Read in a yaml configuration file and parse it into a pandas dataframe.
-    The data in the "alarm" column is temporarily removed to
-    improve the readability of the dataframe.
+    Read in a yaml configuration file and save it to the database.
 
-    Args:
-        filename (str): full path to the config file
+    :param filename: full path to the config file
     """
     with open(filename, "r") as yaml_file:
         config_contents = safe_load(yaml_file)
@@ -34,24 +33,19 @@ def _read_config_yaml(filename: str) -> None:
             )
 
         # create or update the alarms in database
-        dictionary_of_alarms = config_contents["alarms"]
-        create_update_alarm(dictionary_of_alarms, device_id=device)
+        alarm_dicts = config_contents["alarms"]
+        create_update_alarm(alarm_dicts, device_id=device)
 
 
-def yaml_tag_parser(tag_dict: dict) -> dict:
+def yaml_tag_parser(tag_dict: dict) -> Parameter:
     """
-        A helper function for yaml reader that parse the tag
-        data into a dictionary.
+    Helper function for yaml reader: parse the tag data into a Parameter object.
 
-    Args:
-        tag_data (dict): a dictionary that contains the tag data
-                         (setpoint, dtype, display_units)
+    :param tag_dict: a dictionary that contains the tag data (setpoint, dtype, display_units)
 
-    Raises:
-        ValueError: when the dtype is not int, float, or bool
+    :raise ValueError: when the dtype is not int, float, or bool
 
-    Returns:
-        dict: a dictionary that contains the parsed tag data
+    :return: a Parameter object from the parsed tag data
     """
     tag_id = list(tag_dict.keys())[0]
     tag_data = tag_dict[tag_id]
@@ -60,6 +54,8 @@ def yaml_tag_parser(tag_dict: dict) -> dict:
     else:
         setpoint = None
 
+    dtype: type[int] | type[float] | type[bool]
+
     if tag_data["dtype"] == "int":
         dtype = int
     elif tag_data["dtype"] == "float":
@@ -88,7 +84,7 @@ def yaml_tag_parser(tag_dict: dict) -> dict:
     )
 
 
-# A dictionary that map file extensions to reader functions
+# A dictionary that maps file extensions to reader functions
 _file_readers = {
     "yaml": _read_config_yaml
     # Add more file types and reader functions as needed
@@ -97,12 +93,11 @@ def yaml_tag_parser(tag_dict: dict) -> dict:
 
 def read_config(filename: str) -> None:
     """
-    Read in a configuration file with the given path and return
-    it as a pandas dataframe.
-    Raise ValueError when the type of the configuration is incorrect.
+    Read in a configuration file with the given path and save it to the database.
+
+    :param filename: full path to the config file
 
-    Args:
-        filename (str): full path to the config file
+    :raise ValueError: when the type of the configuration is incorrect.
     """
     file_extension = filename.split(".")[-1]
     reader_func = _file_readers.get(file_extension)