numerous.sdk package#

Subpackages#

Submodules#

numerous.sdk.utils module

Module contents#

numerous.sdk - develop numerous applications!

The recommended way to develop applications for running on the numerous platform.

class numerous.sdk.Component(uuid, id, type, name, item_class, is_enabled, is_main, input_variables, components, parameters, is_dummy=False)#

Bases: object

A component of a Scenario, containing parameters, input variables, and any subcomponents.

components: dict[str, list[Component]]#

static from_document(path, component, all_components, input_sources_mapping, file_manager_client)#

Return type:: Component

id: str#

input_variables: dict[str, InputVariable]#

is_dummy: bool = False#: The subcomponents of this component.

is_enabled: bool#

is_main: bool#

item_class: list[str]#

name: str#

parameters: dict[str, Parameter]#

type: str#

uuid: str#

class numerous.sdk.InputVariable(uuid, id, path, display_name, source, scale=1.0, offset=0.0)#

Bases: object

display_name: str#

id: str#

property key: str#

offset: float = 0.0#

path: list[str]#

scale: float = 1.0#

source: Union[InputVariableScenarioSource, InputVariableStaticSource]#

uuid: str#

property value: float | None#

class numerous.sdk.Job(id, name, is_main, parameters, time)#

Bases: object

A job in a numerous.sdk.models.scenario.Scenario

static from_document(job_id, data, file_manager_client, status_handler, last_hibernate_time=None)#

Return type:: Job

id: str#

is_main: bool#

name: str#

parameters: dict[str, Parameter]#

time: JobTime#

class numerous.sdk.JobClient(channel, identity, execution_id, config=None)#

Bases: object

The JobClient is the recommended way to connect to the numerous platform.

static channel_options(config)#

Returns the default gRPC channel options.

Return type:: list[tuple[str, Any]]

close(error=None)#

Close the JobClient.

Closes the JobClient’s connection to the numerous platform, immediately terminating any active communication.

Parameters:: error_occured¶ – If True, sets the job status with JobStatus.ERROR.
Return type:: None

This method is idempotent.

static connect(status=JobStatus.RUNNING, message=None)#

Create a JobClient from environment variables.

Return type:: JobClient

Uses the following environment variables:

NUMEROUS_API_SERVER
NUMEROUS_API_PORT
NUMEROUS_API_REFRESH_TOKEN
NUMEROUS_PROJECT
NUMEROUS_SCENARIO
JOB_ID
NUMEROUS_EXECUTION_ID

property file_manager: FileManager#: Access the file manager of the job.

flush_logs()#

Flush all buffered log messages, so that they are sent to the server.

Return type:: None

static from_connection_params(hostname, port, refresh_token, identity, execution_id, config)#

Create a JobClient from connection parameters.

Parameters:

hostname¶ (str) – Hostname of the numerous server
port¶ (str) – gRPC port of the numerous server
refresh_token¶ (str) – Refresh token for the execution.
identity¶ (JobIdentifier) – Contains identity information for the job object and related objects.

Return type:

JobClient

property group: Group | None#

hibernate()#

Hibernate the current job execution

Terminates the Python process, committing the job state to the back-end. The job can be resumed again, and the state can be reloaded to continue.

If the job is scheduled, it will then follow its schedule, and be resumed at the next scheduled resume.

Return type:: None

property job: Job#: The model of the current job.

property job_time: JobTime#: The associated time setup of the job, which contains start and end times of the simulation, and the duration.

property log: Logger#: Return the Job Client logger for the job.

property message: str#

Status message of the job, reported to the platform. Is truncated to at most 32 characters upon setting.

Getting the status message returns a locally cached value.

open_output_file(file_name)#

Parameters:: file_name¶ (str) – Output file name.
Return type:: FileUploadContext
Returns:: An object to be used under with statement.

Example:

>>> with JobClient.connect() as client:
...     with client.open_output_file("output_data") as output_file:
...         output_file.write("First file content")
...         output_file.write_from_file("first_file.csv")
...         output_file.write("Second file content")
...         output_file.write_from_file("second_file.csv")

property parameters: dict[str, numerous.sdk.models.parameter.Parameter]#

property progress: float#

Progress of the job, reported to the platform. Is clamped between 0.0 and 100.0 upon setting.

Getting the progress returns a locally cached value.

property project: Project#: The associated Project configuration, with parameters. It is a lazy property, that will load the scenario data on access.

read_inputs(step, repeat_spacing=None, hibernating_source_action=HibernatingSourceAction.HIBERNATE_SELF, update_progress=False)#

Get an iterator over all input variables in the scenario.

This method can only be called once in the job runtime.

Parameters:

