System¶

class infrasys.system.System(name: str | None = None, description: str | None = None, auto_add_composed_components: bool = False, time_series_manager: TimeSeriesManager | None = None, uuid: UUID | None = None, **kwargs: Any)¶

Implements behavior for systems

Constructs a System.

Parameters:

name (str | None) – Optional system name
description (str | None) – Optional system description
auto_add_composed_components (bool) – Set to True to automatically add composed components to the system in add_components. The default behavior is to raise an ISOperationNotAllowed when this condition occurs. This handles values that are components, such as generator.bus, and lists of components, such as subsystem.generators, but not any other form of nested components.
time_series_manager (None | TimeSeriesManager) – Users should not pass this. De-serialization (from_json) will pass a constructed manager.
kwargs (Any) –
Configures time series behaviors:
- time_series_storage_type: Defaults to TimeSeriesStorageType.TIME_SERIES_STORE.
- time_series_read_only: Disables add/remove of time series, defaults to False.
- time_series_directory: Location to store time series files, defaults to the system’s tmp directory. Use an alternate location if the space in that directory is limited, such as on a compute node with no local storage.
- time_series_compression: HDF5 compression filter for the infrastore backend; “deflate” (default) or “none”.
- time_series_compression_level: DEFLATE level 0-9, defaults to 3.
- time_series_shuffle: Enable the byte-shuffle filter for DEFLATE, defaults to True.

Examples

>>> system = System(name="my_system")
>>> system2 = System(name="my_system", time_series_directory="/tmp/scratch")

close() → None¶: Close open resources such as SQLite connections.

property auto_add_composed_components: bool¶: Return the setting for auto_add_composed_components.

to_json(filename: Path | str, overwrite=False, indent=None, data=None) → None¶

Write the contents of a system to a JSON file. Time series will be written to a directory at the same level as filename.

Parameters:

filename (Path | str) – Filename to write. If the parent directory does not exist, it will be created.
overwrite (bool) – Set to True to overwrite the file if it already exists.
indent (int | None) – Indentation level in the JSON file. Defaults to no indentation.
data (dict | None) – This is an override for packages that compose this System inside a parent System class. If set, it will be the outer object in the JSON file. It must not set the key ‘system’. Packages that derive a custom instance of this class should leave this field unset.

Raises:

ISOperationNotAllowed – Raised if a time series transaction is open. Move the call outside the time_series_transaction block so the copy reflects committed state.

Examples

>>> system.to_json("systems/system1.json")
INFO: Wrote system data to systems/system1.json
INFO: Copied time series data to systems/system1_time_series

classmethod from_json(filename: Path | str, upgrade_handler: Callable | None = None, **kwargs) → System¶

Deserialize a System from a JSON file. Refer to System constructor for kwargs.

Parameters:

filename (Path | str) – JSON file containing the system data.
upgrade_handler (Callable | None) – Optional function to handle data format upgrades. Should only be set when the parent package composes this package. If set, it will be called before de-serialization of the components.

Examples

>>> system = System.from_json("systems/system1.json")

classmethod load(zip_path: Path | str, time_series_directory: Path | str | None = None, upgrade_handler: Callable | None = None, **kwargs: Any) → System¶

Load a System from a zip archive created by the save() method.

The zip file will be extracted to a temporary directory, the system will be deserialized, and the temporary files will be cleaned up automatically. Time series storage files are copied to a permanent location during deserialization.

Parameters:

zip_path (Path | str) – Path to the zip file containing the system.
time_series_directory (Path | str) – Path to the final time series location
upgrade_handler (Callable | None) – Optional function to handle data format upgrades. Should only be set when the parent package composes this package. If set, it will be called before de-serialization of the components.
**kwargs (Any) – Additional arguments passed to the System constructor. Refer to System constructor for available options. Use time_series_directory to specify where time series files should be stored.

Returns:

The deserialized system.

Return type:

System

Raises:

ISFileExists – Raised if the zip file does not exist.
ISInvalidParameter – Raised if the zip file is not a valid zip archive or doesn’t contain a valid system.
FileNotFoundError – Raised if there is no JSON file in the zip folder.

Examples

