Upload Orchestration¶

@classmethod
def from_processing_result(
    cls,
    *,
    projectsite: str,
    assetlocation: str,
    processing_result: SignalProcessingResult,
    permission_group_ids: Sequence[int] | None = None,
    model_definition: str | None = None,
    sensor_serial_numbers_by_signal: Mapping[str, int] | None = None,
    temperature_compensation_signal_ids: TemperatureCompensationSignalIDMap | None = None,
    temperature_compensation_signal_refs: TemperatureCompensationSignalRefMap | None = None,
) -> AssetSignalUploadRequest:
    """Build an upload request from a processed signal-config result.

    Parameters
    ----------
    projectsite
        Parent SDK project site title.
    assetlocation
        Parent SDK asset location title.
    processing_result
        Processed signal and derived-signal records emitted by a processor.
    permission_group_ids
        Visibility groups applied to created SHM objects.
    model_definition
        Optional parent SDK model-definition title used when an asset has
        subassemblies for multiple model definitions.
    sensor_serial_numbers_by_signal
        Optional map from signal identifier to backend sensor serial
        number used for signal history rows.
    temperature_compensation_signal_ids
        Optional map from legacy temperature-compensation sensor token to
        backend SHM signal id.
    temperature_compensation_signal_refs
        Optional map from legacy temperature-compensation sensor token to
        SHM signal identifier.

    Returns
    -------
    AssetSignalUploadRequest
        Asset-scoped upload request that preserves the archive-compatible
        payload shape.

    Examples
    --------
    >>> from owi.metadatabase.shm.processing import ProcessedSignalRecord, SignalProcessingResult
    >>> signal = ProcessedSignalRecord()
    >>> signal.add_status("01/01/1972 00:00", "ok")
    >>> request = AssetSignalUploadRequest.from_processing_result(
    ...     projectsite="Project A",
    ...     assetlocation="Asset-01",
    ...     processing_result=SignalProcessingResult(signals={"SIG": signal}, derived_signals={}),
    ... )
    >>> request.signals["SIG"]["status"][0]["status"]
    'ok'
    """
    signals, derived_signals = processing_result.to_legacy_data()
    return cls(
        projectsite=projectsite,
        assetlocation=assetlocation,
        signals=signals,
        derived_signals=derived_signals or None,
        permission_group_ids=permission_group_ids,
        model_definition=model_definition,
        sensor_serial_numbers_by_signal=sensor_serial_numbers_by_signal,
        temperature_compensation_signal_ids=temperature_compensation_signal_ids,
        temperature_compensation_signal_refs=temperature_compensation_signal_refs,
    )

def get_sensor_type(self, **kwargs: Any) -> dict[str, Any]:
    """Resolve one SHM sensor type record."""
    ...

def get_sensor(self, **kwargs: Any) -> dict[str, Any]:
    """Resolve one SHM sensor record."""
    ...

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

def get_signal(self, signal_id: str, **kwargs: Any) -> dict[str, Any]:
    """Resolve a signal record by backend identifier."""
    ...

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

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

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

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

def patch_derived_signal_history(
    self,
    history_id: int,
    payload: Mapping[str, Any],
) -> dict[str, Any]:
    """Patch a derived signal history record."""
    ...

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

def signals_process_data(self) -> None:
    """Populate turbine-scoped signal dictionaries."""
    ...

def get_sensor_type(self, **kwargs: Any) -> dict[str, Any]:
    """Resolve one SHM sensor type record."""
    ...

def get_sensor(self, **kwargs: Any) -> dict[str, Any]:
    """Resolve one SHM sensor record."""
    ...

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

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

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

def upload_sensor_types(
    self,
    sensor_types_data: Sequence[Mapping[str, Any]],
    permission_group_ids: Sequence[int] | None,
    path_to_images: str | Path | None = None,
) -> list[dict[str, Any]]:
    """Upload sensor type records, optionally with image attachments.

    Parameters
    ----------
    sensor_types_data
        List of sensor type records (e.g. loaded from ``sensor_types.json``).
    permission_group_ids
        Permission groups applied to every sensor type.
    path_to_images
        Optional directory containing sensor type image files.

    Returns
    -------
    list[dict[str, Any]]
        Raw backend responses for each created sensor type.
    """
    payloads = build_sensor_type_payloads(
        sensor_types_data,
        visibility_groups=permission_group_ids,
        path_to_images=path_to_images,
    )
    return [self._upload_sensor_type(payload) for payload in payloads]

