Time Series¶

pydantic model infrasys.time_series_models.TimeSeriesData¶

Base class for all time series models.

Time series identity is owned by the Rust infrastore core (content hash plus the owner/name/features association key); infrasys does not assign its own id/uuid.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Show JSON schema

{
   "title": "TimeSeriesData",
   "description": "Base class for all time series models.\n\nTime series identity is owned by the Rust ``infrastore`` core (content hash plus\nthe owner/name/features association key); infrasys does not assign its own id/uuid.",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "normalization": {
         "anyOf": [
            {
               "discriminator": {
                  "mapping": {
                     "by_value": "#/$defs/NormalizationByValue",
                     "max": "#/$defs/NormalizationMax"
                  },
                  "propertyName": "normalization_type"
               },
               "oneOf": [
                  {
                     "$ref": "#/$defs/NormalizationMax"
                  },
                  {
                     "$ref": "#/$defs/NormalizationByValue"
                  }
               ]
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Defines the type of normalization performed on the data, if any.",
         "title": "Normalization"
      }
   },
   "$defs": {
      "NormalizationByValue": {
         "additionalProperties": false,
         "description": "Perform normalization by a user-defined value.",
         "properties": {
            "value": {
               "title": "Value",
               "type": "number"
            },
            "normalization_type": {
               "const": "by_value",
               "default": "by_value",
               "title": "Normalization Type",
               "type": "string"
            }
         },
         "required": [
            "value"
         ],
         "title": "NormalizationByValue",
         "type": "object"
      },
      "NormalizationMax": {
         "additionalProperties": false,
         "description": "Perform normalization by the max value in an array.",
         "properties": {
            "max_value": {
               "anyOf": [
                  {
                     "type": "number"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Max Value"
            },
            "normalization_type": {
               "const": "max",
               "default": "max",
               "title": "Normalization Type",
               "type": "string"
            }
         },
         "title": "NormalizationMax",
         "type": "object"
      }
   },
   "additionalProperties": false,
   "required": [
      "name"
   ]
}

Config:

str_strip_whitespace: bool = True
validate_assignment: bool = True
validate_default: bool = True
extra: str = forbid
use_enum_values: bool = False
arbitrary_types_allowed: bool = True
populate_by_name: bool = True
validate_by_alias: bool = True
validate_by_name: bool = True

Fields:

name (str)
normalization (None | infrasys.normalization.NormalizationMax | infrasys.normalization.NormalizationByValue)

field name: str [Required]¶

field normalization: Annotated[None | NormalizationMax | NormalizationByValue, FieldInfo(annotation=NoneType, required=True, description='Defines the type of normalization performed on the data, if any.', discriminator='normalization_type')] = None¶: Defines the type of normalization performed on the data, if any.

property summary: str¶: Return the name of the time series array with its type.

pydantic model infrasys.time_series_models.SingleTimeSeries¶

Defines a time array with a single dimension of floats.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Show JSON schema

{
   "title": "SingleTimeSeries",
   "description": "Defines a time array with a single dimension of floats.",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "normalization": {
         "anyOf": [
            {
               "discriminator": {
                  "mapping": {
                     "by_value": "#/$defs/NormalizationByValue",
                     "max": "#/$defs/NormalizationMax"
                  },
                  "propertyName": "normalization_type"
               },
               "oneOf": [
                  {
                     "$ref": "#/$defs/NormalizationMax"
                  },
                  {
                     "$ref": "#/$defs/NormalizationByValue"
                  }
               ]
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Defines the type of normalization performed on the data, if any.",
         "title": "Normalization"
      },
      "data": {
         "anyOf": [],
         "title": "Data"
      },
      "resolution": {
         "format": "duration",
         "title": "Resolution",
         "type": "string"
      },
      "initial_timestamp": {
         "format": "date-time",
         "title": "Initial Timestamp",
         "type": "string"
      }
   },
   "$defs": {
      "NormalizationByValue": {
         "additionalProperties": false,
         "description": "Perform normalization by a user-defined value.",
         "properties": {
            "value": {
               "title": "Value",
               "type": "number"
            },
            "normalization_type": {
               "const": "by_value",
               "default": "by_value",
               "title": "Normalization Type",
               "type": "string"
            }
         },
         "required": [
            "value"
         ],
         "title": "NormalizationByValue",
         "type": "object"
      },
      "NormalizationMax": {
         "additionalProperties": false,
         "description": "Perform normalization by the max value in an array.",
         "properties": {
            "max_value": {
               "anyOf": [
                  {
                     "type": "number"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Max Value"
            },
            "normalization_type": {
               "const": "max",
               "default": "max",
               "title": "Normalization Type",
               "type": "string"
            }
         },
         "title": "NormalizationMax",
         "type": "object"
      }
   },
   "additionalProperties": false,
   "required": [
      "name",
      "data",
      "resolution",
      "initial_timestamp"
   ]
}

Config:

str_strip_whitespace: bool = True
validate_assignment: bool = True
validate_default: bool = True
extra: str = forbid
use_enum_values: bool = False
arbitrary_types_allowed: bool = True
populate_by_name: bool = True
validate_by_alias: bool = True
validate_by_name: bool = True

Fields:

data (numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy._typing._array_like._ScalarT]] | pint.registry.Quantity)
initial_timestamp (datetime.datetime)
resolution (datetime.timedelta)

Validators:

check_data » data

field data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | Quantity [Required]¶

Validated by:

check_data

field initial_timestamp: datetime [Required]¶

field resolution: timedelta [Required]¶

validator check_data » data¶: Check time series data.

classmethod from_array(data: Sequence | ndarray[tuple[Any, ...], dtype[_ScalarT]] | Quantity, name: str, initial_timestamp: datetime, resolution: timedelta, normalization: Annotated[None | NormalizationMax | NormalizationByValue, FieldInfo(annotation=NoneType, required=True, description='Defines the type of normalization performed on the data, if any.', discriminator='normalization_type')] = None) → SingleTimeSeries¶

Method of SingleTimeSeries that creates an instance from a sequence.

Parameters:

data – Sequence that contains the values of the time series
initial_time – Start time for the time series (e.g., datetime(2020,1,1))
resolution – Resolution of the time series (e.g., 30min, 1hr)
name – Name assigned to the values of the time series (e.g., active_power)

Return type:

SingleTimeSeries

Transaction¶

The transaction object for time series operations, yielded by infrasys.system.System.time_series_transaction(). Call the time series methods on it to batch them together and to make them share one unit of rollback.

class infrasys.time_series_transaction.TimeSeriesTransaction(manager: TimeSeriesManager, context: TimeSeriesStorageContext, *, validate_owner: Callable[[Component | SupplementalAttribute], None])¶

One open batch of time series work, yielded by System.time_series_transaction.

Every method mirrors the System method of the same name and routes through this transaction’s batch: additions are buffered into one bulk write, reads see the batch’s staged work, and if the block raises, everything the transaction did is rolled back.

Examples

>>> with system.time_series_transaction() as txn:
...     for gen, ts in profiles:
...         txn.add_time_series(ts, gen)

property has_staged_data: bool¶: Return True if additions are buffered but not yet written to the store.

add_time_series(time_series: TimeSeriesData, *owners: Component | SupplementalAttribute, **features: Any) → TimeSeriesKey¶

Store a time series array for one or more components or supplemental attributes.

Mirrors System.add_time_series; the addition is buffered and written with the rest of the batch.

copy_time_series(dst: Component | SupplementalAttribute, src: Component | SupplementalAttribute, name_mapping: dict[str, str] | None = None) → None¶: Copy all time series from src to dst. Mirrors System.copy_time_series.

Return a time series array, including ones staged in this transaction.

Mirrors System.get_time_series.

get_time_series_by_key(owner: Component | SupplementalAttribute, key: TimeSeriesKey) → Any¶: Return a time series array by key. Mirrors System.get_time_series_by_key.

has_time_series(owner: ~infrasys.component.Component | ~infrasys.supplemental_attribute.SupplementalAttribute, name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] | None = <class 'infrasys.time_series_models.SingleTimeSeries'>, **features: ~typing.Any) → bool¶

Return True if the owner has matching time series, staged or committed.

Mirrors System.has_time_series.

list_time_series(component: ~infrasys.component.Component, name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] | None = <class 'infrasys.time_series_models.SingleTimeSeries'>, start_time: ~datetime.datetime | None = None, length: int | None = None, **features: ~typing.Any) → list[TimeSeriesData]¶: Return all matching time series. Mirrors System.list_time_series.

list_time_series_keys(owner: ~infrasys.component.Component | ~infrasys.supplemental_attribute.SupplementalAttribute, name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] | None = <class 'infrasys.time_series_models.SingleTimeSeries'>, **features: ~typing.Any) → list[TimeSeriesKey]¶: Return all matching time series keys. Mirrors System.list_time_series_keys.

list_time_series_metadata(component: ~infrasys.component.Component, name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] | None = <class 'infrasys.time_series_models.SingleTimeSeries'>, **features: ~typing.Any) → list[TimeSeriesKey]¶: Return all matching metadata. Mirrors System.list_time_series_metadata.