>>> system = System.load("my_system.zip")
>>> system2 = System.load(Path("archived_systems/system1.zip"))
>>> # Specify where time series files should be stored
>>> system3 = System.load("my_system.zip", time_series_directory="/path/to/storage")

See also

save: Save a system to a directory or zip file
from_json: Load a system from a JSON file

to_records(component_type: Type[Component], filter_func: Callable | None = None, **kwargs) → Iterable[dict]¶

Return a list of dictionaries of components (records) with the requested type(s) and optionally match filter_func.

Parameters:

components – Component types to get as dictionaries
filter_func – A function to filter components. Default is None
kwargs –
Configures Pydantic model_dump behaviour
- exclude: List or dict of excluded fields.

Notes

If a component type is an abstract type, all matching concrete subtypes will be included in the output.

It is only recommended to use this function on a single “concrete” types. For example, if you have an abstract type called Generator and you create two subtypes called ThermalGenerator and RenewableGenerator where some fields are different, if you pass the return of System.to_records(Generator) to pandas.DataFrame.from_records, each ThermalGenerator row will have NaN values for RenewableGenerator-specific fields.

Examples

To get a tabular representation of a certain type you can use: >>> import pandas as pd >>> df = pd.DataFrame.from_records(System.to_records(SimpleGen))

With polars: >>> import polars as pl >>> df = pl.DataFrame(System.to_records(SimpleGen))

classmethod from_dict(data: dict[str, Any], time_series_parent_dir: Path | str, upgrade_handler: Callable | None = None, **kwargs: Any) → System¶

Deserialize a System from a dictionary.

Parameters:

data (dict[str, Any]) – System data in serialized form.
time_series_parent_dir (Path | str) – Directory that contains the system’s time series directory.
upgrade_handler (Callable | None) – Optional function to handle data format upgrades. Should only be set when the parent package composes this package. If set, it will be called before de-serialization of the components.

Examples

>>> system = System.from_dict(data, "systems")

save(fpath: Path | str, filename: str = 'system.json', zip: bool = False, overwrite: bool = False) → None¶

Save the contents of a system and the Time series in a single directory.

By default, this method creates the user specified folder using the to_json method. If user sets zip = True, we create the folder of the user (if it does not exists), zip it to the same location specified and delete the folder.

Parameters:

fpath (Path | str) – Filepath to write the contents of the system.
zip (bool) – Set to True if you want to archive to a zip file.
filename (str) – Name of the sytem to serialize. Default value: “system.json”.
overwrite (bool) – Overwrites the system if it already exist on the fpath.

Raises:

FileExistsError – Raised if the folder provided exists and the overwrite flag was not provided.

Examples

>>> fpath = Path("folder/subfolder/")
>>> system.save(fpath)
INFO: Wrote system data to folder/subfolder/system.json
INFO: Copied time series data to folder/subfolder/system_time_series

>>> system_fname = "my_system.json"
>>> fpath = Path("folder/subfolder/")
>>> system.save(fpath, filename=system_fname, zip=True)
INFO: Wrote system data to folder/subfolder/my_system.json
INFO: Copied time series data to folder/subfolder/my_system_time_series
INFO: System archived at folder/subfolder/my_system.zip

See also

to_json: System serialization

add_component(component: Component, **kwargs) → None¶

Add one component to the system.

Parameters:: component (Component) – Component to add to the system.
Raises:: ISAlreadyAttached – Raised if a component is already attached to a system.

Examples

>>> system.add_component(Bus.example())

See also

add_components

add_components(*components: Component, **kwargs) → None¶

Add one or more components to the system.

Parameters:: components (Component) – Component(s) to add to the system.
Raises:: ISAlreadyAttached – Raised if a component is already attached to a system.

Examples

>>> system.add_components(Bus.example(), Generator.example())

See also

add_component

add_supplemental_attribute(component: Component, attribute: SupplementalAttribute) → None¶

Attach a supplemental attribute to a component. The attribute will get added to the system if it is not already stored.

Parameters:

component – Existing component
attribute – Supplemental attribute to attach to the component

Raises:

ISAlreadyAttached – Raised if the component and attribute are already attached.

Examples

