SHM API Transport¶

io ¶

API client for the shm extension.

This module exposes :class:ShmAPI as the low-level entry point for SHM transport and persistence helpers.

Examples:

>>> from owi.metadatabase.shm import ShmAPI
>>> isinstance(ShmAPI(token="dummy"), ShmAPI)
True

Classes¶

ShmEndpoints `dataclass` ¶

ShmEndpoints(
    api_subdir="/shm/routes/",
    sensor_type="sensortype",
    sensor="sensor",
    sensor_calibration="sensorcalibration",
    signal="signal",
    signal_history="signalhistory",
    signal_calibration="signalcalibration",
    derived_signal="derivedsignal",
    derived_signal_history="derivedsignalhistory",
    derived_signal_calibration="derivedsignalcalibration",
)

Centralized route names for the SHM backend.

Functions¶

mutation_path ¶

mutation_path(endpoint)

Return a collection endpoint path with a trailing slash.

Source code in src/owi/metadatabase/shm/io.py

def mutation_path(self, endpoint: str) -> str:
    """Return a collection endpoint path with a trailing slash."""
    return endpoint.rstrip("/") + "/"

detail_path ¶

detail_path(endpoint, object_id)

Return a detail endpoint path with a trailing slash.

Source code in src/owi/metadatabase/shm/io.py

def detail_path(self, endpoint: str, object_id: int) -> str:
    """Return a detail endpoint path with a trailing slash."""
    return f"{endpoint.rstrip('/')}/{object_id}/"

ShmAPI ¶

ShmAPI(api_subdir=api_subdir, **kwargs)

Bases: API

Low-level API client for the SHM extension.

Parameters:

Name	Type	Description	Default
`api_subdir`	`str`	API sub-path appended to the base root.	`"/shm/routes/"`
`**kwargs`	`Any`	Forwarded to :class:`owi.metadatabase.io.API`.	`{}`

Examples:

>>> api = ShmAPI(token="dummy")
>>> api.ping()
'ok'

Source code in src/owi/metadatabase/shm/io.py

def __init__(self, api_subdir: str = DEFAULT_SHM_ENDPOINTS.api_subdir, **kwargs: Any) -> None:
    self.endpoints: ShmEndpoints = kwargs.pop("endpoints", DEFAULT_SHM_ENDPOINTS)
    super().__init__(**kwargs)
    self.base_api_root = self.api_root
    self.api_root = self.api_root + api_subdir

Functions¶

ping ¶

ping()

Return a basic health response.

Examples:

>>> api = ShmAPI(token="dummy")
>>> api.ping()
'ok'

Source code in src/owi/metadatabase/shm/io.py

def ping(self) -> str:
    """Return a basic health response.

    Examples
    --------
    >>> api = ShmAPI(token="dummy")
    >>> api.ping()
    'ok'
    """
    return "ok"

get_signal ¶

get_signal(signal_id, **kwargs)

Return a single SHM signal by its backend signal identifier.

Parameters:

Name	Type	Description	Default
`signal_id`	`str`	Backend-facing SHM signal identifier.	required
`**kwargs`	`QueryValue`	Additional query parameters forwarded to the SHM route.	`{}`

Returns:

Type	Description
`dict[str, Any]`	Parent-SDK-style result dictionary containing `data`, `exists`, `id`, and `response`.

Examples:

>>> from unittest.mock import patch
>>> api = ShmAPI(token="dummy")
>>> with patch.object(
...     ShmAPI,
...     "process_data",
...     return_value=(pd.DataFrame([{"id": 7, "signal_id": "SG-01"}]), {"existance": True, "id": 7}),
... ):
...     result = api.get_signal("SG-01")
>>> result["id"]
7

Source code in src/owi/metadatabase/shm/io.py