remove_time_series(*owners: ~infrasys.component.Component | ~infrasys.supplemental_attribute.SupplementalAttribute, name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] | None = <class 'infrasys.time_series_models.SingleTimeSeries'>, **features: ~typing.Any) → None¶

Remove matching time series; reversible if the block rolls back.

Mirrors System.remove_time_series.

transform_single_time_series(horizon: timedelta, interval: timedelta) → int¶

Derive Deterministic forecasts from every stored SingleTimeSeries.

Mirrors System.transform_single_time_series.

build_time_series_reader(resolution: timedelta, *, name: str | None = None, name_glob: str | None = None, component_type: Type[Component] | None = None, **features: Any) → TimeSeriesReader¶

Build a per-timestamp reader covering the batch’s staged series as well.

Mirrors System.build_time_series_reader.

build_forecast_reader(resolution: ~datetime.timedelta, *, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] = <class 'infrasys.time_series_models.Deterministic'>, name: str | None = None, name_glob: str | None = None, component_type: ~typing.Type[~infrasys.component.Component] | None = None, **features: ~typing.Any) → ForecastReader¶: Build a per-window forecast reader. Mirrors System.build_forecast_reader.

Readers¶

Readers return every matched component’s data at one timestamp, which is the transpose of what the other read methods provide. Build them with infrasys.system.System.build_time_series_reader() and infrasys.system.System.build_forecast_reader(). See How to read time series by timestamp for a worked example.