>>> bus = Bus.example()
>>> system.add_component(bus)
>>> geo_json = GeographicInfo.example()
>>> system.add_supplemental_attribute(bus, geo_json)

copy_component(component: Component, name: str | None = None, attach: bool = False) → Any¶

Create a copy of the component. Time series data is excluded.

The new component will have a different ID than the original.
The copied component will have shared references to any composed components.

The intention of this method is to provide a way to create variants of a component that will be added to the same system. Please refer to :deepcopy_component: to create copies that are suitable for addition to a different system.

Parameters:

component (Component) – Source component
name (str) – Optional, if None, keep the original name.
attach (bool) – Optional, if True, attach the new component to the system.

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> gen2 = system.copy_component(gen, name="gen2")
>>> gen3 = system.copy_component(gen, name="gen3", attach=True)

See also

deepcopy_component

deepcopy_component(component: Component) → Any¶

Create a deep copy of the component and all composed components. All attributes, including names and IDs, will be identical to the original. Unlike copy_component(), there will be no shared references to composed components.

The intention of this method is to provide a way to create variants of a component that will be added to a different system. Please refer to :copy_component: to create copies that are suitable for addition to the same system.

Parameters:: component (Component) – Source component

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> gen2 = system.deepcopy_component(gen)

See also

copy_component

get_component(component_type: Type[T], name: str) → T¶

Return the component with the passed type and name.

Parameters:

component_type (Type[T]) – Generic component type
name (Type) – Name of component

Raises:

ISDuplicateNames – Raised if more than one component match the inputs.

Examples

>>> system.get_component(Generator, "gen1")

See also

list_by_name

get_component_by_label(label: str) → Any¶

Return the component with the label.

Note that this method is slower than get_component() because the component type cannot be looked up directly. Code that is looping over components repeatedly should not use this method.

Parameters:

label (str)

Raises:

ISNotStored – Raised if the label does not match a stored component.
ISOperationNotAllowed – Raised if there is more than one matching component.

Examples

>>> component = system.get_component_by_label("Bus.bus1")

get_component_by_id(id_: int) → Any¶

Return the component with the input integer ID.

Raises:: ISNotStored – Raised if the ID is not stored.

Examples

>>> component = system.get_component_by_id(5)

get_components(*component_types: Type[T], filter_func: Callable | None = None) → Iterable[T]¶

Return the components with the passed type(s) and that optionally match filter_func.

Parameters:

component_type (Type[T]) – If component_type is an abstract type, all matching subtypes will be returned. The function will return all the matching component_type passed.
filter_func (Callable | None) – Optional function to filter the returned values. The function must accept a component as a single argument.

Examples

>>> for component in system.get_components(Component)
    print(component.label)
>>> names = {"bus1", "bus2", "gen1", "gen2"}
>>> for component in system.get_components(
    Component,
    filter_func=lambda x: x.name in names,
):
    print(component.label)

To request multiple component types: >>> for component in system.get_components(SimpleGenerator, SimpleBus) print(component.label)

get_component_types() → Iterable[Type[Component]]¶

Return an iterable of all component types stored in the system.

Examples

>>> for component_type in system.get_component_types():
print(component_type)

get_components_with_supplemental_attribute(attribute: SupplementalAttribute) → list[Component]¶: Return all components attached to the given supplemental attribute.

get_supplemental_attributes_with_component(component: Component, supplemental_attribute_type: Type[U] | None = None, filter_func: Callable[[U], bool] | None = None) → list[U]¶: Return all supplemental attributes attached to the given component and optionally, with the given attribute type.

get_supplemental_attribute_by_id(id_: int) → SupplementalAttribute¶: Return the supplemental attribute with the given integer ID.

get_supplemental_attribute_counts_by_type() → list[dict[str, Any]]¶: Return a list of dicts of stored supplemental attribute counts by type.

get_num_supplemental_attributes() → int¶: Return the number of supplemental attributes stored in the system.

get_num_components_with_supplemental_attributes() → int¶: Return the number of supplemental attributes stored in the system.

has_supplemental_attribute(component: Component, supplemental_attribute_type: Type[SupplementalAttribute] | None = None) → bool¶: Return True if the component has a supplemental attribute, optionally of the given type.

