finam.modules.UserControl#

class finam.modules.UserControl(start, step=datetime.timedelta(days=1))[source]#

Bases: TimeComponent

Component to allow users to step a simulation.

Prompts for input on the console.

Users can just press ENTER to proceed by one step, or enter a target time in ISO format.

+-------------+
|             |
| UserControl | [Counter] -->
|             |
+-------------+

Examples

import datetime as dt
import finam as fm

component = fm.modules.UserControl(
    start=dt.datetime(2000, 1, 1),
    step=dt.timedelta(days=1),
)
Parameters:
  • start (datetime) – Starting time. Can be None to retrieve it from linked components.

  • step (timedelta or relativedelta, optional) – The component’s time step. Default 1 day.

Attributes:
connector

The component’s tools.ConnectHelper.

inputs

dict: The component’s inputs.

logger

Logger for this component.

logger_name

Logger name derived from base logger name and class name.

metadata

The component’s metadata.

name

Component name.

next_time

The component’s predicted simulation time of the next pulls.

outputs

dict: The component’s outputs.

status

The component’s current status.

time

The component’s current simulation time.

uses_base_logger_name

Whether this class has a base_logger_name attribute.

Methods

connect(start_time)

Connect exchange data and metadata with linked components.

create_connector([pull_data, in_info_rules, ...])

Initialize the component's tools.ConnectHelper.

finalize()

Finalize and clean up the component.

initialize()

Initialize the component.

try_connect(start_time[, exchange_infos, ...])

Exchange the info and data with linked components.

update()

Update the component by one time step.

validate()

Validate the correctness of the component's settings and coupling.

with_name(name)

Renames the component and returns self.

property time#

The component’s current simulation time.

property next_time#

The component’s predicted simulation time of the next pulls.

Can be None if the component has no inputs.

_initialize()[source]#

Initialize the component.

Components must overwrite this method. After the method call, the component’s inputs and outputs must be available.

_connect(start_time)[source]#

Connect exchange data and metadata with linked components.

Components must overwrite this method.

Parameters:

start_time (datetime) – The composition’s starting time. Can be before the component’s actual time.

Should be passed to try_connect() calls.

_validate()[source]#

Validate the correctness of the component’s settings and coupling.

Components should overwrite this method.

_update()[source]#

Update the component by one time step. Push new values to outputs.

Components must overwrite this method.

_finalize()[source]#

Finalize and clean up the component.

Components should overwrite this method.

__getitem__(name)#

Get an input or output by name. Implements access through square brackets.

Allows for the use of comp["Name"] as shortcut for comp.inputs["Name"] and comp.outputs["Name"].

Requires that the name does not appear in inputs as well as outputs.

Returns:

The slot with the given name

Return type:

IInput or IOutput

Raises:

KeyError – If the name occurs in the inputs as well as the outputs, or neither in the inputs nor the outputs.

connect(start_time)#

Connect exchange data and metadata with linked components.

The method can be called multiple times if there are failed pull attempts.

After each method call, the component should have status ComponentStatus.CONNECTED if connecting was completed, ComponentStatus.CONNECTING if some but not all required initial input(s) could be pulled, and ComponentStatus.CONNECTING_IDLE if nothing could be pulled.

Parameters:

start_time (datetime) – The composition’s starting time. Can be before the component’s actual time.

property connector#

The component’s tools.ConnectHelper.

See also create_connector() and try_connect().

create_connector(pull_data=None, in_info_rules=None, out_info_rules=None, cache=True)#

Initialize the component’s tools.ConnectHelper.

See also try_connect(), connector and ConnectHelper for details.

Parameters:

Examples

The following examples show the usage of this method in _initialize().

Simple usage if no input data or any metadata from connected components is required:

self.inputs.add(name="In", time=self.time, grid=fm.NoGrid())
self.outputs.add(name="Out", time=self.time, grid=fm.NoGrid())
self.create_connector()

To pull specific inputs, use pull_data like this:

self.inputs.add(name="In1", time=self.time, grid=fm.NoGrid())
self.inputs.add(name="In2", time=self.time, grid=fm.NoGrid())

self.create_connector(pull_data=["In1", "In2"])

With the in_info_rules and out_info_rules, metadata can be transferred between coupling slots.

Here, the metadata for an output is taken from an input:

self.inputs.add(name="In", time=self.time, grid=None, units=None)
self.outputs.add(name="Out")

self.create_connector(
    out_info_rules={
        "Out": [
            fm.tools.FromInput("In")
        ]
    }
)

The Info object for output Out will be created and pushed automatically in try_connect() as soon as the metadata for In becomes available.

Here, the metadata of an output is composed from the metadata of two inputs and a user-defined value:

self.inputs.add(name="In1", time=self.time, grid=None, units=None)
self.inputs.add(name="In2", time=self.time, grid=None, units=None)
self.outputs.add(name="Out")

self.create_connector(
    out_info_rules={
        "Out": [
            fm.tools.FromInput("In1", ["time", "grid"]),
            fm.tools.FromInput("In2", ["units"]),
            fm.tools.FromValue("source", "FINAM"),
        ]
    }
)

The Info object for output Out would be automatically composed in try_connect() as soon as the infos of both inputs become available. time and grid would be taken from In1, units from In2, and source would be set to "finam".

Rules are evaluated in the given order. Later rules can overwrite attributes set by earlier rules.

finalize()#

Finalize and clean up the component.

After the method call, the component should have status ComponentStatus.FINALIZED.

initialize()#

Initialize the component.

After the method call, the component’s inputs and outputs must be available, and the component should have status ComponentStatus.INITIALIZED.

property inputs#

The component’s inputs.

Type:

dict

property logger#

Logger for this component.

property logger_name#

Logger name derived from base logger name and class name.

property metadata#

The component’s metadata. Will only be called after the connect phase from Composition.metadata.

Components can overwrite this property to add their own specific metadata:

import finam as fm

class MyComponent(fm.Component):

    @property
    def metadata(self):
        # Get the default metadata
        md = super().metadata

        # Add your own metadata
        md["my_field"] = "some value"

        # Return the dictionary
        return md
Returns:

A dict with the following default metadata:
  • name - the component’s name

  • class - the component’s class

  • inputs - dict of metadata for all inputs

  • outputs - dict of metadata for all outputs

Return type:

dict

property name#

Component name.

property outputs#

The component’s outputs.

Type:

dict

property status#

The component’s current status.

try_connect(start_time, exchange_infos=None, push_infos=None, push_data=None)#

Exchange the info and data with linked components.

Values passed by the arguments are cached internally for later calls to the method if the connector was created with cache=True (the default). Thus, it is sufficient to provide only data and infos that became newly available. Giving the same data or infos repeatedly overwrites the cache.

Sets the component’s status according to success of exchange.

See also create_connector(), connector and ConnectHelper for details.

Parameters:
  • start_time (datetime) – the composition’s starting time as passed to Component.try_connect()

  • exchange_infos (dict of [str, Info]) – currently or newly available input data infos by input name

  • push_infos (dict of [str, Info]) – currently or newly available output data infos by output name

  • push_data (dict of [str, array-like]) – currently or newly available output data by output name

update()#

Update the component by one time step. Push new values to outputs.

After the method call, the component should have status ComponentStatus.UPDATED or ComponentStatus.FINISHED.

property uses_base_logger_name#

Whether this class has a base_logger_name attribute. True.

validate()#

Validate the correctness of the component’s settings and coupling.

After the method call, the component should have status ComponentStatus.VALIDATED.

with_name(name)#

Renames the component and returns self.