def upload_sensors(
    self,
    sensor_type_name: str,
    sensor_type_params: Mapping[str, str],
    sensors_data: SensorsDataByTurbine,
    permission_group_ids: Sequence[int] | None,
    turbines: Sequence[str] | None = None,
) -> list[dict[str, Any]]:
    """Upload sensor records for a single sensor category across turbines.

    Parameters
    ----------
    sensor_type_name
        Key identifying the sensor category within each turbine's data
        (e.g. ``"accelerometers"``).
    sensor_type_params
        Query parameters used to resolve the backend sensor type id
        (e.g. ``{"name": "393B04"}``).
    sensors_data
        Per-turbine sensor data keyed by turbine identifier. Each turbine
        has categories mapping to ``{"serial_numbers": [...], "cabinets": [...]}``.
    permission_group_ids
        Permission groups applied to every sensor.
    turbines
        Optional filter to upload only specific turbines. When *None*,
        all turbines in ``sensors_data`` are processed.

    Returns
    -------
    list[dict[str, Any]]
        Raw backend responses for each created sensor.
    """
    sensor_type_result = self.shm_api.get_sensor_type(**dict(sensor_type_params))
    sensor_type_id = self._require_existing_result_id(
        sensor_type_result,
        label=f"sensor type '{sensor_type_name}'",
    )

    serial_numbers: list[str | None] = []
    cabinets: list[str | int | None] = []
    turbine_keys = turbines if turbines is not None else list(sensors_data.keys())

    for turbine in turbine_keys:
        data_turbine = sensors_data.get(turbine)
        self._collect_sensor_columns(
            data_turbine,
            sensor_type_name,
            serial_numbers,
            cabinets,
            turbine,
        )

    if not serial_numbers and not cabinets:
        return []

    payloads = build_sensor_payloads(
        sensor_type_id=sensor_type_id,
        serial_numbers=serial_numbers,
        cabinets=cabinets,
        visibility_groups=permission_group_ids,
    )
    return [self.shm_api.create_sensor(p.to_payload()) for p in payloads]

def upload_sensor_calibrations(
    self,
    signal_sensor_map_data: Mapping[str, Mapping[str, Mapping[str, Any]]],
    signal_calibration_map_data: Mapping[str, Mapping[str, Mapping[str, str]]],
    path_to_datasheets: str | Path,
    turbines: Sequence[str] | None = None,
) -> list[dict[str, Any]]:
    """Upload sensor calibration records with optional PDF attachments.

    Parameters
    ----------
    signal_sensor_map_data
        Per-turbine signal-to-sensor mapping (keyed by turbine, then signal
        name, with sensor lookup params including ``sensor_type_id``).
    signal_calibration_map_data
        Per-turbine calibration data (keyed by turbine, then signal name,
        with ``date`` and ``filename`` fields).
    path_to_datasheets
        Directory containing calibration PDF files.
    turbines
        Optional turbine filter. When *None*, all turbines are processed.

    Returns
    -------
    list[dict[str, Any]]
        Raw backend responses for each created calibration.
    """
    results: list[dict[str, Any]] = []
    turbine_keys = turbines if turbines is not None else list(signal_sensor_map_data.keys())

    for turbine in turbine_keys:
        turbine_ss_map = signal_sensor_map_data.get(turbine)
        turbine_sc_map = signal_calibration_map_data.get(turbine)
        if turbine_ss_map is None:
            continue

        resolved_sensor_ids = self._resolve_sensor_ids_for_turbine(turbine_ss_map, turbine)

        if turbine_sc_map is not None:
            payloads = build_sensor_calibration_payloads(
                signal_sensor_map=resolved_sensor_ids,
                signal_calibration_map=turbine_sc_map,
                path_to_datasheets=path_to_datasheets,
            )
            for payload in payloads:
                results.append(self._upload_sensor_calibration(payload))

    return results