has_supplemental_attribute_association(component: Component, supplemental_attribute: SupplementalAttribute) → bool¶: Return True if the component and supplemental attribute have an association.

has_component(component) → bool¶: Return True if the component is attached.

list_child_components(component: Component, component_type: Type[Component] | None = None) → list[Component]¶

Return a list of all components that this component composes.

Parameters:

component (Component)
component_type (Optional[Type[Component]]) – Filter the returned list to components of this type. If the type has subclasses, all subclasses will be included.

See also

list_parent_components

list_parent_components(component: Component, component_type: Type[Component] | None = None) → list[Component]¶

Return a list of all components that compose this component.

An example usage is where you need to find all components connected to a bus and the Bus class does not contain that information. The system tracks these connections internally and can find those components quickly.

Parameters:

component (Component)
component_type (Optional[Type[Component]]) – Filter the returned list to components of this type. If the type has subclasses, all subclasses will be included.

Examples

>>> components = system.list_parent_components(bus)
>>> print(f"These components are connected to {bus.label}: ", " ".join(components))

See also

list_child_components

list_components_by_name(component_type: Type[Component], name: str) → list[Any]¶

Return all components that match component_type and name.

Parameters:

component_type (Type)
name (str)

Examples

system.list_components_by_name(Generator, “gen1”)

iter_all_components() → Iterable[Any]¶

Return an iterator over all components.

Examples

>>> for component in system.iter_all_components()
    print(component.label)

See also

get_components

rebuild_component_associations() → None¶: Clear the component associations and rebuild the table. This may be necessary if a user reassigns connected components that are part of a system.

remove_component(component: Component, cascade_down: bool = True, force: bool = False) → None¶

Remove the component from the system.

Parameters:

component (Component)
cascade_down (bool) – If True, remove all child components if they have no other parents. Defaults to True. For example, if a generator has a bus, no other component holds a reference to that bus, and you call remove_component on that generator, the bus will get removed as well.
force (bool) – If True, remove the component even if other components hold references to this component. Defaults to False.

Raises:

ISNotStored – Raised if the component is not stored in the system.
ISOperationNotAllowed – Raised if the other components hold references to this component and force=False. Raised if the store or the time series manager is read-only.

Notes

The component’s supplemental attributes are detached along with it. An attribute left with no other component keeping it is removed from the system, taking its own time series with it. Components removed by cascade_down are cleaned up the same way.

Examples

>>> gen = system.get_component(Generator, "gen1")
>>> system.remove_component(gen)

remove_component_by_name(component_type: Type[Component], name: str, cascade_down: bool = True, force: bool = False) → None¶

Remove the component with component_type and name from the system.

Parameters:

component_type (Type)
name (str)
cascade_down (bool) – Refer remove_component().
force (bool) – Refer remove_component().

Raises:

ISNotStored – Raised if the inputs do not match any components in the system.
ISOperationNotAllowed – Raised if there is more than one component with component type and name.

Examples

>>> generators = system.remove_by_name(Generator, "gen1")

remove_component_by_id(id_: int, cascade_down: bool = True, force: bool = False) → None¶

Remove the component with the given integer ID from the system.

Parameters:

id (int)
cascade_down (bool) – Refer remove_component().
force (bool) – Refer remove_component().

Raises:

ISNotStored – Raised if the ID is not stored in the system.

Examples

>>> generator = system.remove_component_by_id(5)

remove_supplemental_attribute(attribute: SupplementalAttribute) → None¶

Remove the supplemental attribute from the system.

Raises:

ISNotStored – Raised if the attribute is not attached to the system.
ISOperationNotAllowed – Raised if the store or the time series manager is read-only.

remove_supplemental_attribute_from_component(component: Component, attribute: SupplementalAttribute) → None¶

Remove the association between the component and supplemental attribute. If the attribute is not attached to any other components, remove it from the system.

Raises:

ISNotStored – Raised if the attribute is not attached to the system.
ISOperationNotAllowed – Raised if the store or the time series manager is read-only.

update_components(component_type: Type[Component], update_func: Callable, filter_func: Callable | None = None) → None¶

Update multiple components of a given type.

Parameters:

component_type (Type[Component]) – Type of component to update. Can be abstract.
update_func (Callable) – Function to call on each component. Must take a component as a single argument.
filter_func (Callable | None) – Optional function to filter the components to update. Must take a component as a single argument.

Examples

>>> system.update_components(Generator, lambda x: x.active_power *= 10)

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.

Parameters:

time_series (TimeSeriesData) – Time series data to store.
owners (Component | SupplementalAttribute) – Add the time series to all of these components or supplemental attributes.
features (Any) – Key/value pairs to store with the time series data. Must be JSON-serializable.

Returns:

Returns a key that can be used to retrieve the time series.

Return type:

TimeSeriesKey

Raises:

ISAlreadyAttached – Raised if the variable name and user attributes match any time series already attached to one of the owners.
ISOperationNotAllowed – Raised if the manager was created in read-only mode.

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> gen2 = system.get_component(Generator, "gen2")
>>> ts = SingleTimeSeries.from_array(
    data=[0.86, 0.78, 0.81, 0.85, 0.79],
    name="active_power",
    start_time=datetime(year=2030, month=1, day=1),
    resolution=timedelta(hours=1),
)
>>> system.add_time_series(ts, gen1, gen2)

See also

time_series_transaction: Batch many additions into one bulk write.

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.

Parameters:

dst (Component | SupplementalAttribute) – Destination owner.
src (Component | SupplementalAttribute) – Source owner.
name_mapping (dict[str, str] | None) – Optionally map source names to destination names. If provided, only source series whose names are present in the mapping are copied. If omitted, all series are copied with their existing names.

Notes

Source and destination must both be components or both be supplemental attributes. Cross-category copies raise ISInvalidParameter.

Raises:: ISNotStored – Raised if either owner is not attached to this system.

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> gen2 = system.get_component(Generator, "gen2")
>>> system.copy_time_series(gen1, gen2)

Return a time series array.

Parameters:

component (Component) – Component to which the time series must be attached.
name (str | None) – Optional, search for time series with this name. Required if the other inputs will match more than one time series.
time_series_type (Type[TimeSeriesData] | None) – Optional, search for time series of this type. Required if the other inputs will match more than one time series.
start_time (datetime | None) – If not None, take a slice of the time series starting at this time.
length (int | None) – If not None, take a slice of the time series with this length.
features (str) – Optional, search for time series with these attributes.

Raises:

ISNotStored – Raised if no time series matches the inputs. Raised if the inputs match more than one time series.
ISOperationNotAllowed – Raised if the inputs match more than one time series.

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> ts_full = system.get_time_series(gen1, "active_power")
>>> ts_slice = system.get_time_series(
    gen1,
    "active_power",
    start_time=datetime(year=2030, month=1, day=1, hour=5),
    length=5,
)

See also

list_time_series

get_time_series_by_key(owner: Component | SupplementalAttribute, key: TimeSeriesKey) → Any¶: Return a time series array 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 component has time series matching the inputs.

Parameters:

component (Component) – Component to check for matching time series.
name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData] | None) – Optional, search for time series with this type. Pass None to match any type.
features (str) – Optional, search for time series with these attributes.

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 time series that match the inputs.

Parameters:

component (Component) – Component to which the time series must be attached.
name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData] | None) – Optional, search for time series with this type. Pass None to match any type.
start_time (datetime | None) – If not None, take a slice of the time series starting at this time.
length (int | None) – If not None, take a slice of the time series with this length.
features (str) – Optional, search for time series with these attributes.

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> for ts in system.list_time_series(gen1):
    print(ts)

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 time series keys that match the inputs.

Parameters:

owner (Component | SupplementalAttribute) – Component to which the time series must be attached.
name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData] | None) – Optional, search for time series with this type. Pass None to match any type.
features (str) – Optional, search for time series with these attributes.

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> for key in system.list_time_series_keys(gen1):
...     time_series = system.get_time_series_by_key(gen1, key)

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 time series keys that match the inputs.

Parameters:

