Services¶

def list_analyses(self, name: str | None = None, **kwargs: Any) -> pd.DataFrame:
    """Retrieve analysis rows from the backend."""
    return self.api.list_analyses(name=name, **kwargs)["data"]

def get_analysis(self, analysis_id: int) -> pd.DataFrame:
    """Retrieve one analysis row from the backend."""
    return self.api.get_analysis(id=analysis_id)["data"]

def list_results(self, query: ResultQuery) -> pd.DataFrame:
    """Retrieve raw result rows from the backend."""
    return self.api.list_results(**query.to_backend_filters())["data"]

def create_analysis(self, payload: Mapping[str, Any]) -> Mapping[str, Any]:
    """Create an analysis row."""
    return self.api.create_analysis(dict(payload))

def create_result(self, payload: Mapping[str, Any]) -> Mapping[str, Any]:
    """Create a single result row."""
    return self.api.create_result(dict(payload))

def create_results_bulk(self, payloads: Sequence[Mapping[str, Any]]) -> Mapping[str, Any]:
    """Create multiple result rows."""
    return self.api.create_results_bulk([dict(payload) for payload in payloads])

def create_or_update_results_bulk(self, payloads: Sequence[Mapping[str, Any]]) -> Mapping[str, Any]:
    """Create missing result rows and patch existing ones."""
    return self.api.create_or_update_results_bulk([dict(payload) for payload in payloads])

def update_result(self, result_id: int, payload: Mapping[str, Any]) -> Mapping[str, Any]:
    """Patch a single result row."""
    return self.api.update_result(result_id, dict(payload))

def get_location_frame(self, location_ids: Sequence[int]) -> pd.DataFrame:
    """Retrieve location metadata required by map-oriented plots."""
    if not location_ids:
        return pd.DataFrame(columns=["id", "title", "northing", "easting", "elevation"])
    location_api = LocationsAPI(api_root=self.api.base_api_root, **self.api._auth_kwargs())
    assetlocations = location_api.get_assetlocations()["data"]
    if assetlocations.empty or "id" not in assetlocations.columns:
        return pd.DataFrame(columns=["id", "title", "northing", "easting", "elevation"])
    location_frame = assetlocations[assetlocations["id"].isin(location_ids)].copy()
    columns = [
        column for column in ["id", "title", "northing", "easting", "elevation"] if column in location_frame.columns
    ]
    return location_frame.loc[:, columns]

def deserialize_result_series(
    self,
    raw_data: Sequence[Mapping[str, Any]] | pd.DataFrame,
) -> list[ResultSeries]:
    """Deserialize raw backend rows into typed result series."""
    if isinstance(raw_data, pd.DataFrame):
        raw_records = cast(list[dict[str, Any]], raw_data.to_dict(orient="records"))
    else:
        raw_records = [dict(row) for row in raw_data]
    return [self.serializer.from_mapping(row) for row in raw_records]

def get_result_series(
    self,
    analysis_name: str,
    filters: ResultQuery | Mapping[str, Any] | None = None,
) -> list[ResultSeries]:
    """Fetch and deserialize typed result series for the given analysis."""
    query = self._coerce_query(analysis_name, filters)
    frame = self.repository.list_results(query)
    return self.deserialize_result_series(frame)

def get_location_frame(self, location_ids: Sequence[int]) -> pd.DataFrame:
    """Return location metadata for the given backend location identifiers."""
    columns = ["id", "title", "northing", "easting", "elevation"]
    if not location_ids:
        return pd.DataFrame(columns=columns)
    get_location_frame = getattr(self.repository, "get_location_frame", None)
    if not callable(get_location_frame):
        return pd.DataFrame(columns=columns)
    location_frame = get_location_frame(location_ids)
    if not isinstance(location_frame, pd.DataFrame):
        return pd.DataFrame(columns=columns)
    if location_frame.empty:
        return pd.DataFrame(columns=columns)
    available_columns = [column for column in columns if column in location_frame.columns]
    selected_frame = cast(pd.DataFrame, location_frame.loc[:, available_columns])
    return selected_frame.copy()

def get_results(self, analysis_name: str, filters: ResultQuery | Mapping[str, Any] | None = None) -> pd.DataFrame:
    """Return normalized analysis data for the given analysis."""
    records = self.get_result_series(analysis_name, filters)
    analysis = self.registry.get(analysis_name)
    return analysis.from_results(records)

def plot_results(
    self,
    analysis_name: str | None = None,
    filters: ResultQuery | Mapping[str, Any] | None = None,
    *,
    plot_type: str | None = None,
    source_filters: Mapping[str, ResultQuery | Mapping[str, Any]] | None = None,
) -> PlotResponse:
    """Render a chart for a single analysis or a registered cross-analysis plot.

    Parameters
    ----------
    analysis_name : str | None, optional
        Analysis name used for ordinary single-analysis plot types. For
        owner-invariant cross-analysis plot types such as
        ``cross_analysis_fleetwide``, this can be omitted.
    filters : ResultQuery | Mapping[str, Any] | None, optional
        Shared query filters. For cross-analysis plots, these filters are
        applied to every named source unless overridden in
        ``source_filters``.
    plot_type : str | None, optional
        Plot type to render.
    source_filters : Mapping[str, ResultQuery | Mapping[str, Any]] | None, optional
        Per-source query overrides for registered cross-analysis plots.
        The mapping keys must match the plot definition source keys, such as
        ``"frequency"`` and ``"verification"`` for
        ``cross_analysis_fleetwide``.
    """
    requested_analysis_name = analysis_name or self._coerce_query(None, filters).analysis_name
    plot_definition = get_plot_definition(plot_type, analysis_name=requested_analysis_name)
    if plot_definition is not None:
        return self._plot_defined_results(
            plot_definition,
            requested_analysis_name=requested_analysis_name,
            filters=filters,
            plot_type=plot_type,
            source_filters=source_filters,
        )
    if source_filters:
        raise ValueError("source_filters can only be used with registered cross-analysis plot types.")
    if requested_analysis_name is None:
        raise ValueError("analysis_name is required for plot types without a registered cross-analysis definition.")
    query = self._coerce_query(requested_analysis_name, filters)
    records = self.get_result_series(requested_analysis_name, query)
    analysis = self.registry.get(requested_analysis_name)
    plot_request = PlotRequest(
        analysis_name=requested_analysis_name,
        filters=query.model_dump(),
        plot_type=plot_type,
        context=self._plot_context(
            records,
            plot_type=plot_type,
            analysis_name=requested_analysis_name,
            query=query,
        ),
    )
    return analysis.plot(records, request=plot_request)