Merge remote-tracking branch 'origin/v0.26' into v0.26

Author: velovix

brainframe/api/bf_codecs/alarm_codecs.py

@@ -1,46 +1,47 @@
 from typing import List, Optional
 
+from dataclasses import dataclass
+
 from .base_codecs import Codec
 from .condition_codecs import ZoneAlarmCountCondition, ZoneAlarmRateCondition
 
 
+@dataclass
 class ZoneAlarm(Codec):
     """This is the configuration for an alarm."""
 
-    def __init__(self, *,
-                 name: str,
-                 count_conditions: List[ZoneAlarmCountCondition],
-                 rate_conditions: List[ZoneAlarmRateCondition],
-                 use_active_time: bool,
-                 active_start_time: str,
-                 active_end_time: str,
-                 id_: int = None,
-                 zone_id: int = None,
-                 stream_id: int = None):
-        self.name: str = name
-        """A friendly name for the zone alarm"""
-        self.id: int = id_
-        """A unique identifier"""
-        self.zone_id: int = zone_id
-        """The ID of the zone this alarm is associated with"""
-        self.stream_id: int = stream_id
-        """The ID of the stream the associated zone is in"""
-        self.count_conditions: List[ZoneAlarmCountCondition] = count_conditions
-        """All count conditions for this alarm"""
-        self.rate_conditions: List[ZoneAlarmRateCondition] = rate_conditions
-        """All rate conditions for this alarm"""
-        self.use_active_time: bool = use_active_time
-        """If True, the alarm will only be triggered when the current time is
-        between the active_start_time and active_end_time.
-        """
-        self.active_start_time: str = active_start_time
-        """The time of day where this alarm starts being active, in the format
-        "hh:mm:ss"
-        """
-        self.active_end_time: str = active_end_time
-        """The time of day where this alarm starts being active, in the format
-        "hh:mm:ss"
-        """
+    name: str
+    """A friendly name for the zone alarm"""
+
+    zone_id: int
+    """The ID of the zone this alarm is associated with"""
+
+    stream_id: int
+    """The ID of the stream the associated zone is in"""
+
+    count_conditions: List[ZoneAlarmCountCondition]
+    """All count conditions for this alarm"""
+
+    rate_conditions: List[ZoneAlarmRateCondition]
+    """All rate conditions for this alarm"""
+
+    use_active_time: bool
+    """If True, the alarm will only be triggered when the current time is
+    between the active_start_time and active_end_time.
+    """
+
+    active_start_time: str
+    """The time of day where this alarm starts being active, in the format
+    "hh:mm:ss"
+    """
+
+    active_end_time: str
+    """The time of day where this alarm starts being active, in the format
+    "hh:mm:ss"
+    """
+
+    id: Optional[int] = None
+    """A unique identifier"""
 
     def to_dict(self):
         d = dict(self.__dict__)
@@ -59,7 +60,7 @@ def from_dict(d):
                            for cond in d["rate_conditions"]]
 
         return ZoneAlarm(name=d["name"],
-                         id_=d["id"],
+                         id=d["id"],
                          count_conditions=count_conditions,
                          rate_conditions=rate_conditions,
                          use_active_time=d["use_active_time"],
@@ -69,45 +70,44 @@ def from_dict(d):
                          stream_id=d["stream_id"])
 
 
+@dataclass
 class Alert(Codec):
     """This is sent when an Alarm has been triggered."""
 