@classmethod
def from_clients(
    cls,
    shm_api: ShmSignalUploadClient,
    locations_client: ParentLocationsLookupClient,
    geometry_client: ParentGeometryLookupClient,
) -> ShmSignalUploader:
    """Construct the uploader from SHM and parent SDK clients.

    Parameters
    ----------
    shm_api
        SHM transport client used for backend mutations.
    locations_client
        Parent SDK client that resolves project and asset locations.
    geometry_client
        Parent SDK client that resolves subassemblies and model
        definitions.

    Returns
    -------
    ShmSignalUploader
        Uploader wired to the canonical SHM lookup service.
    """
    return cls(
        shm_api=shm_api,
        lookup_service=ParentSDKLookupService(
            locations_client=locations_client,
            geometry_client=geometry_client,
        ),
    )

def upload_asset(self, request: AssetSignalUploadRequest) -> AssetSignalUploadResult:
    """Upload main and secondary SHM records for one asset.

    Parameters
    ----------
    request
        Asset-scoped upload request containing the archive-compatible main
        and derived signal mappings.

    Returns
    -------
    AssetSignalUploadResult
        Created backend ids plus raw backend responses grouped by upload
        phase.
    """
    lookup_kwargs: dict[str, Any] = {
        "projectsite": request.projectsite,
        "assetlocation": request.assetlocation,
        "permission_group_ids": request.permission_group_ids,
    }
    if request.model_definition is not None:
        lookup_kwargs["model_definition"] = request.model_definition
    upload_context = self.lookup_service.get_signal_upload_context(**lookup_kwargs)
    signal_ids_by_name, results_main = self._upload_main_signals(
        request.signals,
        upload_context,
    )
    temperature_compensation_signal_ids = self._resolve_temperature_compensation_signal_ids(
        explicit_ids=request.temperature_compensation_signal_ids,
        refs=request.temperature_compensation_signal_refs,
        signal_ids_by_name=signal_ids_by_name,
        asset_key=request.result_key,
    )
    results_secondary = self._upload_signal_secondary_data(
        request.signals,
        signal_ids_by_name=signal_ids_by_name,
        sensor_serial_numbers_by_signal=request.sensor_serial_numbers_by_signal,
        temperature_compensation_signal_ids=temperature_compensation_signal_ids,
    )

    derived_signal_ids_by_name: dict[str, int] = {}
    results_derived_main: list[dict[str, Any]] = []
    results_derived_secondary: list[dict[str, Any]] = []
    if request.derived_signals:
        derived_signal_ids_by_name, results_derived_main = self._upload_main_derived_signals(
            request.derived_signals,
            upload_context,
        )
        results_derived_secondary = self._upload_derived_signal_secondary_data(
            request.derived_signals,
            signal_ids_by_name=signal_ids_by_name,
            derived_signal_ids_by_name=derived_signal_ids_by_name,
        )

    return AssetSignalUploadResult(
        asset_key=request.result_key,
        signal_ids_by_name=signal_ids_by_name,
        derived_signal_ids_by_name=derived_signal_ids_by_name,
        results_main=results_main,
        results_secondary=results_secondary,
        results_derived_main=results_derived_main,
        results_derived_secondary=results_derived_secondary,
    )

def upload_assets(
    self,
    requests: Sequence[AssetSignalUploadRequest],
) -> dict[str, AssetSignalUploadResult]:
    """Upload SHM signal data for multiple assets.

    Parameters
    ----------
    requests
        Asset-scoped upload requests to execute in order.

    Returns
    -------
    dict[str, AssetSignalUploadResult]
        Upload results keyed by each request's stable result key.
    """
    return {request.result_key: self.upload_asset(request) for request in requests}

