Skip to content
This repository was archived by the owner on Jun 28, 2024. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ Commits to `main` following [Conventional Commits](https://www.conventionalcommi

Our tests use a seam sandbox environment given by the environment
variables `SEAM_SANDBOX_API_KEY`. If you want to run the tests, you should
first create a sandbox workspace [on your dashboard](https://dashboard.getseam.com)
first create a sandbox workspace [on your Developer Console](https://console.getseam.com)
then create a sandbox workspace.

> NOTE: For installation on m1 mac, you may need to export the following lines
Expand Down
4 changes: 3 additions & 1 deletion seamapi/action_attempts.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,8 +100,10 @@ def poll_until_ready(

Parameters
----------
action_attempt : ActionAttemptId or ActionAttempt
action_attempt: ActionAttemptId or ActionAttempt
Action attempt id or ActionAttempt to be polled
should_raise: bool
Should raise an exception if action attempt status is 'error'

Returns
------
Expand Down
42 changes: 41 additions & 1 deletion seamapi/noise_sensors.py
Original file line number Diff line number Diff line change
@@ -1 +1,41 @@
from seamapi.types import AbstractNoiseSensors, AbstractNoiseThresholds
from seamapi.types import AbstractNoiseSensors, AbstractSeam as Seam
from seamapi.noise_thresholds import NoiseThresholds
from seamapi.utils.report_error import report_error


class NoiseSensors(AbstractNoiseSensors):
"""
A class to interact with noise sensors through the Seam API

...

Attributes
----------
seam : Seam
Initial seam class

Properties
-------
noise_thresholds
An instance of the NoiseThresholds class designed to interact with noise thresholds
"""

seam: Seam

def __init__(self, seam: Seam):
"""
Parameters
----------
seam : Seam
Intial seam class
"""
self.seam = seam
self._noise_thresholds = NoiseThresholds(seam=seam)

@property
def noise_thresholds(self) -> NoiseThresholds:
return self._noise_thresholds

@report_error
def list_noise_levels(self, noise_threshold_id):
raise NotImplementedError()
125 changes: 102 additions & 23 deletions seamapi/noise_thresholds.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
from dataclasses import asdict
from seamapi.types import (
NoiseThreshold,
AbstractNoiseThresholds,
AbstractSeam as Seam,
ActionAttempt,
ActionAttemptError,
)
from typing import List, Optional
from typing import List, Optional, Union
import requests
import json

from seamapi.utils.report_error import report_error

Expand All @@ -28,6 +30,8 @@ class NoiseThresholds(AbstractNoiseThresholds):
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
delete(noise_threshold_id, device_id, sync=None)
Deletes a noise threshold on a noise-monitoring device
"""

seam: Seam
Expand Down Expand Up @@ -68,30 +72,35 @@ def list(
params={"device_id": device_id},
)

return res["noise_thresholds"]
noise_thresholds = res["noise_thresholds"]

return [NoiseThreshold.from_dict(nt) for nt in noise_thresholds]

@report_error
def create(
self,
device_id: str,
starts_daily_at: str,
ends_daily_at: str,
sync: Optional[bool] = None,
name: Optional[str] = None,
noise_threshold_decibels: Optional[float] = None,
noise_threshold_nrs: Optional[float] = None,
) -> List[NoiseThreshold]:
wait_for_action_attempt: Optional[bool] = True,
) -> Union[ActionAttempt, NoiseThreshold]:
"""Creates a noise threshold.

Parameters
----------
device_id : str
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
Time when noise threshold becomes active daily
ends_daily_at: str,
Time when noise threshold becomes inactive
Time when noise threshold becomes inactive daily
name: Optional[str]
Noise threshold name
wait_for_action_attempt: Optional[bool]
Should wait for action attempt to resolve
noise_threshold_decibels: Optional[float],
The noise level in decibels
noise_threshold_nrs: Optional[float],
Expand All @@ -103,7 +112,7 @@ def create(

Returns
------
ActionAttempt
ActionAttempt or NoiseThreshold
"""
params = {
"device_id": device_id,
Expand All @@ -112,9 +121,9 @@ def create(
}

arguments = {
"sync": sync,
"noise_threshold_decibels": noise_threshold_decibels,
"noise_threshold_nrs": noise_threshold_nrs,
"name": name,
}

for name in arguments:
Expand All @@ -124,29 +133,99 @@ def create(
res = self.seam.make_request(
"POST",
"/noise_sensors/noise_thresholds/create",
params={"device_id": device_id},
json=params,
)

json_aa = res["action_attempt"]
error = None
aa_error = None
if "error" in json_aa and json_aa["error"] is not None:
error = ActionAttemptError(
aa_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,
if not wait_for_action_attempt or aa_error:
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=aa_error,
)

updated_action_attempt = self.seam.action_attempts.poll_until_ready(
json_aa["action_attempt_id"]
)

@report_error
def delete(self, noise_threshold_id):
raise NotImplementedError()
action_attempt_result = getattr(updated_action_attempt, "result", None)
noise_threshold = action_attempt_result.get("noise_threshold", None)
if not action_attempt_result or not noise_threshold:
raise Exception(
"Failed to create noise_threshold: no noise_threshold returned: "
+ json.dumps(asdict(updated_action_attempt))
)

return NoiseThreshold.from_dict(noise_threshold)

@report_error
def update(self, noise_threshold_id):
raise NotImplementedError()

@report_error
def delete(
self,
noise_threshold_id: str,
device_id: str,
wait_for_action_attempt: Optional[bool] = True,
) -> ActionAttempt:
"""Deletes a noise threshold.

Parameters
----------
noise_threshold_id : str
Id of a noise threshold to delete
device_id : str
Device ID of a device to delete noise threshold of
wait_for_action_attempt: Optional[bool]
Should wait for delete action attempt to resolve

Raises
------
Exception
If the API request wasn't successful.

Returns
------
ActionAttempt
"""
res = self.seam.make_request(
"DELETE",
"/noise_sensors/noise_thresholds/delete",
json={
"noise_threshold_id": noise_threshold_id,
"device_id": device_id,
},
)

json_aa = res["action_attempt"]
aa_error = None
if "error" in json_aa and json_aa["error"] is not None:
aa_error = ActionAttemptError(
type=json_aa["error"]["type"],
message=json_aa["error"]["message"],
)

if not wait_for_action_attempt or aa_error:
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=aa_error,
)

updated_action_attempt = self.seam.action_attempts.poll_until_ready(
json_aa["action_attempt_id"]
)

return updated_action_attempt
4 changes: 2 additions & 2 deletions seamapi/routes.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from .noise_sensors import NoiseSensors
from .types import AbstractRoutes
from .workspaces import Workspaces
from .devices import Devices, UnmanagedDevices
Expand All @@ -7,7 +8,6 @@
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):
Expand All @@ -19,7 +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)
self.noise_sensors = NoiseSensors(seam=self)

def make_request(self):
raise NotImplementedError()
Loading