step¶ (float) – The distance (in time) between each read. Zero-order hold is used for values at steps that are not in the original data set.
repeat_spacing¶ (Optional[float]) – The distance (in time) between the last data point of the source data set, and the repeated first data point when repeating.

Return type:

InputReader

Returns an InputReader, that can be iterated over.

Example:

>>> result = 0
>>> with JobClient.connect() as job_client:
...     for t, data in job_client.read_inputs(step=3600.0):
...          result += data["component_name.input_variable_id"]

property reader: Reader#: Data reader for the job.

property scenario: Scenario#

The associated Scenario configuration, with jobs, components, parameters and input variables.

It is a lazy property, that will load the scenario data on access.

set_hibernate_callback(callback)#

Set a custom function to be called if the job is hibernated.

Return type:: None

set_terminate_callback(callback)#

Set a custom function to be called if the job is terminated.

Return type:: None

property state: JobState#

The job state, which can be persisted across hibernations.

It is a lazy property, that will load any remote state on access.

property status: JobStatus#

Status of the job, reported to the platform.

Getting the status returns a locally cached value.

property writer: Writer#: Data writer for the job.

class numerous.sdk.JobState(spm_client, identity, execution_id)#

Bases: object

JobState enables jobs to store state remotely, and cache it locally. This is useful for keeping track of values across hibernations, or if an error has restarted the job.

commit()#

Commit the state to the server. This is useful for batching state updates.

Return type:: None

get(key, default=None)#

Get the state value keyed by key from the JobState.

Parameters:

key¶ (str) – The key to look up in the state.
default¶ (Optional[Any]) – A default value to return if the key does not exist.

Return type:

Any

Returns:

The state value at the key if it exists, otherwise the default, or None if no default is specified.

set(key, value, commit=True)#

Set the state value keyed by key to the value value in the JobState, and optionally commit the state to the server.

If commit is True the new state resulting from setting the value, will be comitted to the server.

Parameters:

key¶ (str) – The key to set in the state.
value¶ (Any) – The value to set.
commit¶ (bool) – Whether or not to commit the resulting state to the server.

Return type:

None

unset(key, commit=True)#

Unset the state value keyed by key in the JobState, and optionally commit the state to the server.

If commit is True the new state resulting from unsetting the value, will be comitted to the server.

Parameters:

key¶ (str) – The key to unset in the state.
commit¶ (bool) – Whether or not to commit the resulting state to the server.

Return type:

None

class numerous.sdk.JobStatus(value)#

Bases: str, Enum

An enumeration.

ERROR = 'error'#

FAILED = 'failed'#

FINISHED = 'finished'#

HIBERNATING = 'hibernating'#

INITIALIZING = 'initializing'#

RUNNING = 'running'#

STOPPED = 'stopped'#

class numerous.sdk.JobTime(run_mode, start, end, duration, status_handler, _last_hibernate_time=None)#

Bases: object

Contains information about the job runtime.

current_time: Optional[datetime] = None#

duration: Optional[timedelta]#

property elapsed: timedelta | None#

end: Optional[datetime]#

static from_document(job_data, last_hibernate_time, status_handler)#

Return type:: JobTime

range(step=1.0, start=None, update_progress=False)#

Range over the runtime of the job, from start to end, inclusive. If end is not defined, range forever. If the job is resumed from hibernation, the start time will be the job time before hibernation.

Parameters:

step¶ (Union[timedelta, float]) – The step in time between yielded time value. If it is a float it is interpreted as seconds.
start¶ (Optional[datetime]) – If specified, it is used instead of the job start time.

Return type:

Generator[datetime, None, None]

run_mode: RunMode#

start: datetime#

start_to_time(t)#

Return type:: timedelta

status_handler: StatusHandler#

class numerous.sdk.Parameter(id, uuid, display_name, type, value)#

Bases: object

display_name: str#

static from_document(data, file_manager_client)#

Return type:: Parameter

id: str#

type: str#

uuid: str#

value: Any#

class numerous.sdk.RunMode(value)#

Bases: Enum

An enumeration.

CONTINUOUS = 'continuous'#

DURATION = 'duration'#

class numerous.sdk.Scenario(id, name, components, parameters, jobs, input_sources, input_variables, group_id, optimization=None)#

Bases: object

components: dict[str, Component]#

static from_document(data, file_manager_client, status_handler, optimization=None, last_hibernate_time=None)#

Return type:: Scenario

group_id: str#

id: str#

input_sources: list[InputSource]#

input_variables: list[InputVariable]#

jobs: dict[str, Job]#

name: str#

optimization: Optional[OptimizationConfiguration] = None#

parameters: dict[str, Parameter]#