def upload_turbines(
    self,
    *,
    projectsite: str,
    signals_by_turbine: SignalConfigMapByTurbine,
    derived_signals_by_turbine: SignalConfigMapByTurbine | None = None,
    assetlocations_by_turbine: Mapping[str, str] | None = None,
    permission_group_ids: Sequence[int] | None = None,
    model_definition: str | None = None,
    sensor_serial_numbers_by_turbine: Mapping[str, Mapping[str, int]] | None = None,
    temperature_compensation_signal_ids_by_turbine: Mapping[str, TemperatureCompensationSignalIDMap] | None = None,
    temperature_compensation_signal_refs_by_turbine: (
        Mapping[str, TemperatureCompensationSignalRefMap] | None
    ) = None,
) -> dict[str, AssetSignalUploadResult]:
    """Upload SHM signal data for multiple turbine-scoped config bundles.

    Parameters
    ----------
    projectsite
        Parent SDK project site title shared by the turbine batch.
    signals_by_turbine
        Main signal mappings keyed by turbine identifier.
    derived_signals_by_turbine
        Optional derived signal mappings keyed by turbine identifier.
    assetlocations_by_turbine
        Optional turbine-to-asset-location override mapping.
    permission_group_ids
        Visibility groups applied to created SHM objects.
    model_definition
        Optional parent SDK model-definition title used when the project
        has multiple model definitions for the same asset location.
    sensor_serial_numbers_by_turbine
        Optional per-turbine mapping of signal identifiers to sensor serial
        numbers used for signal history rows.
    temperature_compensation_signal_ids_by_turbine
        Optional per-turbine mapping of temperature-compensation tokens to
        backend SHM signal ids.
    temperature_compensation_signal_refs_by_turbine
        Optional per-turbine mapping of temperature-compensation tokens to
        SHM signal identifiers resolved during upload.

    Returns
    -------
    dict[str, AssetSignalUploadResult]
        Upload results keyed by turbine identifier.

    This keeps the response keyed by turbine while parent lookups use the
    corresponding asset-location title.
    """
    results: dict[str, AssetSignalUploadResult] = {}
    for turbine, signals in signals_by_turbine.items():
        assetlocation = turbine
        if assetlocations_by_turbine is not None:
            assetlocation = assetlocations_by_turbine.get(turbine, turbine)

        derived_signals = None
        if derived_signals_by_turbine is not None:
            derived_signals = derived_signals_by_turbine.get(turbine)

        sensor_serial_numbers_by_signal = None
        if sensor_serial_numbers_by_turbine is not None:
            sensor_serial_numbers_by_signal = sensor_serial_numbers_by_turbine.get(turbine)

        temperature_compensation_signal_ids = None
        if temperature_compensation_signal_ids_by_turbine is not None:
            temperature_compensation_signal_ids = temperature_compensation_signal_ids_by_turbine.get(turbine)

        temperature_compensation_signal_refs = None
        if temperature_compensation_signal_refs_by_turbine is not None:
            temperature_compensation_signal_refs = temperature_compensation_signal_refs_by_turbine.get(turbine)

        results[turbine] = self.upload_asset(
            AssetSignalUploadRequest(
                projectsite=projectsite,
                assetlocation=assetlocation,
                signals=signals,
                derived_signals=derived_signals,
                permission_group_ids=permission_group_ids,
                model_definition=model_definition,
                sensor_serial_numbers_by_signal=sensor_serial_numbers_by_signal,
                temperature_compensation_signal_ids=temperature_compensation_signal_ids,
                temperature_compensation_signal_refs=temperature_compensation_signal_refs,
            )
        )
    return results

