feat: Noise Sensors (#84)

Co-authored-by: andrii-balitskyi <[email protected]>

Author: seveibar

seamapi/noise_sensors.py

@@ -0,0 +1 @@
+from seamapi.types import AbstractNoiseSensors, AbstractNoiseThresholds

seamapi/noise_thresholds.py

@@ -0,0 +1,152 @@
+from seamapi.types import (
+    NoiseThreshold,
+    AbstractNoiseThresholds,
+    AbstractSeam as Seam,
+    ActionAttempt,
+    ActionAttemptError,
+)
+from typing import List, Optional
+import requests
+
+from seamapi.utils.report_error import report_error
+
+
+class NoiseThresholds(AbstractNoiseThresholds):
+    """
+    A class to interact with noise thresholds through the Seam API
+
+    ...
+
+    Attributes
+    ----------
+    seam : Seam
+        Initial seam class
+
+    Methods
+    -------
+    list(device_id)
+        Gets a list of noise thresholds of a noise-monitoring device
+    create(device_id, starts_daily_at, ends_daily_at, sync=None, noise_threshold_decibels=None, noise_threshold_nrs=None)
+        Creates a noise threshold on a noise-monitoring device
+    """
+
+    seam: Seam
+
+    def __init__(self, seam: Seam):
+        """
+        Parameters
+        ----------
+        seam : Seam
+          Intial seam class
+        """
+        self.seam = seam
+
+    @report_error
+    def list(
+        self,
+        device_id: str,
+    ) -> List[NoiseThreshold]:
+        """Gets a list of noise thresholds.
+
+        Parameters
+        ----------
+        device_id : str
+            Device ID of a device to list noise thresholds of
+
+        Raises
+        ------
+        Exception
+            If the API request wasn't successful.
+
+        Returns
+        ------
+            A list of noise thresholds.
+        """
+        res = self.seam.make_request(
+            "GET",
+            "/noise_sensors/noise_thresholds/list",
+            params={"device_id": device_id},
+        )
+
+        return res["noise_thresholds"]
+
+    @report_error
+    def create(
+        self,
+        device_id: str,
+        starts_daily_at: str,
+        ends_daily_at: str,
+        sync: Optional[bool] = None,
+        noise_threshold_decibels: Optional[float] = None,
+        noise_threshold_nrs: Optional[float] = None,
+    ) -> List[NoiseThreshold]:
+        """Creates a noise threshold.
+
+        Parameters
+        ----------
+        device_id : str
+            Device ID of a device to list noise thresholds of
+        sync: Optional[bool]
+            Should wait for action attempt to resolve
+        starts_daily_at: str,
+            Time when noise threshold becomes active
+        ends_daily_at: str,
+            Time when noise threshold becomes inactive
+        noise_threshold_decibels: Optional[float],
+            The noise level in decibels
+        noise_threshold_nrs: Optional[float],
+            Noise Level in Noiseaware Noise Risk Score (NRS)
+        Raises
+        ------
+        Exception
+            If the API request wasn't successful.
+
+        Returns
+        ------
+            ActionAttempt
+        """
+        params = {
+            "device_id": device_id,
+            "starts_daily_at": starts_daily_at,
+            "ends_daily_at": ends_daily_at,
+        }
+
+        arguments = {
+            "sync": sync,
+            "noise_threshold_decibels": noise_threshold_decibels,
+            "noise_threshold_nrs": noise_threshold_nrs,
+        }
+
+        for name in arguments:
+            if arguments[name]:
+                params.update({name: arguments[name]})
+
+        res = self.seam.make_request(
+            "POST",
+            "/noise_sensors/noise_thresholds/create",
+            params={"device_id": device_id},
+        )
+
+        json_aa = res["action_attempt"]
+        error = None
+        if "error" in json_aa and json_aa["error"] is not None:
+            error = ActionAttemptError(
+                type=json_aa["error"]["type"],
+                message=json_aa["error"]["message"],
+            )
+
+        return ActionAttempt(
+            action_attempt_id=json_aa["action_attempt_id"],
+            status=json_aa["status"],
+            action_type=json_aa["action_type"],
+            result=json_aa["result"],
+            error=error,
+        )
+
+    @report_error
+    def delete(self, noise_threshold_id):
+        raise NotImplementedError()
+
+    @report_error
+    def update(self, noise_threshold_id):
+        raise NotImplementedError()

seamapi/routes.py

@@ -7,6 +7,7 @@
 from .locks import Locks
 from .access_codes import AccessCodes
 from .action_attempts import ActionAttempts
+from .noise_thresholds import NoiseThresholds
 
 class Routes(AbstractRoutes):
     def __init__(self):
@@ -18,6 +19,7 @@ def __init__(self):
       self.locks = Locks(seam=self)
       self.access_codes = AccessCodes(seam=self)
       self.action_attempts = ActionAttempts(seam=self)
+      self.noise_thresholds = NoiseThresholds(seam=self)
 
     def make_request(self):
       raise NotImplementedError()

seamapi/types.py

@@ -46,18 +46,12 @@ def __init__(
             f'Action Attempt for "{action_type}" Failed. {error_type}: {error_message} (action_attempt_id={action_attempt_id})'
         )
 
+
 class WaitForAccessCodeFailedException(Exception):
-    def __init__(
-        self,
-        message: str,
-        access_code_id: str,
-        errors: Optional[list] = []
-    ):
+    def __init__(self, message: str, access_code_id: str, errors: Optional[list] = []):
         self.access_code_id = access_code_id
         self.errors = errors
-        super().__init__(
-            f'Failed while waiting for access code. ${message}'
-        )
+        super().__init__(f"Failed while waiting for access code. ${message}")
 
 
 @dataclass
@@ -92,15 +86,9 @@ class UnmanagedDevice:
 @dataclass
 class Event:
     event_id: str
-    noiseaware_id: str
-    activity_zone_id: str
     event_class: Union[str, None]
     event_type: Union[str, None]
-    duration: Union[int, None]
-    local_end_time: Union[str, None]
-    local_start_time: Union[str, None]
-    start_time: Union[str, None]
-    end_time: Union[str, None]
+    device_id: Optional[str]
     created_at: Union[str, None]
 
 
@@ -161,6 +149,7 @@ class ConnectedAccount:
     errors: List[str]
     custom_metadata: Dict[str, Union[str, int, bool, None]]
 
+
 @dataclass_json
 @dataclass
 class AccessCode:
@@ -174,6 +163,21 @@ class AccessCode:
     common_code_key: Optional[str] = None
 
 
+@dataclass_json
+@dataclass
+class NoiseThreshold:
+    noise_threshold_id: str
+    name: str
+    noise_threshold_decibels: float
+
+    starts_daily_at: Optional[str]
+    ends_daily_at: Optional[str]
+
+    noise_threshold_nrs: Optional[float]
+
+    created_at: str
+
+
 class AbstractActionAttempts(abc.ABC):
     @abc.abstractmethod
     def get(
@@ -265,6 +269,41 @@ def delete(
         raise NotImplementedError
 
 
+class AbstractNoiseThresholds(abc.ABC):
+    @abc.abstractmethod
+    def create(
+        self,
+        name: str,
+        noise_threshold_decibels: Optional[float] = None,
+        noise_threshold_nrs: Optional[float] = None,
+    ) -> NoiseThreshold:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def update(
+        self,
+        noise_threshold_id: str,
+        name: Optional[str] = None,
+        noise_threshold_decibels: Optional[str] = None,
+        noise_threshold_nrs: Optional[str] = None,
+    ) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def list(self) -> List[NoiseThreshold]:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def delete(self, noise_threshold_id: str) -> None:
+        raise NotImplementedError
+
+
+class AbstractNoiseSensors(abc.ABC):
+    @abc.abstractmethod
+    def list_noise_levels(self, starting_after=None, ending_before=None) -> None:
+        raise NotImplementedError
+
+
 class AbstractDevices(abc.ABC):
     @abc.abstractmethod
     def list(self) -> List[Device]:
@@ -345,6 +384,7 @@ class AbstractRoutes(abc.ABC):
     devices: AbstractDevices
     access_codes: AbstractAccessCodes
     action_attempts: AbstractActionAttempts
+    noise_thresholds: AbstractNoiseThresholds
 
     @abc.abstractmethod
     def make_request(self, method: str, path: str, **kwargs) -> Any:

tests/conftest.py

@@ -37,19 +37,20 @@ def seam_backend():
             container_host = "host.docker.internal"
             db_url = db_url.replace(db_host, container_host)
             db_host = container_host
+        elif db_host == "localhost":
+            container_host = "172.17.0.1"  # linux equivalent of host.docker.internal
+            db_url = db_url.replace(db_host, container_host)
+            db_host = container_host
 
         with DockerContainer(
-            os.environ.get(
-                "SEAM_CONNECT_IMAGE", "ghcr.io/seamapi/seam-connect"
-            )
-        ).with_env(
-            "DATABASE_URL", db_url
-        ).with_env(
+            os.environ.get("SEAM_CONNECT_IMAGE", "ghcr.io/seamapi/seam-connect")
+        ).with_env("DATABASE_URL", db_url).with_env(
             "POSTGRES_DATABASE", "postgres"
         ).with_env(
             "NODE_ENV", "test"
         ).with_env(
-            "POSTGRES_HOST", db_host,
+            "POSTGRES_HOST",
+            db_host,
         ).with_env(
             "SERVER_BASE_URL", "http://localhost:3020"
         ).with_env(
@@ -77,6 +78,7 @@ def seam(seam_backend: Any, dotenv_fixture: Any):
     seam.make_request("POST", "/workspaces/reset_sandbox")
     yield seam
 
+
 @pytest.fixture
 def fake_sentry(monkeypatch):
     sentry_dsn = "https://[email protected]/123"
@@ -86,9 +88,11 @@ def fake_sentry(monkeypatch):
     sentry_init_args = {}
     sentry_capture_exception_calls = []
     sentry_add_breadcrumb_calls = []
+
     class TestSentryClient(object):
         def __init__(self, *args, **kwargs):
             sentry_init_args.update(kwargs)
+
         def set_context(self, *args, **kwargs):
             pass
 
@@ -100,9 +104,11 @@ def set_context(self, *args, **kwargs):
 
     class TestSentryHub(object):
         def __init__(self, *args, **kwargs):
-          self.scope = TestSentryScope()
+            self.scope = TestSentryScope()
+
         def capture_exception(self, *args, **kwargs):
-          sentry_capture_exception_calls.append((args, kwargs))
+            sentry_capture_exception_calls.append((args, kwargs))
+
         def add_breadcrumb(self, *args, **kwargs):
             sentry_add_breadcrumb_calls.append((args, kwargs))