#188 Improve Activity History (7)

Move activity history code from utils to interfaces and types

packages/core/interfaces/__init__.py

@@ -1,3 +1,4 @@
-from .state_interface import StateInterface
-from .plc_interface import PLCInterface
+from .activity_history import ActivityHistoryInterface
 from .os_interface import OSInterface
+from .plc_interface import PLCInterface
+from .state_interface import StateInterface

packages/core/utils/activity_history.py → packages/core/interfaces/activity_history.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
+from typing import Optional
 import datetime
 import json
 import os
-from typing import Optional
 import pydantic
+from packages.core import types
 
 _dir = os.path.dirname
 _PROJECT_DIR = _dir(_dir(_dir(_dir(os.path.abspath(__file__)))))
@@ -14,60 +15,33 @@
 # TODO: add cli calls to activity history
 
 
-class ActivityDatapoint(pydantic.BaseModel):
-    """A datapoint of the activity history."""
-
-    local_time: datetime.time
-    is_measuring: bool = False
-    has_errors: bool = False
-    is_uploading: bool = False
-    camtracker_startups: int = 0
-    opus_startups: int = 0
-    cli_calls: int = 0
-
-
-class ActivityDatapointList(pydantic.RootModel[list[ActivityDatapoint]]):
-    """A datapoint of the activity history."""
-
-    root: list[ActivityDatapoint]
-
-
-example = {
-    "localTime": "12:00", "measuring": True, "errors": False, "uploading": False
-}
-
-
-class ActivityHistory(pydantic.BaseModel):
-    """A datapoint of the activity history."""
-
-    datapoints: ActivityDatapointList = ActivityDatapointList(root=[])
-    date: datetime.date
-
-
 def _date_to_filepath(date: datetime.date) -> str:
     return os.path.join(
         _PROJECT_DIR, "logs", "activity", f"activity-{date}.json"
     )
 
 
-def _load_current_activity_history() -> ActivityHistory:
+def _load_current_activity_history() -> types.ActivityHistory:
     today = datetime.date.today()
 
     try:
         with open(_date_to_filepath(today), "r") as f:
-            return ActivityHistory(datapoints=json.load(f), date=today)
+            return types.ActivityHistory(datapoints=json.load(f), date=today)
     except (
         json.JSONDecodeError,
         FileNotFoundError,
         pydantic.ValidationError,
     ):
-        return ActivityHistory(date=today)
+        return types.ActivityHistory(date=today)
 
 
 class ActivityHistoryInterface:
-    """Logging the system activity every minute to `logs/activity/activity-YYYY-MM-DD.json` to plot it in the UI."""
+    """Logging the system activity every minute to 
+    `logs/activity/activity-YYYY-MM-DD.json` to plot
+    it in the UI."""
 
-    current_activity_history: ActivityHistory = _load_current_activity_history()
+    current_activity_history: types.ActivityHistory = _load_current_activity_history(
+    )
     last_write_time: datetime.datetime = datetime.datetime(1970, 1, 1)
 
     @staticmethod
@@ -79,7 +53,9 @@ def add_datapoint(
         opus_startups: Optional[int] = None,
         cli_calls: Optional[int] = None,
     ) -> None:
-        """Add a new activity datapoint"""
+        """Add a new activity datapoint. When this function is called
+        multiple times in the same minute, the datapoints are aggregated
+        into one datapoint per minute."""
 
         current_local_datetime = datetime.datetime.now()
 
@@ -88,7 +64,7 @@ def add_datapoint(
             != current_local_datetime.date()
         ):
             ActivityHistoryInterface.dump_current_activity_history()
