System¶
- class infrasys.system.System(name: str | None = None, description: str | None = None, auto_add_composed_components: bool = False, con: Connection | None = None, 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.
con (None | sqlite3.Connection) – Users should not pass this. De-serialization (from_json) will pass a Connection.
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_in_memory: Defaults to true.
time_series_read_only: Disables add/remove of time series, defaults to false.
time_series_directory: Location to store time series file, defaults to the system’s tmp directory.
Examples
>>> system = System(name="my_system") >>> system2 = System(name="my_system", time_series_directory="/tmp/scratch")
- 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.
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")
- 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(*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
- change_component_uuid(component: Component) None ¶
Change the component UUID. This is required if you copy a component and attach it to the same system.
- Parameters:
component (Component)
- 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 UUID 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(component: Component) Any ¶
Create a deep copy of the component and all composed components. All attributes, including names and UUIDs, 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
- get_component(component_type: Type[Component], name: str) Any ¶
Return the component with the passed type and name.
- Parameters:
component_type (Type[Component]) – Type of component
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 UUID is not stored.
ISOperationNotAllowed – Raised if there is more than one matching component.
Examples
>>> component = system.get_component_by_label("Bus.bus1")
- get_component_by_uuid(uuid: UUID) Any ¶
Return the component with the input UUID.
- Parameters:
uuid (UUID)
- Raises:
ISNotStored – Raised if the UUID is not stored.
Examples
>>> uuid = UUID("714c8311-8dff-4ae2-aa2e-30779a317d42") >>> component = system.get_component_by_uuid(uuid)
- get_components(*component_type: Type[Component], filter_func: Callable | None = None) Iterable[Any] ¶
Return the components with the passed type(s) and that optionally match filter_func.
- Parameters:
component_type (Type[Component]) – 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)
- 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:
See also
- 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:
Examples
>>> components = system.list_parent_components(bus) >>> print(f"These components are connected to {bus.label}: ", " ".join(components))
See also
- 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
- 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) Any ¶
Remove the component from the system and return it.
- 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.
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) Any ¶
Remove the component with component_type and name from the system and return it.
- 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_uuid(uuid: UUID, cascade_down: bool = True, force: bool = False) Any ¶
Remove the component with uuid from the system and return it.
- Parameters:
uuid (UUID)
cascade_down (bool) – Refer
remove_component()
.force (bool) – Refer
remove_component()
.
- Raises:
ISNotStored – Raised if the UUID is not stored in the system.
Examples
>>> uuid = UUID("714c8311-8dff-4ae2-aa2e-30779a317d42") >>> generator = system.remove_component_by_uuid(uuid)
- 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, *components: Component, **user_attributes: Any) None ¶
Store a time series array for one or more components.
- Parameters:
time_series (TimeSeriesData) – Time series data to store.
components (Component) – Add the time series to all of these components.
user_attributes (Any) – Key/value pairs to store with the time series data. Must be JSON-serializable.
- Raises:
ISAlreadyAttached – Raised if the variable name and user attributes match any time series already attached to one of the components.
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], variable_name="active_power", start_time=datetime(year=2030, month=1, day=1), resolution=timedelta(hours=1), ) >>> system.add_time_series(ts, gen1, gen2)
- copy_time_series(dst: Component, src: Component, name_mapping: dict[str, str] | None = None) None ¶
Copy all time series from src to dst.
- Parameters:
dst (Component) – Destination component
src (Component) – Source component
name_mapping (dict[str, str]) – Optionally map src names to different dst names. If provided and src has a time_series with a name not present in name_mapping, that time_series will not copied. If name_mapping is nothing then all time_series will be copied with src’s names.
Notes
name_mapping is currently not implemented.
Examples
>>> gen1 = system.get_component(Generator, "gen1") >>> gen2 = system.get_component(Generator, "gen2") >>> system.copy_time_series(gen1, gen2)
- get_time_series(component: ~infrasys.component.Component, variable_name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] = <class 'infrasys.time_series_models.SingleTimeSeries'>, start_time: ~datetime.datetime | None = None, length: int | None = None, **user_attributes: str) Any ¶
Return a time series array.
- Parameters:
component (Component) – Component to which the time series must be attached.
variable_name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData]) – Optional, search for time series with this 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.
user_attributes (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
- has_time_series(component: ~infrasys.component.Component, variable_name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] = <class 'infrasys.time_series_models.SingleTimeSeries'>, **user_attributes: str) bool ¶
Return True if the component has time series matching the inputs.
- Parameters:
component (Component) – Component to check for matching time series.
variable_name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData]) – Optional, search for time series with this type.
user_attributes (str) – Optional, search for time series with these attributes.
- list_time_series(component: ~infrasys.component.Component, variable_name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] = <class 'infrasys.time_series_models.SingleTimeSeries'>, start_time: ~datetime.datetime | None = None, length: int | None = None, **user_attributes: ~typing.Any) list[TimeSeriesData] ¶
Return all time series that match the inputs.
- Parameters:
component (Component) – Component to which the time series must be attached.
variable_name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData]) – Optional, search for time series with this 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.
user_attributes (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_metadata(component: ~infrasys.component.Component, variable_name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] = <class 'infrasys.time_series_models.SingleTimeSeries'>, **user_attributes: ~typing.Any) list[TimeSeriesMetadata] ¶
Return all time series metadata that match the inputs.
- Parameters:
component (Component) – Component to which the time series must be attached.
variable_name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData]) – Optional, search for time series with this type.
user_attributes (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(*components: ~infrasys.component.Component, variable_name: str | None = None, time_series_type: ~typing.Type[~infrasys.time_series_models.TimeSeriesData] = <class 'infrasys.time_series_models.SingleTimeSeries'>, **user_attributes: ~typing.Any) None ¶
Remove all time series arrays attached to the components matching the inputs.
- Parameters:
components (Component) – Affected components
variable_name (str | None) – Optional, search for time series with this name.
time_series_type (Type[TimeSeriesData]) – Optional, search for time series with this type.
user_attributes (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")
- 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.
- 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.