API Reference¶

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

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}/"

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

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

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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",
    )

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

def get_projectsite_detail(self, projectsite: str, **kwargs: Any) -> dict[str, Any]:
    """Return a single project site lookup response."""

def get_assetlocation_detail(
    self,
    assetlocation: str,
    projectsite: str | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """Return a single asset location lookup response."""

def get_subassemblies(
    self,
    projectsite: str | None = None,
    assetlocation: str | None = None,
    subassembly_type: str | None = None,
    model_definition: str | None = None,
) -> dict[str, Any]:
    """Return a subassembly lookup response."""

def get_projectsite(self, projectsite: str, **kwargs: Any) -> LookupRecord:
    """Resolve a project site detail lookup."""
    result = self.locations_client.get_projectsite_detail(projectsite=projectsite, **kwargs)
    return self._build_record(
        result=result,
        label=f"project site '{projectsite}'",
        error_type=ProjectSiteLookupError,
    )

def get_assetlocation(
    self,
    assetlocation: str,
    projectsite: str | None = None,
    **kwargs: Any,
) -> LookupRecord:
    """Resolve an asset location detail lookup."""
    result = self.locations_client.get_assetlocation_detail(
        assetlocation=assetlocation,
        projectsite=projectsite,
        **kwargs,
    )
    label = f"asset location '{assetlocation}'"
    if projectsite is not None:
        label += f" in project site '{projectsite}'"
    return self._build_record(
        result=result,
        label=label,
        error_type=AssetLocationLookupError,
    )

def get_subassemblies(
    self,
    assetlocation: str,
    projectsite: str | None = None,
    **kwargs: Any,
) -> LookupRecord:
    """Resolve a subassembly lookup."""
    result = self.geometry_client.get_subassemblies(
        projectsite=projectsite,
        assetlocation=assetlocation,
        **kwargs,
    )
    label = f"subassemblies for asset location '{assetlocation}'"
    if projectsite is not None:
        label += f" in project site '{projectsite}'"
    return self._build_record(
        result=result,
        label=label,
        error_type=SubassembliesLookupError,
    )

def get_asset_context(
    self,
    projectsite: str | None,
    assetlocation: str,
    model_definition: str | None = None,
) -> AssetLookupContext:
    """Resolve the lookup context needed for an SHM asset workflow.

    Parameters
    ----------
    projectsite
        Parent SDK project site title. When omitted, the service
        derives it from the asset-location lookup data.
    assetlocation
        Parent SDK asset location title.
    model_definition
        Optional parent SDK model-definition title used when an asset has
        subassemblies for multiple model definitions.

    Returns
    -------
    AssetLookupContext
        Typed lookup records plus the resolved model definition.

    Examples
    --------
    >>> from unittest.mock import Mock
    >>> locations_client = Mock()
    >>> geometry_client = Mock()
    >>> locations_client.get_assetlocation_detail.return_value = {
    ...     "data": pd.DataFrame([{"id": 11, "projectsite_name": "Project A"}]),
    ...     "exists": True,
    ...     "id": 11,
    ... }
    >>> locations_client.get_projectsite_detail.return_value = {
    ...     "data": pd.DataFrame([{"id": 10, "title": "Project A"}]),
    ...     "exists": True,
    ...     "id": 10,
    ... }
    >>> geometry_client.get_subassemblies.return_value = {
    ...     "data": pd.DataFrame([{"id": 40, "subassembly_type": "TP", "model_definition": "MD-01"}]),
    ...     "exists": True,
    ... }
    >>> geometry_client.get_modeldefinition_id.return_value = {"id": 99}
    >>> service = ParentSDKLookupService(locations_client=locations_client, geometry_client=geometry_client)
    >>> context = service.get_asset_context(projectsite=None, assetlocation="Asset-01")
    >>> (context.site.record_id, context.asset.record_id, context.model_definition)
    (10, 11, 99)
    """
    asset = self.get_assetlocation(assetlocation=assetlocation, projectsite=projectsite)
    resolved_projectsite = projectsite or self._resolve_projectsite_name(asset, assetlocation)
    site = self.get_projectsite(projectsite=resolved_projectsite)
    subassembly_kwargs: dict[str, str] = {}
    if model_definition is not None:
        subassembly_kwargs["model_definition"] = model_definition
    subassemblies = self.get_subassemblies(
        assetlocation=assetlocation,
        projectsite=resolved_projectsite,
        **subassembly_kwargs,
    )
    model_definition = self.get_model_definition(
        subassemblies=subassemblies,
        assetlocation=assetlocation,
        projectsite=resolved_projectsite,
        model_definition=model_definition,
    )
    return AssetLookupContext(
        site=site,
        asset=asset,
        subassemblies=subassemblies,
        model_definition=model_definition,
    )

def get_signal_upload_context(
    self,
    projectsite: str | None,
    assetlocation: str,
    permission_group_ids: Sequence[int] | None = None,
    model_definition: str | None = None,
) -> SignalUploadContext:
    """Resolve the payload-builder context for SHM signal uploads.

    Parameters
    ----------
    projectsite
        Parent SDK project site title. When omitted, the service
        derives it from the asset-location lookup data.
    assetlocation
        Parent SDK asset location title.
    permission_group_ids
        Visibility groups applied to created SHM records.
    model_definition
        Optional parent SDK model-definition title used when an asset has
        subassemblies for multiple model definitions.

    Returns
    -------
    SignalUploadContext
        Upload context compatible with legacy payload builders.

    Examples
    --------
    >>> from unittest.mock import Mock
    >>> locations_client = Mock()
    >>> geometry_client = Mock()
    >>> locations_client.get_projectsite_detail.return_value = {
    ...     "data": pd.DataFrame([{"id": 10, "title": "Project A"}]),
    ...     "exists": True,
    ...     "id": 10,
    ... }
    >>> locations_client.get_assetlocation_detail.return_value = {
    ...     "data": pd.DataFrame([{"id": 11, "title": "Asset-01"}]),
    ...     "exists": True,
    ...     "id": 11,
    ... }
    >>> geometry_client.get_subassemblies.return_value = {
    ...     "data": pd.DataFrame(
    ...         [
    ...             {"id": 40, "subassembly_type": "TP", "model_definition": "MD-01"},
    ...             {"id": 41, "subassembly_type": "TW", "model_definition": "MD-01"},
    ...         ]
    ...     ),
    ...     "exists": True,
    ... }
    >>> service = ParentSDKLookupService(locations_client=locations_client, geometry_client=geometry_client)
    >>> context = service.get_signal_upload_context("Project A", "Asset-01", permission_group_ids=[7])
    >>> context.site_id, context.asset_location_id, context.subassembly_id_for("TP")
    (10, 11, 40)
    """
    asset_context = self.get_asset_context(
        projectsite=projectsite,
        assetlocation=assetlocation,
        model_definition=model_definition,
    )
    return self.build_signal_upload_context(
        asset_context=asset_context,
        permission_group_ids=permission_group_ids,
    )

@staticmethod
def build_signal_upload_context(
    asset_context: AssetLookupContext,
    permission_group_ids: Sequence[int] | None = None,
) -> SignalUploadContext:
    """Translate parent lookup records into upload payload ids.

    Parameters
    ----------
    asset_context
        Normalized parent SDK lookup context.
    permission_group_ids
        Visibility groups applied to created SHM records.

    Returns
    -------
    SignalUploadContext
        Upload context compatible with legacy payload builders.

    Raises
    ------
    SignalUploadContextError
        If required parent lookup ids or subassembly columns are missing.

    Examples
    --------
    >>> asset_context = AssetLookupContext(
    ...     site=LookupRecord(pd.DataFrame([{"id": 10}]), record_id=10),
    ...     asset=LookupRecord(pd.DataFrame([{"id": 11}]), record_id=11),
    ...     subassemblies=LookupRecord(
    ...         pd.DataFrame(
    ...             [
    ...                 {"id": 40, "subassembly_type": "TP", "model_definition": "MD-01"},
    ...                 {"id": 41, "subassembly_type": "TW", "model_definition": "MD-01"},
    ...             ]
    ...         )
    ...     ),
    ...     model_definition="MD-01",
    ... )
    >>> upload_context = ParentSDKLookupService.build_signal_upload_context(asset_context, [3, 5])
    >>> upload_context.permission_group_ids
    [3, 5]
    """
    if asset_context.site.record_id is None:
        raise SignalUploadContextError("Project site lookup did not provide a record id.")
    if asset_context.asset.record_id is None:
        raise SignalUploadContextError("Asset location lookup did not provide a record id.")

    return SignalUploadContext(
        site_id=int(asset_context.site.record_id),
        asset_location_id=int(asset_context.asset.record_id),
        model_definition_id=asset_context.model_definition,
        permission_group_ids=(list(permission_group_ids) if permission_group_ids is not None else None),
        subassembly_ids_by_type=ParentSDKLookupService._build_subassembly_ids_by_type(asset_context.subassemblies),
    )

def get_model_definition(
    self,
    subassemblies: LookupRecord,
    assetlocation: str,
    projectsite: str,
    model_definition: str | None = None,
) -> int | str:
    """Resolve the model definition reference used by SHM payload builders.

    The lookup uses an explicit model-definition title when provided.
    Otherwise it derives the title from transition-piece subassembly rows.
    When the parent geometry client exposes ``get_modeldefinition_id()``,
    it upgrades a model-definition title into the corresponding backend id.
    """
    if model_definition is None:
        model_definition = self.get_transition_piece_model_definition(subassemblies=subassemblies)
    if isinstance(model_definition, int):
        return model_definition

    get_modeldefinition_id = getattr(self.geometry_client, "get_modeldefinition_id", None)
    if not callable(get_modeldefinition_id):
        return model_definition

    try:
        result = get_modeldefinition_id(
            assetlocation=assetlocation,
            projectsite=projectsite,
            model_definition=model_definition,
        )
    except ValueError as exc:
        raise ModelDefinitionLookupError(str(exc)) from exc

    if not isinstance(result, Mapping):
        return model_definition

    record_id = result.get("id")
    normalized_record_id = self._normalize_model_definition(record_id)
    if normalized_record_id is None:
        return model_definition

    if isinstance(normalized_record_id, int):
        return normalized_record_id

    try:
        return int(normalized_record_id)
    except (TypeError, ValueError):
        return model_definition