-            ActivityHistoryInterface.current_activity_history = ActivityHistory(
+            ActivityHistoryInterface.current_activity_history = types.ActivityHistory(
                 date=current_local_datetime.date()
             )
 
@@ -97,7 +73,7 @@ def add_datapoint(
 
         # determining if the last datapoint is from the same minute
         # if so, we update it, otherwise we create a new one
-        last_activity_datapoint: Optional[ActivityDatapoint] = None
+        last_activity_datapoint: Optional[types.ActivityDatapoint] = None
         if len(new_history.datapoints.root) > 0:
             last_activity_datapoint = new_history.datapoints.root[-1]
 
@@ -111,7 +87,7 @@ def add_datapoint(
         # creating a new datapoint if none exist for the current minute
         current_activity_datapoint = last_activity_datapoint
         if current_activity_datapoint is None:
-            current_activity_datapoint = ActivityDatapoint(
+            current_activity_datapoint = types.ActivityDatapoint(
                 local_time=datetime.time(
                     hour=current_local_datetime.hour,
                     minute=current_local_datetime.minute,

packages/core/main.py

@@ -3,7 +3,6 @@
 import time
 from typing import Any, Callable, Literal, Optional
 from packages.core import types, utils, interfaces, modules, threads
-from packages.core.utils.activity_history import ActivityHistoryInterface
 
 logger = utils.Logger(origin="main")
 
@@ -145,7 +144,7 @@ def run() -> None:
     # that the core is shutting down
     def _graceful_teardown(*args: Any) -> None:
         logger.info("Received shutdown signal, starting graceful teardown")
-        ActivityHistoryInterface.dump_current_activity_history()
+        interfaces.ActivityHistoryInterface.dump_current_activity_history()
         logger.info("Graceful teardown complete")
         exit(0)
 
@@ -158,7 +157,7 @@ def _graceful_teardown(*args: Any) -> None:
         start_time = time.time()
         logger.info("Starting iteration")
 
-        ActivityHistoryInterface.add_datapoint(
+        interfaces.ActivityHistoryInterface.add_datapoint(
             has_errors=len(current_exceptions) > 0,
         )
 
@@ -214,7 +213,7 @@ def _graceful_teardown(*args: Any) -> None:
         seconds_per_core_interval = config.general.seconds_per_core_interval
         if config.general.test_mode:
             seconds_per_core_interval = 10
-            ActivityHistoryInterface.dump_current_activity_history()
+            interfaces.ActivityHistoryInterface.dump_current_activity_history()
         time_to_wait = seconds_per_core_interval - elapsed_time
         if time_to_wait > 0:
             logger.debug(f"Waiting {round(time_to_wait, 2)} second(s)")

packages/core/modules/measurement_conditions.py

@@ -1,6 +1,5 @@
 import datetime
 from packages.core import types, utils, interfaces
-from packages.core.utils.activity_history import ActivityHistoryInterface
 
 logger = utils.Logger(origin="measurement-conditions")
 
@@ -79,7 +78,7 @@ def run(self, new_config: types.Config) -> None:
         logger.info(
             f"Measurements should be running is set to: {measurements_should_be_running}."
         )
-        ActivityHistoryInterface.add_datapoint(
+        interfaces.ActivityHistoryInterface.add_datapoint(
             is_measuring=measurements_should_be_running
         )
         interfaces.StateInterface.update_state(

packages/core/types/__init__.py

@@ -1,3 +1,4 @@
+from .activity_history import ActivityDatapoint, ActivityDatapointList, ActivityHistory
 from .config import Config, ConfigPartial, StrictIPAdress
 from .plc_specification import (
     PLCSpecification,

packages/core/types/activity_history.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+import datetime
+import pydantic
+
+
+class ActivityDatapoint(pydantic.BaseModel):
+    """A datapoint of the activity history."""
+
+    local_time: datetime.time
+    is_measuring: bool = False
+    has_errors: bool = False
+    is_uploading: bool = False
+    camtracker_startups: int = 0
+    opus_startups: int = 0
+    cli_calls: int = 0
+
+
+class ActivityDatapointList(pydantic.RootModel[list[ActivityDatapoint]]):
+    """A datapoint of the activity history."""
+
+    root: list[ActivityDatapoint]
+
+
+example = {
+    "localTime": "12:00", "measuring": True, "errors": False, "uploading": False
+}
+
+
+class ActivityHistory(pydantic.BaseModel):
+    """A datapoint of the activity history."""
+
+    datapoints: ActivityDatapointList = ActivityDatapointList(root=[])
+    date: datetime.date

packages/core/utils/__init__.py

@@ -1,6 +1,5 @@
-from .activity_history import ActivityHistoryInterface as ActivityHistoryInterface
-from .logger import Logger
 from .astronomy import Astronomy
 from .exception_email_client import ExceptionEmailClient
-from .helios_image_processing import HeliosImageProcessing
 from .functions import read_last_file_line
+from .helios_image_processing import HeliosImageProcessing
+from .logger import Logger