def upload_from_processor(
    self,
    *,
    projectsite: str,
    processor: SignalConfigUploadSource,
    assetlocations_by_turbine: Mapping[str, str] | None = None,
    permission_group_ids: Sequence[int] | None = None,
    model_definition: str | None = None,
    sensor_serial_numbers_by_turbine: Mapping[str, Mapping[str, int]] | None = None,
    temperature_compensation_signal_ids_by_turbine: Mapping[str, TemperatureCompensationSignalIDMap] | None = None,
    temperature_compensation_signal_refs_by_turbine: (
        Mapping[str, TemperatureCompensationSignalRefMap] | None
    ) = None,
) -> dict[str, AssetSignalUploadResult]:
    """Process turbine configs and upload them through the generic SHM seam.

    Parameters
    ----------
    projectsite
        Parent SDK project site title shared by the processor output.
    processor
        Processor instance that populates ``signals_data`` and
        ``signals_derived_data``.
    assetlocations_by_turbine
        Optional turbine-to-asset-location override mapping.
    permission_group_ids
        Visibility groups applied to created SHM objects.
    model_definition
        Optional parent SDK model-definition title used when the project
        has multiple model definitions for the same asset location.
    sensor_serial_numbers_by_turbine
        Optional per-turbine mapping of signal identifiers to sensor serial
        numbers used for signal history rows.
    temperature_compensation_signal_ids_by_turbine
        Optional per-turbine mapping of temperature-compensation tokens to
        backend SHM signal ids.
    temperature_compensation_signal_refs_by_turbine
        Optional per-turbine mapping of temperature-compensation tokens to
        SHM signal identifiers resolved during upload.

    Returns
    -------
    dict[str, AssetSignalUploadResult]
        Upload results keyed by turbine identifier.
    """
    processor.signals_process_data()
    return self.upload_turbines(
        projectsite=projectsite,
        signals_by_turbine=processor.signals_data,
        derived_signals_by_turbine=processor.signals_derived_data,
        assetlocations_by_turbine=assetlocations_by_turbine,
        permission_group_ids=permission_group_ids,
        model_definition=model_definition,
        sensor_serial_numbers_by_turbine=sensor_serial_numbers_by_turbine,
        temperature_compensation_signal_ids_by_turbine=(temperature_compensation_signal_ids_by_turbine),
        temperature_compensation_signal_refs_by_turbine=(temperature_compensation_signal_refs_by_turbine),
    )

def upload_from_processor_files(
    self,
    *,
    projectsite: str,
    processor: SignalConfigUploadSource,
    path_signal_sensor_map: str | Path | None = None,
    path_sensor_tc_map: str | Path | None = None,
    assetlocations_by_turbine: Mapping[str, str] | None = None,
    permission_group_ids: Sequence[int] | None = None,
    model_definition: str | None = None,
) -> dict[str, AssetSignalUploadResult]:
    """Process configs, resolve optional file maps, and upload by turbine.

    Parameters
    ----------
    projectsite
        Parent SDK project site title shared by the batch.
    processor
        Processor that populates turbine-scoped signal mappings.
    path_signal_sensor_map
        Optional JSON file keyed by turbine and signal id with SHM sensor
        lookup parameters. When ``sensor_type_id`` is itself a mapping,
        the uploader resolves it through ``get_sensor_type()`` before the
        final sensor lookup.
    path_sensor_tc_map
        Optional JSON file keyed by turbine with temperature-
        compensation signal identifiers resolved after main signal
        creation.
    assetlocations_by_turbine
        Optional turbine-to-asset-location override mapping.
    permission_group_ids
        Visibility groups applied to created SHM objects.
    model_definition
        Optional parent SDK model-definition title used when the project
        has multiple model definitions for the same asset location.

    Returns
    -------
    dict[str, AssetSignalUploadResult]
        Upload results keyed by turbine identifier.

    Examples
    --------
    >>> from unittest.mock import Mock
    >>> uploader = ShmSignalUploader(shm_api=Mock(), lookup_service=Mock())
    >>> processor = Mock()
    >>> processor.signals_data = {}
    >>> processor.signals_derived_data = {}
    >>> uploader.upload_from_processor_files(projectsite="Project A", processor=processor)
    {}
    """
    processor.signals_process_data()
    sensor_serial_numbers_by_turbine = self._resolve_sensor_serial_numbers_by_turbine(path_signal_sensor_map)
    temperature_compensation_signal_refs_by_turbine = self._load_temperature_compensation_signal_refs_by_turbine(
        path_sensor_tc_map
    )
    return self.upload_turbines(
        projectsite=projectsite,
        signals_by_turbine=processor.signals_data,
        derived_signals_by_turbine=processor.signals_derived_data,
        assetlocations_by_turbine=assetlocations_by_turbine,
        permission_group_ids=permission_group_ids,
        model_definition=model_definition,
        sensor_serial_numbers_by_turbine=sensor_serial_numbers_by_turbine,
        temperature_compensation_signal_refs_by_turbine=(temperature_compensation_signal_refs_by_turbine),
    )