-    def __init__(self, *,
-                 alarm_id: int,
-                 zone_id: int,
-                 stream_id: int,
-                 start_time: float,
-                 end_time: float,
-                 verified_as: Optional[bool],
-                 id_: int = None):
-        self.id: int = id_
-        """A unique identifier"""
-        self.alarm_id: int = alarm_id
-        """The ID of the alarm that this alert came from"""
-        self.zone_id: int = zone_id
-        """The ID of the zone this alert pertains to"""
-        self.stream_id: int = stream_id
-        """The ID of the stream this alert pertains to"""
-        self.start_time: float = start_time
-        """When the event started happening, in Unix Time (seconds)"""
-        self.end_time: Optional[float] = end_time
-        """When the event stopped happening, in Unix Time (seconds), or None
-        if the alert is ongoing.
-        """
-        self.verified_as: Optional[bool] = verified_as
-        """
-        - If True, then this alert was labeled by a person as legitimate
-        - If False, then this alert was labeled by a person as a false alarm
-        - If None, then this alert has not been reviewed by a person
-        """
+    alarm_id: int
+    """The ID of the alarm that this alert came from"""
+
+    zone_id: int
+    """The ID of the zone this alert pertains to"""
+
+    stream_id: int
+    """The ID of the stream this alert pertains to"""
+
+    start_time: float
+    """When the event started happening, in Unix Time (seconds)"""
+
+    end_time: Optional[float]
+    """When the event stopped happening, in Unix Time (seconds), or None
+    if the alert is ongoing.
+    """
+
+    verified_as: Optional[bool]
+    """
+    - If True, then this alert was labeled by a person as legitimate
+    - If False, then this alert was labeled by a person as a false alarm
+    - If None, then this alert has not been reviewed by a person
+    """
+
+    id: Optional[int] = None
+    """A unique identifier"""
 
     def to_dict(self):
         d = dict(self.__dict__)
         return d
 
     @staticmethod
     def from_dict(d):
-        return Alert(id_=d["id"],
+        return Alert(id=d["id"],
                      alarm_id=d["alarm_id"],
                      zone_id=d["zone_id"],
                      stream_id=d["stream_id"],

brainframe/api/bf_codecs/base_codecs.py

@@ -20,16 +20,6 @@ def to_json(self) -> str:
     def from_json(cls, j: str):
         return cls.from_dict(json.loads(j))
 
-    def __repr__(self):
-        """Formats a string in the format of
-        ClassName(arg1=val, arg2=val, arg3=val)"""
-        argstring = ""
-        for argname, argval in self.to_dict().items():
-            if len(argstring):
-                argstring += ", "
-            argstring += f"{argname}={argval}"
-        return f"{type(self).__name__}({argstring})"
-
     def __eq__(self, other):
         if type(other) is dict:
             return self.to_dict() == other

brainframe/api/bf_codecs/condition_codecs.py

@@ -1,6 +1,8 @@
 from enum import Enum
 from typing import Optional
 
+from dataclasses import dataclass
+
 from .base_codecs import Codec
 from .detection_codecs import Attribute
 
@@ -46,43 +48,42 @@ def values(cls):
         return [v.value for v in cls]
 
 
+@dataclass
 class ZoneAlarmCountCondition(Codec):
     """A condition that must be met for an alarm to go off. Compares the number
     of objects in a zone to some number.
     """
 
-    def __init__(self, *,
-                 test: CountConditionTestType,
-                 check_value: int,
-                 with_class_name: str,
-                 with_attribute: Optional[Attribute],
-                 window_duration: float,
-                 window_threshold: float,
-                 intersection_point: IntersectionPointType,
-                 id_: int = None):
-        self.test: CountConditionTestType = test
-        """The way that the check value will be compared to the actual count
-        """
-        self.check_value: int = check_value
-        """The value to test the actual count against"""
-        self.with_class_name: str = with_class_name
-        """The class name of the objects to count"""
-        self.with_attribute: Optional[Attribute] = with_attribute
-        """If provided, only objects that have this attribute value will be
-        counted.
-        """
-        self.window_duration: float = window_duration
-        """The sliding window duration for this condition"""
-        self.window_threshold: float = window_threshold
-        """The portion of time during the sliding window duration that this
-        condition must be true in order for the associated alarm to trigger
-        """
-        self.intersection_point: IntersectionPointType = intersection_point
-        """The point in each detection that must be within the zone in order
-        for that detection to be counted as in that zone
-        """
-        self.id: int = id_
-        """A unique identifier"""
+    test: CountConditionTestType
+    """The way that the check value will be compared to the actual count
+    """
+
+    check_value: int
+    """The value to test the actual count against"""
+
+    with_class_name: str
+    """The class name of the objects to count"""
+
+    with_attribute: Optional[Attribute]
+    """If provided, only objects that have this attribute value will be
+    counted.
+    """
+
+    window_duration: float
+    """The sliding window duration for this condition"""
+
+    window_threshold: float
+    """The portion of time during the sliding window duration that this
+    condition must be true in order for the associated alarm to trigger
+    """
+
+    intersection_point: IntersectionPointType
+    """The point in each detection that must be within the zone in order
+    for that detection to be counted as in that zone
+    """
+
+    id: Optional[int] = None
+    """A unique identifier"""
 
     def __repr__(self):
         condition_str = condition_test_map[self.test.value]
@@ -118,7 +119,7 @@ def from_dict(d: dict):
             window_duration=d["window_duration"],
             window_threshold=d["window_threshold"],
             intersection_point=intersection_point,
-            id_=d["id"])
+            id=d["id"])
 
 
 class RateConditionTestType(Enum):
