"""
Environmental Sensors API client and functional wrappers for pykada.
The :class:`SensorsClient` provides access to sensor alert history and
time-series sensor readings (temperature, humidity, air quality, etc.)
through Verkada's Sensor API.
"""
from typeguard import typechecked
from typing import List, Dict, Any, Generator
from pykada.endpoints import SENSOR_ALERT_ENDPOINT, SENSOR_DATA_ENDPOINT
from pykada.helpers import remove_null_fields
from pykada.enums import SENSOR_FIELD_ENUM
from pykada.verkada_client import BaseClient
from pykada.verkada_requests import *
[docs]
def check_sensor_fields(fields):
"""
Check if the provided sensor fields are valid.
:param fields: List of sensor fields to check.
"""
if not fields:
return
valid = set(SENSOR_FIELD_ENUM.values())
invalid_fields = set(fields) - valid
if invalid_fields:
raise ValueError(
f"Sensor field types {invalid_fields} are not in the "
f"list of valid types: {list(valid)}"
)
[docs]
class SensorsClient(BaseClient):
"""
Client for interacting with Verkada's Environmental Sensors API.
This client provides methods to retrieve sensor alerts and sensor data.
"""
@typechecked
def __init__(self,
api_key: Optional[str] = None,
token_manager: Optional[VerkadaTokenManager] = None):
super().__init__(api_key, token_manager)
[docs]
def get_all_sensor_alerts(
self,
device_ids: List[str],
start_time: Optional[int] = None,
end_time: Optional[int] = None,
fields: Optional[List[str]] = None
) -> Generator[Any, None, None]:
"""
Returns all alerts for a set of sensors in an organization over a specified time range.
:param device_ids: List of sensor IDs.
:param start_time: Start time (Unix timestamp in seconds).
:param end_time: End time (Unix timestamp in seconds).
:param fields: List of sensor fields to filter alerts.
:return: A dictionary representing the JSON response containing sensor alerts.
:raises ValueError: If device_ids is an empty list.
"""
if not device_ids:
raise ValueError("device_ids must be a non-empty list")
check_sensor_fields(fields)
params = {
"device_ids": device_ids,
"start_time": start_time,
"end_time": end_time,
"fields": fields,
}
return VerkadaRequestManager.iterate_paginated_results(
lambda **kwargs: self.get_sensor_alerts(**kwargs),
initial_params=params,
next_token_key="page_cursor",
items_key="alert_events"
)
[docs]
@typechecked
def get_sensor_alerts(self,
device_ids: List[str],
start_time: Optional[int] = None,
end_time: Optional[int] = None,
page_size: Optional[int] = None,
page_token: Optional[str] = None,
fields: Optional[List[str]] = None) -> Dict:
"""
Returns all alerts for a set of sensors in an organization over a specified time range.
:param device_ids: List of sensor IDs.
:param start_time: Start time (Unix timestamp in seconds).
:param end_time: End time (Unix timestamp in seconds).
:param page_size: Number of items per page.
:param page_token: Token for pagination.
:param fields: List of sensor fields to filter alerts.
:return: A dictionary representing the JSON response containing sensor alerts.
:raises ValueError: If device_ids is an empty list.
"""
if not device_ids:
raise ValueError("device_ids must be a non-empty list")
check_sensor_fields(fields)
params = {
"device_ids": ",".join(device_id for device_id in device_ids),
"start_time": start_time,
"end_time": end_time,
"page_size": page_size,
"page_token": page_token if page_token else None,
"fields": ",".join(field for field in fields) if fields else None,
}
# Remove keys with a value of None.
params = remove_null_fields(params)
return self.request_manager.get(SENSOR_ALERT_ENDPOINT, params=params)
[docs]
def get_all_sensor_data(
self,
device_id: str,
start_time: Optional[int] = None,
end_time: Optional[int] = None,
fields: Optional[List[str]] = None,
interval: Optional[str] = None
) -> Generator[Any, None, None]:
"""
Returns sensor readings for a particular sensor over a specified time range.
:param device_id: The unique identifier of the sensor.
:param start_time: Start time for sensor data (Unix timestamp in seconds).
:param end_time: End time for sensor data (Unix timestamp in seconds).
:param fields: List of sensor fields to include in the response.
:param interval: The time interval for the requested sensor data.
Data is stored at 1-second intervals for 30 days and at 5-minute intervals
for data between 30 and 365 days old. Specify a number followed by a unit
(``s``, ``m``, or ``h``), e.g. ``"5m"`` for 5-minute intervals.
If omitted, a default resolution is calculated from the time range.
:return: A dictionary representing the JSON response containing sensor data.
"""
check_sensor_fields(fields)
params = {
"device_id": device_id,
"start_time": start_time,
"end_time": end_time,
"fields": fields,
"interval": interval if interval else None,
}
params = remove_null_fields(params)
return VerkadaRequestManager.iterate_paginated_results(
lambda **kwargs: self.get_sensor_data(**kwargs),
initial_params=params,
next_token_key="page_cursor",
items_key="data"
)
[docs]
@typechecked
def get_sensor_data(self,
device_id: str,
start_time: Optional[int] = None,
end_time: Optional[int] = None,
page_size: Optional[int] = None,
page_token: Optional[str] = None,
fields: Optional[List[str]] = None,
interval: Optional[str] = None) -> Dict:
"""
Returns sensor readings for a particular sensor over a specified time range.
:param device_id: The unique identifier of the sensor.
:param start_time: Start time for sensor data (Unix timestamp in seconds).
:param end_time: End time for sensor data (Unix timestamp in seconds).
:param page_size: Number of items per page.
:param page_token: Token for pagination.
:param fields: List of sensor fields to include in the response.
:param interval: The time interval for the requested sensor data.
Data is stored at 1-second intervals for 30 days and at 5-minute intervals
for data between 30 and 365 days old. Specify a number followed by a unit
(``s``, ``m``, or ``h``), e.g. ``"5m"`` for 5-minute intervals.
If omitted, a default resolution is calculated from the time range.
:return: A dictionary representing the JSON response containing sensor data.
"""
check_sensor_fields(fields)
params = {
"device_id": device_id,
"start_time": start_time,
"end_time": end_time,
"page_size": page_size,
"page_token": page_token if page_token else None,
"fields": ",".join(str(field) for field in fields) if fields else None,
"interval": interval if interval else None,
}
params = remove_null_fields(params)
return self.request_manager.get(SENSOR_DATA_ENDPOINT, params=params)
# ---------------------------------------------------------------------------
# Module-level default client — shared across all functional wrappers.
# ---------------------------------------------------------------------------
_default_sensors_client: Optional[SensorsClient] = None
def _get_default_client() -> SensorsClient:
global _default_sensors_client
if _default_sensors_client is None:
_default_sensors_client = SensorsClient()
return _default_sensors_client
[docs]
@typechecked
def get_all_sensor_alerts(device_ids: List[str], start_time: Optional[int] = None, end_time: Optional[int] = None, fields: Optional[List[str]] = None):
"""
Returns all alerts for a set of sensors in an organization over a specified time range.
:param device_ids: List of sensor IDs.
:param start_time: Start time (Unix timestamp in seconds).
:param end_time: End time (Unix timestamp in seconds).
:param fields: List of sensor fields to filter alerts.
:return: A dictionary representing the JSON response containing sensor alerts.
:raises ValueError: If device_ids is an empty list.
---
**Note:** This is a functional wrapper for its equivalent method in the SensorsClient. It creates a new client instance on every call, making it best for single, convenient operations. For making multiple API calls, instantiate and use a SensorsClient object directly for better performance.
"""
return _get_default_client().get_all_sensor_alerts(device_ids, start_time, end_time, fields)
[docs]
@typechecked
def get_all_sensor_data(device_id: str, start_time: Optional[int] = None, end_time: Optional[int] = None, fields: Optional[List[str]] = None, interval: Optional[str] = None):
"""
Returns sensor readings for a particular sensor over a specified time range.
:param device_id: The unique identifier of the sensor.
:param start_time: Start time for sensor data (Unix timestamp in seconds).
:param end_time: End time for sensor data (Unix timestamp in seconds).
:param fields: List of sensor fields to include in the response.
:param interval: The time interval for the requested sensor data.
Data is stored at 1-second intervals for 30 days and at 5-minute intervals
for data between 30 and 365 days old. Specify a number followed by a unit
(``s``, ``m``, or ``h``), e.g. ``"5m"`` for 5-minute intervals.
If omitted, a default resolution is calculated from the time range.
:return: A dictionary representing the JSON response containing sensor data.
---
**Note:** This is a functional wrapper for its equivalent method in the SensorsClient. It creates a new client instance on every call, making it best for single, convenient operations. For making multiple API calls, instantiate and use a SensorsClient object directly for better performance.
"""
return _get_default_client().get_all_sensor_data(device_id, start_time, end_time, fields, interval)
[docs]
@typechecked
def get_sensor_alerts(device_ids: List[str], start_time: Optional[int] = None, end_time: Optional[int] = None, page_size: Optional[int] = None, page_token: Optional[str] = None, fields: Optional[List[str]] = None):
"""
Returns all alerts for a set of sensors in an organization over a specified time range.
:param device_ids: List of sensor IDs.
:param start_time: Start time (Unix timestamp in seconds).
:param end_time: End time (Unix timestamp in seconds).
:param page_size: Number of items per page.
:param page_token: Token for pagination.
:param fields: List of sensor fields to filter alerts.
:return: A dictionary representing the JSON response containing sensor alerts.
:raises ValueError: If device_ids is an empty list.
---
**Note:** This is a functional wrapper for its equivalent method in the SensorsClient. It creates a new client instance on every call, making it best for single, convenient operations. For making multiple API calls, instantiate and use a SensorsClient object directly for better performance.
"""
return _get_default_client().get_sensor_alerts(device_ids, start_time, end_time, page_size, page_token, fields)
[docs]
@typechecked
def get_sensor_data(device_id: str, start_time: Optional[int] = None, end_time: Optional[int] = None, page_size: Optional[int] = None, page_token: Optional[str] = None, fields: Optional[List[str]] = None, interval: Optional[str] = None):
"""
Returns sensor readings for a particular sensor over a specified time range.
:param device_id: The unique identifier of the sensor.
:param start_time: Start time for sensor data (Unix timestamp in seconds).
:param end_time: End time for sensor data (Unix timestamp in seconds).
:param page_size: Number of items per page.
:param page_token: Token for pagination.
:param fields: List of sensor fields to include in the response.
:param interval: The time interval for the requested sensor data.
Data is stored at 1-second intervals for 30 days and at 5-minute intervals
for data between 30 and 365 days old. Specify a number followed by a unit
(``s``, ``m``, or ``h``), e.g. ``"5m"`` for 5-minute intervals.
If omitted, a default resolution is calculated from the time range.
:return: A dictionary representing the JSON response containing sensor data.
---
**Note:** This is a functional wrapper for its equivalent method in the SensorsClient. It creates a new client instance on every call, making it best for single, convenient operations. For making multiple API calls, instantiate and use a SensorsClient object directly for better performance.
"""
return _get_default_client().get_sensor_data(device_id, start_time, end_time, page_size, page_token, fields, interval)