atomicds.timeseries.provider.TimeseriesProvider

class atomicds.timeseries.provider.TimeseriesProvider[source]

Bases: ABC, Generic[R]

Strategy interface for parsing timeseries by domain.

__init__()

Methods

__init__()

build_result(client, data_id, data_type, ts_df)

Build time series result object

fetch_raw(client, data_id)

Perform the HTTP GET(s) to retrieve raw payload(s).

fetch_snapshot(client, req)

Resolve one snapshot request → domain-specific ImageResult (or None).

snapshot_image_uuids(frames_payload)

Extract requests from frames payload.

snapshot_url(data_id)

API endpoint that exposes extracted/snapshot frames.

to_dataframe(raw)

Convert raw payload to a tidy DataFrame with domain-specific renames/index.

Attributes

TYPE

abstractmethod fetch_raw(client: BaseClient, data_id: str) Any[source]

Perform the HTTP GET(s) to retrieve raw payload(s).

Parameters:
  • client (BaseClient)

  • data_id (str)

Return type:

Any

abstractmethod to_dataframe(raw: Any) DataFrame[source]

Convert raw payload to a tidy DataFrame with domain-specific renames/index.

Parameters:

raw (Any)

Return type:

DataFrame

abstractmethod build_result(client: BaseClient, data_id: str, data_type: str, ts_df: DataFrame) R[source]

Build time series result object

Parameters:
  • client (BaseClient)

  • data_id (str)

  • data_type (str)

  • ts_df (DataFrame)

Return type:

TypeVar(R)

snapshot_url(data_id: str) str[source]

API endpoint that exposes extracted/snapshot frames.

Parameters:

data_id (str)

Return type:

str

snapshot_image_uuids(frames_payload: dict[str, Any]) list[dict][source]

Extract requests from frames payload. Default: no snapshots.

Parameters:

frames_payload (dict[str, Any])

Return type:

list[dict]

fetch_snapshot(client: BaseClient, req: dict) Any | None[source]

Resolve one snapshot request → domain-specific ImageResult (or None).

Parameters:
  • client (BaseClient)

  • req (dict)

Return type:

Any | None