component (Component) – Component to which the time series must be attached.
name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData] | None) – Optional, search for time series with this type. Pass None to match any type.
features (str) – Optional, search for time series with these attributes.

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> for metadata in system.list_time_series_metadata(gen1):
    print(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 all time series arrays attached to the components or supplemental attributes matching the inputs.

Parameters:

owners – Affected components or supplemental attributes
name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData] | None) – Optional, search for time series with this type. Pass None to match any type.
features (str) – Optional, search for time series with these attributes.

Raises:

ISNotStored – Raised if no time series match the inputs.
ISOperationNotAllowed – Raised if the manager was created in read-only mode.

Examples

>>> gen1 = system.get_component(Generator, "gen1")
>>> system.remove_time_series(gen1, "active_power")

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

Derive Deterministic forecasts from every stored SingleTimeSeries.

Each SingleTimeSeries gains a forecast view that shares the same underlying array (a “perfect forecast”). After transforming, retrieve a forecast by passing time_series_type=Deterministic to get_time_series(). Returns the number of series transformed.

Parameters:

horizon – Duration of each forecast window (e.g., timedelta(hours=24)).
interval – Time between consecutive forecast windows (e.g., timedelta(hours=1)). Pass timedelta(0) to derive a single window per series, which requires horizon to span each series in full; the interval is stored verbatim, so the forecast reads back with window_count=1 and interval=timedelta(0).

Raises:

ISOperationNotAllowed – Raised if the system’s time series are read-only.

Examples

>>> system.transform_single_time_series(
...     horizon=timedelta(hours=24), interval=timedelta(hours=1)
... )
>>> forecast = system.get_time_series(gen1, "active_power", time_series_type=Deterministic)

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

Build a reader that returns every matching component’s value at one timestamp.

This is the access pattern for stepping a simulation through time. The other read methods return one component’s whole array; a reader returns one timestamp’s value across components, without holding the arrays in memory.

All matched series must share one grid (initial timestamp, resolution, and length). The reader covers the associations that match at build time; add or remove time series afterwards and you need a new reader.

Parameters:

resolution – Resolution of the series to read. One resolution per reader.
name – Only read series with this name.
name_glob – Only read series whose name matches this glob pattern (* and ?).
component_type – Only read series owned by components of this type.
zoneless – Only read series spelled one way: True for the wall-clock (zoneless) series, False for every series that names an instant, which includes any that left the spelling unset. A reader materializes one timestamp axis, so the store refuses to build one over a mix; this is how a mixed system is split into two readers that each build.
features – Only read series carrying these feature key/value pairs.

Returns:

Reader whose read(timestamp) returns {component id: value}.

Return type:

TimeSeriesReader

Raises:

InvalidParameterError – Raised by the store if no series match, if the matched series do not share one grid, or if they disagree about how their timestamps are spelled — see zoneless.

Examples

>>> reader = system.build_time_series_reader(timedelta(hours=1), name="active_power")
>>> for timestamp in reader.timestamps:
...     for component_id, value in reader.read(timestamp).items():
...         ...

See also

build_forecast_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, zoneless: bool | None = None, **features: ~typing.Any) → ForecastReader¶

Build a reader that returns every matching component’s forecast window at one time.

Components whose forecasts share a backing array collapse to a single slot, and the store reads each slot once per step rather than once per component. This matters after transform_single_time_series(), and wherever many components were given the same profile. reader.num_slots reports how many distinct windows a step actually reads.

Parameters:

resolution – Resolution of the forecasts to read. One resolution per reader.
time_series_type – Forecast type to read. A Deterministic reader also covers the forecasts derived by transform_single_time_series().
name – Only read forecasts with this name.
name_glob – Only read forecasts whose name matches this glob pattern (* and ?).
component_type – Only read forecasts owned by components of this type.
zoneless – Only read forecasts spelled one way: True for the wall-clock (zoneless) forecasts, False for every forecasts that names an instant, which includes any that left the spelling unset. A reader materializes one timestamp axis, so the store refuses to build one over a mix; this is how a mixed system is split into two readers that each build.
features – Only read forecasts carrying these feature key/value pairs.

Returns:

Reader whose read(timestamp) returns {component id: window array}.

Return type:

ForecastReader

Raises:

InvalidParameterError – Raised by the store if no forecasts match, if the matched forecasts do not share one window timeline, or if they disagree about how their timestamps are spelled — see zoneless.