class infrasys.time_series_reader.TimeSeriesReader(store: Any, reader: Any, group_component_ids: list[tuple[int, ...]], units: dict[int, QuantityMetadata | None])¶

Reads the value of every matched SingleTimeSeries at one timestamp.

All series covered by a reader share one grid (initial timestamp, resolution, and length); the store rejects the build otherwise.

Examples

>>> reader = system.build_time_series_reader(timedelta(hours=1), name="active_power")
>>> for timestamp in reader.timestamps:
...     values = reader.read(timestamp)  # {component id: float}

property component_ids: tuple[int, ...]¶: Return the ids of every component covered by this reader.

property units: dict[int, QuantityMetadata | None]¶

Return each component’s stored units, or None where the series is unitless.

Values returned by read() are raw magnitudes in these units. Attaching pint quantities per value would dominate the cost of a stepping loop, so the readers leave that to the caller.

property grid: dict[str, Any]¶

initial timestamp, resolution, and length.

Type:: Return the shared grid

property timestamps: list[datetime]¶

Return every timestamp on the reader’s grid, in order.

Each timestamp is spelled the way the cohort’s series were written: naive for a zoneless cohort, aware otherwise. Feed them straight back to read().

read(when: datetime) → dict[int, Any]¶

Return {component id: value} for every covered series at when.

Raises:: InvalidParameterError – Raised by the store if when is not on the reader’s grid, or if it is spelled differently than the cohort — a naive when against series that record instants, or an aware one against a zoneless cohort. Use a timestamp from timestamps and the spelling always matches.

read_columns(when: datetime) → list[tuple[tuple[int, ...], Any]]¶

Return (component ids, values array) per columnar group at when.

The escape hatch for loops where building a dict per step is measurable: the arrays come straight from the store with no per-value Python objects. Each array is shaped (num_components, *element_shape) and aligned to its id tuple.

class infrasys.time_series_reader.ForecastReader(store: Any, reader: Any, component_ids: tuple[int, ...], slots: tuple[int, ...], units: dict[int, QuantityMetadata | None])¶

Reads the forecast window of every matched forecast at one timestamp.

Components whose forecasts are backed by the same array collapse to a single slot. The store performs one physical read per slot rather than one per component, and read() hands every component in a slot the same array object rather than a copy. read_slots() exposes the deduplicated form directly.

Examples

>>> reader = system.build_forecast_reader(timedelta(hours=1), name="active_power")
>>> for timestamp in reader.timestamps:
...     windows = reader.read(timestamp)  # {component id: ndarray of (horizon,)}

property component_ids: tuple[int, ...]¶: Return the ids of every component covered by this reader.

property units: dict[int, QuantityMetadata | None]¶: Return each component’s stored units, or None where the series is unitless.

property timeline: dict[str, Any]¶

initial timestamp, resolution, interval, and count.

Type:: Return the window timeline

property timestamps: list[datetime]¶

Return every window start timestamp, in order.

Each timestamp is spelled the way the cohort’s forecasts were written: naive for a zoneless cohort, aware otherwise. Feed them straight back to read().

property num_slots: int¶

Return the number of distinct windows, one physical read each per step.

This is at most len(component_ids), and smaller whenever components share a backing array — for example after System.transform_single_time_series(), or wherever many components were given the same profile.

property slots: dict[int, int]¶: Return {component id: slot}. Equal slots mean a shared window.

components_by_slot() → dict[int, list[int]]¶: Return {slot: [component ids]} for callers driving work per unique window.

read(when: datetime) → dict[int, Any]¶

Return {component id: window array} for every covered forecast at when.

Components sharing a slot share one array object, so a shared profile is materialized once no matter how many components carry it. Treat the arrays as read-only for that reason.

Raises:: InvalidParameterError – Raised by the store if when is not on the reader’s timeline, or if it is spelled differently than the cohort. Use a timestamp from timestamps and the spelling always matches.

read_slots(when: datetime) → dict[int, Any]¶: Return {slot: window array} at when, one entry per unique window.