def get_signal(self, signal_id: str, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM signal by its backend signal identifier.

    Parameters
    ----------
    signal_id
        Backend-facing SHM signal identifier.
    **kwargs
        Additional query parameters forwarded to the SHM route.

    Returns
    -------
    dict[str, Any]
        Parent-SDK-style result dictionary containing ``data``,
        ``exists``, ``id``, and ``response``.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> api = ShmAPI(token="dummy")
    >>> with patch.object(
    ...     ShmAPI,
    ...     "process_data",
    ...     return_value=(pd.DataFrame([{"id": 7, "signal_id": "SG-01"}]), {"existance": True, "id": 7}),
    ... ):
    ...     result = api.get_signal("SG-01")
    >>> result["id"]
    7
    """
    return self._get_resource(self.endpoints.signal, signal_id=signal_id, **kwargs)

get_sensor_type ¶

get_sensor_type(**kwargs)

Return a single SHM sensor type by query parameters.

Parameters:

Name	Type	Description	Default
`**kwargs`	`QueryValue`	Query parameters forwarded to the SHM sensor-type route.	`{}`

Returns:

Type	Description
`dict[str, Any]`	Parent-SDK-style result dictionary containing `data`, `exists`, `id`, and `response`.

Source code in src/owi/metadatabase/shm/io.py

def get_sensor_type(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM sensor type by query parameters.

    Parameters
    ----------
    **kwargs
        Query parameters forwarded to the SHM sensor-type route.

    Returns
    -------
    dict[str, Any]
        Parent-SDK-style result dictionary containing ``data``,
        ``exists``, ``id``, and ``response``.
    """
    return self._get_resource(self.endpoints.sensor_type, **kwargs)

get_sensor ¶

get_sensor(**kwargs)

Return a single SHM sensor by query parameters.

Parameters:

Name	Type	Description	Default
`**kwargs`	`QueryValue`	Query parameters forwarded to the SHM sensor route.	`{}`

Returns:

Type	Description
`dict[str, Any]`	Parent-SDK-style result dictionary containing `data`, `exists`, `id`, and `response`.

Source code in src/owi/metadatabase/shm/io.py

def get_sensor(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM sensor by query parameters.

    Parameters
    ----------
    **kwargs
        Query parameters forwarded to the SHM sensor route.

    Returns
    -------
    dict[str, Any]
        Parent-SDK-style result dictionary containing ``data``,
        ``exists``, ``id``, and ``response``.
    """
    return self._get_resource(self.endpoints.sensor, **kwargs)

get_sensor_calibration ¶

get_sensor_calibration(**kwargs)

Return a single SHM sensor calibration by query parameters.

Source code in src/owi/metadatabase/shm/io.py

def get_sensor_calibration(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM sensor calibration by query parameters."""
    return self._get_resource(self.endpoints.sensor_calibration, **kwargs)

get_signal_history ¶

get_signal_history(**kwargs)

Return a single SHM signal history row by query parameters.

Source code in src/owi/metadatabase/shm/io.py

def get_signal_history(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM signal history row by query parameters."""
    return self._get_resource(self.endpoints.signal_history, **kwargs)

get_signal_calibration ¶

get_signal_calibration(**kwargs)

Return a single SHM signal calibration by query parameters.

Source code in src/owi/metadatabase/shm/io.py

def get_signal_calibration(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM signal calibration by query parameters."""
    return self._get_resource(self.endpoints.signal_calibration, **kwargs)

get_derived_signal ¶

get_derived_signal(**kwargs)

Return a single SHM derived signal by query parameters.

Source code in src/owi/metadatabase/shm/io.py

def get_derived_signal(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM derived signal by query parameters."""
    return self._get_resource(self.endpoints.derived_signal, **kwargs)

get_derived_signal_history ¶

get_derived_signal_history(**kwargs)

Return a single SHM derived signal history row by query parameters.

Source code in src/owi/metadatabase/shm/io.py

def get_derived_signal_history(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM derived signal history row by query parameters."""
    return self._get_resource(self.endpoints.derived_signal_history, **kwargs)

get_derived_signal_calibration ¶

get_derived_signal_calibration(**kwargs)

Return a single SHM derived signal calibration by query parameters.

Source code in src/owi/metadatabase/shm/io.py

def get_derived_signal_calibration(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return a single SHM derived signal calibration by query parameters."""
    return self._get_resource(self.endpoints.derived_signal_calibration, **kwargs)

create_signal ¶

create_signal(payload)

Create a signal record.

Examples:

>>> from unittest.mock import patch
>>> api = ShmAPI(token="dummy")
>>> with patch.object(ShmAPI, "_mutate_resource", return_value={"id": 12, "exists": True}) as mocker:
...     result = api.create_signal({"signal_id": "SG-01"})
>>> mocker.assert_called_once_with(api.endpoints.signal, {"signal_id": "SG-01"})
>>> result["id"]
12

Source code in src/owi/metadatabase/shm/io.py

def create_signal(self, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Create a signal record.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> api = ShmAPI(token="dummy")
    >>> with patch.object(ShmAPI, "_mutate_resource", return_value={"id": 12, "exists": True}) as mocker:
    ...     result = api.create_signal({"signal_id": "SG-01"})
    >>> mocker.assert_called_once_with(api.endpoints.signal, {"signal_id": "SG-01"})
    >>> result["id"]
    12
    """
    return self._mutate_resource(self.endpoints.signal, payload)

create_signal_history ¶

create_signal_history(payload)

Create a signal history record.

Source code in src/owi/metadatabase/shm/io.py

def create_signal_history(self, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Create a signal history record."""
    return self._mutate_resource(self.endpoints.signal_history, payload)

create_signal_calibration ¶

create_signal_calibration(payload)

Create a signal calibration record.

Source code in src/owi/metadatabase/shm/io.py

def create_signal_calibration(self, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Create a signal calibration record."""
    return self._mutate_resource(self.endpoints.signal_calibration, payload)

create_derived_signal ¶

create_derived_signal(payload)

Create a derived signal record.

Source code in src/owi/metadatabase/shm/io.py

def create_derived_signal(self, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Create a derived signal record."""
    return self._mutate_resource(self.endpoints.derived_signal, payload)

create_derived_signal_history ¶

create_derived_signal_history(payload)

Create a derived signal history record.

Source code in src/owi/metadatabase/shm/io.py

def create_derived_signal_history(self, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Create a derived signal history record."""
    return self._mutate_resource(self.endpoints.derived_signal_history, payload)

patch_derived_signal_history ¶

patch_derived_signal_history(history_id, payload)

Patch a derived signal history record by id.

Source code in src/owi/metadatabase/shm/io.py

def patch_derived_signal_history(self, history_id: int, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Patch a derived signal history record by id."""
    return self._mutate_resource(
        self.endpoints.derived_signal_history,
        payload,
        object_id=history_id,
        method="patch",
    )

create_derived_signal_calibration ¶

create_derived_signal_calibration(payload)

Create a derived signal calibration record.

Source code in src/owi/metadatabase/shm/io.py

def create_derived_signal_calibration(self, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Create a derived signal calibration record."""
    return self._mutate_resource(self.endpoints.derived_signal_calibration, payload)

list_sensor_types ¶

list_sensor_types(**kwargs)

Return all SHM sensor types matching the query parameters.

Parameters:

Name	Type	Description	Default
`**kwargs`	`QueryValue`	Query parameters forwarded to the sensor-type list route.	`{}`

Returns:

Type	Description
`dict[str, Any]`	Parent-SDK-style result dictionary containing `data`, `exists`, and `response`.

Source code in src/owi/metadatabase/shm/io.py

def list_sensor_types(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM sensor types matching the query parameters.

    Parameters
    ----------
    **kwargs
        Query parameters forwarded to the sensor-type list route.

    Returns
    -------
    dict[str, Any]
        Parent-SDK-style result dictionary containing ``data``,
        ``exists``, and ``response``.
    """
    return self._list_resource(self.endpoints.sensor_type, **kwargs)

list_sensors ¶

list_sensors(**kwargs)

Return all SHM sensors matching the query parameters.

Parameters:

Name	Type	Description	Default
`**kwargs`	`QueryValue`	Query parameters forwarded to the sensor list route.	`{}`

Returns:

Type	Description
`dict[str, Any]`	Parent-SDK-style result dictionary containing `data`, `exists`, and `response`.

Source code in src/owi/metadatabase/shm/io.py

def list_sensors(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM sensors matching the query parameters.

    Parameters
    ----------
    **kwargs
        Query parameters forwarded to the sensor list route.

    Returns
    -------
    dict[str, Any]
        Parent-SDK-style result dictionary containing ``data``,
        ``exists``, and ``response``.
    """
    return self._list_resource(self.endpoints.sensor, **kwargs)

list_sensor_calibrations ¶

list_sensor_calibrations(**kwargs)

Return all SHM sensor calibrations matching the query parameters.

Source code in src/owi/metadatabase/shm/io.py

def list_sensor_calibrations(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM sensor calibrations matching the query parameters."""
    return self._list_resource(self.endpoints.sensor_calibration, **kwargs)

list_signals ¶

list_signals(**kwargs)

Return all SHM signals matching the query parameters.

Source code in src/owi/metadatabase/shm/io.py

def list_signals(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM signals matching the query parameters."""
    return self._list_resource(self.endpoints.signal, **kwargs)

list_signal_history ¶

list_signal_history(**kwargs)

Return all SHM signal history rows matching the query parameters.

Source code in src/owi/metadatabase/shm/io.py

def list_signal_history(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM signal history rows matching the query parameters."""
    return self._list_resource(self.endpoints.signal_history, **kwargs)

list_signal_calibrations ¶

list_signal_calibrations(**kwargs)

Return all SHM signal calibrations matching the query parameters.

Source code in src/owi/metadatabase/shm/io.py

def list_signal_calibrations(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM signal calibrations matching the query parameters."""
    return self._list_resource(self.endpoints.signal_calibration, **kwargs)

list_derived_signals ¶

list_derived_signals(**kwargs)

Return all SHM derived signals matching the query parameters.

Source code in src/owi/metadatabase/shm/io.py

def list_derived_signals(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM derived signals matching the query parameters."""
    return self._list_resource(self.endpoints.derived_signal, **kwargs)

list_derived_signal_history ¶

list_derived_signal_history(**kwargs)

Return all SHM derived signal history rows matching the query parameters.

Source code in src/owi/metadatabase/shm/io.py

def list_derived_signal_history(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM derived signal history rows matching the query parameters."""
    return self._list_resource(self.endpoints.derived_signal_history, **kwargs)

list_derived_signal_calibrations ¶

list_derived_signal_calibrations(**kwargs)

Return all SHM derived signal calibrations matching the query parameters.

Source code in src/owi/metadatabase/shm/io.py

def list_derived_signal_calibrations(self, **kwargs: QueryValue) -> dict[str, Any]:
    """Return all SHM derived signal calibrations matching the query parameters."""
    return self._list_resource(self.endpoints.derived_signal_calibration, **kwargs)

create_sensor_type ¶

create_sensor_type(payload, files=None)

Create a sensor type record, optionally with an image attachment.

Parameters:

Name	Type	Description	Default
`payload`	`Mapping[str, Any]`	Form fields for the sensor type resource.	required
`files`	`Mapping[str, Any] \| None`	Optional file mapping (e.g. `{"photo": open_file}`).	`None`

Returns:

Type	Description
`dict[str, Any]`	Parent-SDK-style result dictionary.

Source code in src/owi/metadatabase/shm/io.py

def create_sensor_type(
    self,
    payload: Mapping[str, Any],
    files: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Create a sensor type record, optionally with an image attachment.

    Parameters
    ----------
    payload
        Form fields for the sensor type resource.
    files
        Optional file mapping (e.g. ``{"photo": open_file}``).

    Returns
    -------
    dict[str, Any]
        Parent-SDK-style result dictionary.
    """
    if files:
        return self._mutate_multipart_resource(self.endpoints.sensor_type, payload, files=files)
    return self._mutate_resource(self.endpoints.sensor_type, payload)

create_sensor ¶

create_sensor(payload)

Create a sensor record.

Parameters:

Name	Type	Description	Default
`payload`	`Mapping[str, Any]`	JSON payload for the sensor resource.	required

Returns:

Type	Description
`dict[str, Any]`	Parent-SDK-style result dictionary.

Source code in src/owi/metadatabase/shm/io.py

def create_sensor(self, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Create a sensor record.

    Parameters
    ----------
    payload
        JSON payload for the sensor resource.

    Returns
    -------
    dict[str, Any]
        Parent-SDK-style result dictionary.
    """
    return self._mutate_resource(self.endpoints.sensor, payload)

create_sensor_calibration ¶

create_sensor_calibration(payload, files=None)

Create a sensor calibration record, optionally with a PDF attachment.

Parameters:

Name	Type	Description	Default
`payload`	`Mapping[str, Any]`	Form fields for the sensor calibration resource.	required
`files`	`Mapping[str, Any] \| None`	Optional file mapping (e.g. `{"datasheet": open_file}`).	`None`

Returns:

Type	Description
`dict[str, Any]`	Parent-SDK-style result dictionary.

Source code in src/owi/metadatabase/shm/io.py

def create_sensor_calibration(
    self,
    payload: Mapping[str, Any],
    files: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Create a sensor calibration record, optionally with a PDF attachment.

    Parameters
    ----------
    payload
        Form fields for the sensor calibration resource.
    files
        Optional file mapping (e.g. ``{"datasheet": open_file}``).

    Returns
    -------
    dict[str, Any]
        Parent-SDK-style result dictionary.
    """
    if files:
        return self._mutate_multipart_resource(self.endpoints.sensor_calibration, payload, files=files)
    return self._mutate_resource(self.endpoints.sensor_calibration, payload)

SHM API Transport¶

io ¶

Classes¶

ShmEndpoints dataclass ¶

Functions¶

mutation_path ¶

detail_path ¶

ShmAPI ¶

Functions¶

ping ¶

get_signal ¶

get_sensor_type ¶

get_sensor ¶

get_sensor_calibration ¶

get_signal_history ¶

get_signal_calibration ¶

get_derived_signal ¶

get_derived_signal_history ¶

get_derived_signal_calibration ¶

create_signal ¶

create_signal_history ¶

create_signal_calibration ¶

create_derived_signal ¶

create_derived_signal_history ¶

patch_derived_signal_history ¶

create_derived_signal_calibration ¶

list_sensor_types ¶

list_sensors ¶

list_sensor_calibrations ¶

list_signals ¶

list_signal_history ¶

list_signal_calibrations ¶

list_derived_signals ¶

list_derived_signal_history ¶

list_derived_signal_calibrations ¶

create_sensor_type ¶

create_sensor ¶

create_sensor_calibration ¶

ShmEndpoints `dataclass` ¶