@@ -146,48 +147,47 @@ def values(cls):
         return [v.value for v in cls]
 
 
+@dataclass
 class ZoneAlarmRateCondition(Codec):
     """A condition that must be met for an alarm to go off. Compares the rate
     of change in the count of some object against a test value.
     """
 
-    direction_map = {DirectionType.ENTERING: "entered",
-                     DirectionType.EXITING: "exited",
-                     DirectionType.ENTERING_OR_EXITING: "entered or exited"}
-
-    def __init__(self, *,
-                 test: RateConditionTestType,
-                 duration: float,
-                 change: float,
-                 direction: DirectionType,
-                 with_class_name: str,
-                 with_attribute: Optional[Attribute],
-                 intersection_point: IntersectionPointType,
-                 id_: int = None):
-        self.test: RateConditionTestType = test
-        """The way that the change value will be compared to the actual rate"""
-        self.duration: float = duration
-        """The time in seconds for this rate change to occur"""
-        self.change: float = change
-        """The rate change value to compare the actual rate value against"""
-        self.direction: DirectionType = direction
-        """The direction of flow this condition tests for"""
-        self.with_class_name: str = with_class_name
-        """The class name of the objects to measure rate of change for"""
-        self.with_attribute: Optional[Attribute] = with_attribute
-        """If provided, only objects that have this attribute will be counted
-        in the rate calculation
-        """
-        self.intersection_point: IntersectionPointType = intersection_point
-        """The point in each detection that must be within the zone in order
-        for that detection to be counted as in that zone
-        """
-        self.id: int = id_
-        """A unique identifier"""
+    _direction_map = {DirectionType.ENTERING: "entered",
+                      DirectionType.EXITING: "exited",
+                      DirectionType.ENTERING_OR_EXITING: "entered or exited"}
+
+    test: RateConditionTestType
+    """The way that the change value will be compared to the actual rate"""
+
+    duration: float
+    """The time in seconds for this rate change to occur"""
+
+    change: float
+    """The rate change value to compare the actual rate value against"""
+
+    direction: DirectionType
+    """The direction of flow this condition tests for"""
+
+    with_class_name: str
+    """The class name of the objects to measure rate of change for"""
+
+    with_attribute: Optional[Attribute]
+    """If provided, only objects that have this attribute will be counted
+    in the rate calculation
+    """
+
+    intersection_point: IntersectionPointType
+    """The point in each detection that must be within the zone in order
+    for that detection to be counted as in that zone
+    """
+
+    id: int = None
+    """A unique identifier"""
 
     def __repr__(self):
         condition_str = condition_test_map[self.test.value]
-        direction_str = self.direction_map[self.direction]
+        direction_str = self._direction_map[self.direction]
         return f"{condition_str} {self.change} {self.with_class_name}(s) " \
                f"{direction_str} within {self.duration} seconds"
 
@@ -220,4 +220,4 @@ def from_dict(d: dict):
             with_class_name=d["with_class_name"],
             with_attribute=with_attribute,
             intersection_point=intersection_point,
-            id_=d["id"])
+            id=d["id"])