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 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"])
    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"])
    location_frame = assetlocations[assetlocations["id"].isin(location_ids)].copy()
    columns = [column for column in ["id", "title", "northing", "easting"] 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"]
    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,
    filters: ResultQuery | Mapping[str, Any] | None = None,
    *,
    plot_type: str | None = None,
) -> PlotResponse:
    """Render a chart for the requested analysis."""
    query = self._coerce_query(analysis_name, filters)
    records = self.get_result_series(analysis_name, query)
    analysis = self.registry.get(analysis_name)
    plot_request = PlotRequest(
        analysis_name=analysis_name,
        filters=query.model_dump(),
        plot_type=plot_type,
        context=self._plot_context(records, plot_type=plot_type),
    )
    return analysis.plot(records, request=plot_request)