DO NOT MERGE

pyproject.toml

@@ -12,7 +12,11 @@ classifiers = [
     "Programming Language :: Python :: 3.13",
 ]
 description = "Reads bacnet data to an IOC"
-dependencies = [] # Add project dependencies here, e.g. ["click", "numpy"]
+dependencies = [
+    "BAC0",
+    "fastcs[epicsca]",
+    "epicscorelibs",
+] # Add project dependencies here, e.g. ["click", "numpy"]
 dynamic = ["version"]
 license.file = "LICENSE"
 readme = "README.md"
@@ -21,6 +25,7 @@ requires-python = ">=3.11"
 [dependency-groups]
 dev = [
     "copier",
+    "matplotlib",
     "pre-commit",
     "pyright",
     "pytest",
@@ -45,7 +50,7 @@ version_file = "src/fastcs_bacnet/_version.py"
 
 [tool.pyright]
 typeCheckingMode = "standard"
-reportMissingImports = false # Ignore missing stubs in imported modules
+reportMissingImports = false  # Ignore missing stubs in imported modules
 
 [tool.pytest.ini_options]
 # Run pytest with all our checkers, and don't spam us with massive tracebacks on error

src/fastcs_bacnet/diagnostics/BAC0/response_timer.py

@@ -0,0 +1,101 @@
+from datetime import timedelta
+
+from fastcs_bacnet.diagnostics.BAC0.subscription_pair import SubscriptionPair
+from fastcs_bacnet.dummy.BAC0.analog_output_object import AnalogOutputObject
+from fastcs_bacnet.practical.BAC0.object_subscription import ObjectSubscription
+
+
+class ResponseTimer:
+    """
+    Class for managing a list of SubscriptionPair s
+    """
+
+    # May want to change to a dictionary??
+    # Most of the operations invlolve iterating through it
+    # But getting a specific value using keys is also used
+    _subscription_pairs: list[SubscriptionPair] = []
+
+    def __init__(self, recent_times_buffer_length: int = 20):
+        self._recent_times_buffer_length = recent_times_buffer_length
+
+    def add_subscription_pair(
+        self,
+        analog_output_object: AnalogOutputObject,
+        object_subscription: ObjectSubscription,
+    ):
+        """
+        Creates a subscription pair from the input objects and adds it to the list
+        """
+        self._subscription_pairs.append(
+            SubscriptionPair(
+                analog_output_object,
+                object_subscription,
+                recent_times_buffer_length=self._recent_times_buffer_length,
+            )
+        )
+
+    def remove_subscription_pair(self, index: int):
+        """
+        Removes a subscription pair from the list at a specific index
+        Stops the subscription
+        Find index of a pair using index_from methods
+        """
+        subscription_pair = self._subscription_pairs.pop(index)
+        subscription_pair.stop_recording()
+
+    def index_from_device_object(
+        self, analog_output_object: AnalogOutputObject
+    ) -> int | None:
+        """
+        Gets index of a subscription pair from its analog_output_object
+        """
+        for index in range(len(self._subscription_pairs)):
+            index_analog_output_object = self._subscription_pairs[
+                index
+            ].get_analog_output_object()
+            if analog_output_object == index_analog_output_object:
+                return index
+        return None
+
+    def index_from_object_subscription(
+        self, object_subscription: ObjectSubscription
+    ) -> int | None:
+        """
+        Gets index of a subscription pair from its object_subscription
+        """
+        for index in range(len(self._subscription_pairs)):
+            index_object_subscription = self._subscription_pairs[
+                index
+            ].get_object_subscription()
+            if object_subscription == index_object_subscription:
+                return index
+        return None
+
+    def get_average_response_time(self) -> timedelta:
+        total_average_wait_times = timedelta(0)
+
+        for pair in self._subscription_pairs:
+            pair_average_time = pair.get_average_update_time()
+            if pair_average_time is not None:
+                total_average_wait_times += pair_average_time
+
+        total_average_wait_times /= len(self._subscription_pairs)
+
+        return total_average_wait_times
+
+    def get_average_reliability(self) -> float:
+        total_reliability = 0.0
+        total_pairs_averaged = 0
+
+        for pair in self._subscription_pairs:
+            pair_reliability = pair.get_reliability()
+            if pair_reliability is not None:
+                total_reliability += pair_reliability
+                total_pairs_averaged += 1
+
+        if total_pairs_averaged == 0:
+            return 0
+
+        total_reliability /= total_pairs_averaged
+
+        return total_reliability

src/fastcs_bacnet/diagnostics/BAC0/subscription_pair.py

@@ -0,0 +1,175 @@
+import math
+from datetime import datetime, timedelta
+
+from bacpypes3.primitivedata import PropertyIdentifier
+
+from fastcs_bacnet.dummy.BAC0.analog_output_object import AnalogOutputObject
+from fastcs_bacnet.practical.BAC0.object_subscription import ObjectSubscription
+
+
+class SubscriptionPair:
+    """
+    A class to track the time between an AnalogOutputObject being updated and its
+    corresponding ObjectSubscription recieving the update and related statistics
+    """
+
+    # subscription messages that have been sent but not recieved yet
+    _sent_buffer: list[tuple[float, datetime]]
+
+    # time between sending and recieving updates
+    # size limited, only the most recent sent updates are kept
+    _recent_receival_times: list[timedelta | None]
+
+    # max size of recent_recieval_times
+    _recent_times_buffer_length: int
+
+    _total_missed_updates: int = 0
+    _total_updates_recieved: int = 0
+    _total_update_wait_time: timedelta
+
+    # when we cant match a recieved value to a sent value
+    # shouldnt ever happed but we should definitely know if it does
+    _total_mystery_updates: int = 0
+
+    def __init__(
+        self,
+        analog_output_object: AnalogOutputObject,
+        object_subscription: ObjectSubscription,
+        recent_times_buffer_length: int = 20,
+    ):
+        """
+        analog_output_object: analog object of a dummy bacnet device
+            (cannot replicate this for real devices)
+        object_subscription: subscription object of a bacnet client device that
+            subscribes to the analog output object
+        recent_times_buffer_length: The number of send-recieve time differences to hold
+            at once. List would quickly get very large if this was not capped
+        Changes the diagnostic callback function of both objects to _on_send
+        and _on_recieve respectively
+        """
+        self._recent_times_buffer_length = recent_times_buffer_length
+        self._total_update_wait_time = timedelta(0)
+        self._analog_output_object = analog_output_object
+        self._object_subscription = object_subscription
+
+        self._sent_buffer = []
+        self._recent_receival_times = []
+
+        # maybe a bit cheeky to acess the analog_ouput_object underlying
+        # device variable directly. Should be with a getter??
+        self._analog_output_object.device_variable.set_diagnostic_callback(
+            self._on_send
+        )
+        self._object_subscription.set_diagnostic_callback(self._on_recieve)
+
+    def _on_send(self, old_value: float | None, new_value: float):
+        """
+        Callback to run when an update is sent by the device output object
+        Adds the update value and the current time to the _sent_buffer list
+        """
+        if old_value is None or abs(old_value - new_value) > 0.1:
+            self._sent_buffer.append((new_value, datetime.now()))
+        else:
+            print("updated value in dead zone")
+
+    def _on_recieve(self, property_identifier: str, property_value: float):
+        """
+        Callback to run when an update is recieved by the subscription object
+        Matches the recieved value to one that was sent by the device object
+        Values sent before the matched send value are marked as missed
+        Records the time between the update being sent and recieved
+        Adds recorded time as well as missed values to the _recent_recieval_times
+        """
+        if property_identifier != PropertyIdentifier.presentValue:
+            return
+
+        recieved_time = datetime.now()
+
+        sent_index = self._match(property_value)
+        if sent_index is None:
+            print("mystery update")
+            self._total_mystery_updates += 1
+            return
+
+        sent_time = self._sent_buffer[sent_index][1]
+
+        # every sent update before the current one must have been missed
+        # remove the sent buffer before the one we recieved
+        # add
+        for _ in range(sent_index):
+            self._sent_buffer.pop(0)
+            self._add_recieval_time(None)
+
+        self._sent_buffer.pop(0)
+        self._add_recieval_time(recieved_time - sent_time)
+
+    def _match(self, recieved_value: float) -> int | None:
+        """
+        Matches a recieved update value to the most recent matching value from the
+        sent updates list
+        Returns the index of the matching value
+        """
+
+        for i in range(len(self._sent_buffer)):
+            sent_value_i = self._sent_buffer[i][0]
+            if math.isclose(sent_value_i, recieved_value, rel_tol=0.0001):
+                return i
+
+        return None
+
+    def _add_recieval_time(self, receival_time: timedelta | None):
+        """
+        Adds a time difference (between sending and recieving)
+        to the _recent_receival_times list
+        Also handles changing of total missed updates, recieved updates and wait time
+        As well as keepin gthe list below its maximum length
+        """
+
+        if receival_time is None:
+            print("missed message")
+            self._total_missed_updates += 1
+        else:
+            print("valid receival")
+            self._total_updates_recieved += 1
+            self._total_update_wait_time += receival_time
+
+        self._recent_receival_times.append(receival_time)
+
+        while len(self._recent_receival_times) > self._recent_times_buffer_length:
+            self._recent_receival_times.pop(0)
+
+    def get_recent_recieval_times(self) -> list[timedelta | None]:
+        """
+        Returns a copy of recent_recieval_times
+        Dont have to deep copy as timedeltas are immutable
+        """
+        return list(self._recent_receival_times)
+
+    def get_total_missed_updates(self) -> int:
+        return self._total_missed_updates + len(self._sent_buffer)
+
+    def get_average_update_time(self) -> timedelta | None:
+        if self._total_updates_recieved == 0:
+            return None
+        return self._total_update_wait_time / self._total_updates_recieved
+
+    def get_reliability(self) -> float | None:
+        """
+        Total messages recieved / total messages sent
+        """
+        total_updates_sent = (
+            self._total_updates_recieved + self.get_total_missed_updates()
+        )
+        if total_updates_sent == 0:
+            return None
+        return self._total_updates_recieved / (total_updates_sent)
+
+    def get_analog_output_object(self) -> AnalogOutputObject:
+        return self._analog_output_object
+
+    def get_object_subscription(self) -> ObjectSubscription:
+        return self._object_subscription
+
+    def stop_recording(self):
+        self._analog_output_object.device_variable.set_diagnostic_callback(None)
+        self._object_subscription.set_diagnostic_callback(None)

src/fastcs_bacnet/diagnostics/FastCS/attribute_tracker.py

@@ -0,0 +1,49 @@
+from datetime import datetime, timedelta
+
+from fastcs_bacnet.diagnostics.FastCS.tracked_read_attribute import TrackedAttrR
+
+
+class AttributeTracker:
+    """
+    Records and stores updates times from a TrackedAttrR
+    """
+
+    _recent_times: list[timedelta]
+    _first_update: datetime | None
+    _last_update: datetime | None
+
+    def __init__(self, tracked_attribute: TrackedAttrR, history_size: int = 20):
+        """
+        tracked_attribute: Attribute to track
+        history_size: Maximum number of update times to store in the recent_times queue
+        """
+
+        self._tracked_attribute = tracked_attribute
+        self._history_size = history_size
+
+        self._tracked_attribute.set_diagnostic_callback(self.update_callback)
+
+    def update_callback(self, start_time: datetime, end_time: datetime):
+        """
+        The callback function that gets called after the attributes
+        update method is called
+        start_time is when the update method is first called
+        end_time is after the original update method has been resolved
+        """
+
+        self._recent_times.append(end_time - start_time)
+        while len(self._recent_times) > self._history_size:
+            self._recent_times.pop(0)
+
+        if self._first_update is None:
+            self._first_update = start_time
+        self._last_update = start_time
+
+    def get_first_update_time(self) -> datetime | None:
+        return self._first_update
+
+    def get_last_update_time(self) -> datetime | None:
+        return self._last_update
+
+    def get_recent_times(self) -> list[timedelta]:
+        return list(self._recent_times)