Examples

>>> reader = system.build_forecast_reader(timedelta(hours=1), name="active_power")
>>> for timestamp in reader.timestamps:
...     windows = reader.read(timestamp)

Drive work per distinct window rather than per component:

>>> components_by_slot = reader.components_by_slot()
>>> for slot, window in reader.read_slots(timestamp).items():
...     for component_id in components_by_slot[slot]:
...         ...

See also

build_time_series_reader

time_series_transaction(auto_flush_threshold: int = 10000, auto_flush_bytes: int = 268435456) → Generator[TimeSeriesTransaction, None, None]¶

Open a transaction that batches every time series operation called on it.

Batching lets the store pay one catalog transaction for the whole block instead of one per call, which matters when adding many arrays. The transaction is also the unit of rollback: if the block raises, everything it did is undone.

Call the time series methods on the yielded transaction. A System method called inside the block runs on its own and sees only committed data — excluding data this block has staged but not yet flushed.

Parameters:

auto_flush_threshold (int) – Number of staged additions the transaction buffers before writing them out on its own. The default keeps each bulk write near the store’s preferred chunk size; flushed work stays inside the transaction and rolls back with it, so there is rarely a reason to change this.
auto_flush_bytes (int) – Bytes of staged array data the transaction buffers before writing them out, bounding the block’s memory use no matter how large individual arrays are. Whichever limit is reached first triggers the flush.

Returns:

Handle exposing the time series methods for this batch.

Return type:

TimeSeriesTransaction

Examples

>>> with system.time_series_transaction() as txn:
...     txn.add_time_series(ts1, gen1)
...     txn.add_time_series(ts2, gen1)

open_metadata_store() → Generator[Store, None, None]¶

Open a transactional context for supplemental attribute metadata.

Any failure restores the supplemental attribute associations that were stored on entry and rolls back in-memory supplemental attribute cache updates.

Returns:: The store that holds the supplemental attribute associations.
Return type:: Store

Examples

>>> with system.open_metadata_store():
...     system.add_supplemental_attribute(bus, geo1)
...     system.add_supplemental_attribute(bus, geo2)

serialize_system_attributes() → dict[str, Any]¶: Allows subclasses to serialize attributes at the root level.

deserialize_system_attributes(data: dict[str, Any]) → None¶

Allows subclasses to deserialize attributes stored in the JSON at the root level.

The method should modify self with its custom attributes in data.

handle_data_format_upgrade(data: dict[str, Any], from_version: str | None, to_version: str | None) → None¶

Allows subclasses to upgrade data models.

The parameter data contains the full contents of the serialized JSON file. The method should modify the data models in-place.

merge_system(other: System) → None¶: Merge the contents of another system into this one.

property data_format_version: str | None¶: Return the data format version of the component models.

property name: str | None¶: Return the name of the system.

property description: str | None¶: Return the description of the system.

property label: str¶: Provides a description of the system.

property time_series: TimeSeriesManager¶: Return the time series manager.

property uuid: UUID¶: Return the UUID of the system.

get_time_series_directory() → Path | None¶: Return the directory containing time series files. Will be none for in-memory time series.

Display a table of components of the specified type.

Parameters:

component_type (Type[Component]) – The type of components to display. If component_type is an abstract type, all matching subtypes will be included.
columns (str | ComponentColumn | tuple | list | bool | None) – Component field name(s) or computed column specifications to include after the Name column. A bare string is treated as one field name. Passing a bool preserves compatibility with legacy positional show_id calls.
show_id (bool) – Whether to include the ID column in the table. Defaults to False.
show_time_series (bool) – Whether to include the Time Series count column in the table. Defaults to False.
show_supplemental (bool) – Whether to include the Supplemental Attributes count column in the table. Defaults to False.
filter_func (Callable | None) – Optional function to filter the displayed components. The function must accept a component as a single argument.

Examples

>>> system.show_components(Generator)  # Shows only names
>>> system.show_components(Bus, "voltage")
>>> system.show_components(Bus, ("name", "voltage"))
>>> system.show_components(Generator, show_time_series=True)
>>> system.show_components(Generator, show_supplemental=True)