upstage package#

Subpackages#

Submodules#

upstage.actor module#

This file contains the fundamental Actor class for UPSTAGE.

class Actor(*, name: str, debug_log: bool = True, **states: Any)#

Bases: SettableEnv, NamedUpstageEntity

Actors perform tasks and are composed of states.

You can subclass, but do not overwrite __init_subclass__.

Always super().__init__().

Parameters:

  • name (str) – The name of the actor. This is a required attribute.

  • debug_log (bool, optional) – Run the debug logger which captures runtime information about the actor.

  • **states – Keyword arguments to set the values of the actor’s states.

  • Raises

  • ------

  • ValueError – If the keys of the states passed as keyword arguments do not match the names of the actor’s states.

activate_linear_state(*, state: str, rate: float, task: Task) → None#

Shortcut for activating a LinearChangingState.

Parameters:

  • state (str) – The name of the LinearChangingState to set as active.

  • rate (float) – The rate of the change

  • task (Task) – The task that is activating

activate_location_state(*, state: str, speed: float, waypoints: list[GeodeticLocation] | list[CartesianLocation], task: Task) → None#

Shortcut for activating a (Cartesian|Geodetic)LocationChangingState.

Parameters:

  • state (str) – State name

  • speed (float) – The speed to move at

  • waypoints (LOC_LIST) – Waypoints to move over

  • task (Task) – The task that the state is activated during.

activate_mimic_state(*, self_state: str, mimic_state: str, mimic_actor: Actor, task: Task) → None#

Activate a state to mimic a state on another actor.

Parameters:

  • self_state (str) – State name to be the mimic

  • mimic_state (str) – State on the other actor to be mimiced

  • mimic_actor (Actor) – The other actor.

  • task (Task) – The task during which the state is mimiced.

activate_state(*, state: str, task: Task, **kwargs: Any) → None#

Set a state as active.

Note

This method is used by the tasks for activating states they use/modify.

TODO: on init, create activate_<statename> methods that type-hint the inputs

Parameters:

  • state (str) – The name of the state to set as active.

  • task (Task) – The task that is activating the state.

  • **kwargs (Any) – key:values as kwargs for the state activation.

add_task_network(network: TaskNetwork) → None#

Add a task network to the actor.

Parameters:

network (TaskNetwork) – The task network to add to the actor.

clear_knowledge(name: str, caller: str | None = None) → None#

Clear a knowledge value.

Raises an error if the knowledge does not exist.

Parameters:

  • name (str) – The name of the knowledge item to clear.

  • caller (str) – The name of the Task that called the method. Used for debug logging purposes.

clear_task_queue(network_name: str) → None#

Empty the actor’s task queue.

This will cause the task network to be used for task flow.

Parameters:

network_name (str) – The name of the network to clear the task queue.

clone(new_env: MockEnvironment | None = None, knowledge: dict[str, Any] | None = None, **new_states: Any) → Self#

Clones an actor and assigns it a new environment.

Note

This function is useful when testing if an actor can accomplish a task.

In general, cloned actor are referred to as understudy to keep with the theater analogy.

The clones’ names are appended with the label '[CLONE #]' where '#' indicates the number of clones of the actor.

Parameters:

  • new_env (Optional[MockEnvironment], optional) – Environment for cloning. Defaults to None.

  • knowledge (Optional[dict[str, Any]], optional) – Knowledge for the clone. Defaults to None.

  • new_states (Any) – New states to add to the actor when cloning.

Returns:

The cloned actor of the same type

Return type:

Actor

create_knowledge_event(*, name: str, rehearsal_time_to_complete: float = 0.0) → Event#

Create an event and store it in knowledge.

Useful for creating simple hold points in Tasks that can be succeeded by other processes.

Example

>>> def task(self, actor):
>>>     evt = actor.create_knowledge_event(name="hold")
>>>     yield evt
>>>     ... # do things
...
>>> def other_task(self, actor):
>>>     if condition:
>>>         actor.succeed_knowledge_event(name="hold")
Parameters:

  • name (str) – Name of the knowledge slot to store the event in.

  • rehearsal_time_to_complete (float, optional) – The event’s expected time to complete. Defaults to 0.0.

Returns:

The event to yield on

Return type:

Event

deactivate_all_mimic_states(*, task: Task) → None#

Deactivate all mimicking states in the actor for a given task.

Parameters:

task (Task) – The task where states are mimicking others.

deactivate_all_states(*, task: Task) → None#

Deactivate all states in the actor for a given task.

Parameters:

task (Task) – The task that is deactivating the states.

deactivate_mimic_state(*, self_state: str, task: Task) → None#

Deactivate a mimicking state.

Parameters:

  • self_state (str) – State name

  • task (Task) – Task it’s running in.

deactivate_state(*, state: str, task: Task) → None#

Deactivate a specific state.

Parameters:

  • state (str) – The name of the state to deactivate.

  • task (Task) – The task that is deactivating the state.

deactivate_states(*, states: str | Iterable[str], task: Task) → None#

Set a list of active states to not active.

Parameters:

  • states (str | Iterable[str]) – The names of the states to deactivate.

  • task (Task) – The task that is deactivating the states.

delete_task_network(network_id: Any) → None#

Deletes a task network reference.

Be careful, the network may still be running!

Do any interruptions on your own.

Parameters:

network_id (Any) – Typically a string for the network name.

get_active_state_data(state_name: str, without_update: bool = False) → dict[str, Any]#

Get the data for a specific state.

Parameters:

  • state_name (str) – The name of the state for which to retrieve the data.

  • without_update (bool) – Whether or not to update the state to the current sim time. Defaults to True

Returns:

The state data.

Return type:

dict[str, Any]

get_all_task_queues() → dict[str, list[str]]#

Get the task queues for all running networks.

Returns:

Task names, keyed on task network name.

Return type:

dict[str, list[str]]

get_knowledge(name: str, must_exist: bool = False) → Any#

Get a knowledge value from the actor.

Parameters:

  • name (str) – The name of the knowledge

  • must_exist (bool) – Raise an error if the knowledge isn’t present. Defaults to false.

Returns:

The knowledge value. None if the name doesn’t exist.

Return type:

Any

get_log() → list[str]#

Get the debug log.

Returns:

List of log messages.

Return type:

list[str]

get_next_task(network_name: str) → None | str#

Return the next task the actor has been told if there is one.

This does not clear the task, it’s information only.

Parameters:

network_name (str) – The name of the network

Returns:

The name of the next task, None if no next task.

Return type:

None | str

get_nucleus() → TaskNetworkNucleus#

Return the actor’s nucleus.

Returns:

The nucleus on the actor.

Return type:

TaskNetworkNucleus

get_remaining_waypoints(location_state: str) → list[GeodeticLocation] | list[CartesianLocation]#

Convenience method for interacting with LocationChangingStates.

Primary use case is when restarting a Task that has a motion element to allow updating waypoint knowledge easily.

Parameters:

location_state (str) – The name of the <LocationChangingState>

Returns:

List of waypoints yet to be reached

Return type:

list[Location]

get_running_task(network_name: str) → TaskData | None#

Return name and process reference of a task on this Actor’s task network.

Useful for finding a process to call interrupt() on.

Parameters:

network_name (str) – Network name.

Returns:

Dataclass of name and process for the current task.

{“name”: Name, “process”: the Process simpy is holding.}

Return type:

TaskData

get_running_tasks() → dict[str, TaskData]#

Get all running task data.

Returns:

Dictionary of all running tasks.

Keyed on network name, then {“name”: Name, “process”: …}

Return type:

dict[str, dict[str, TaskData]]

get_task_queue(network_name: str) → list[str]#

Get the actor’s task queue on a single network.

Parameters:

network_name (str) – The network name

Returns:

List of task names in the queue

Return type:

list[str]

has_task_network(network_id: Any) → bool#

Test if a network id exists.

Parameters:

network_id (Any) – Typically a string for the network name.

Returns:

If the task network is on this actor.

Return type:

bool

interrupt_network(network_name: str, **interrupt_kwargs: Any) → None#

Interrupt a running task network.

Parameters:

  • network_name (str) – The name of the network.

  • interrupt_kwargs (Any) – kwargs to pass to the interrupt.

log(msg: str | None = None) → list[str] | None#

Add to the log or return it.

Only adds to log if debug_logging is True.

Parameters:

msg (str, Optional) – The message to log.

Returns:

The log if no message is given. None otherwise.

Return type:

list[str] | None

rehearse_network(network_name: str, task_name_list: list[str], knowledge: dict[str, Any] | None = None, end_task: str | None = None) → Self#

Rehearse a network on this actor.

Supply the network name, the tasks to rehearse from this state, and any knowledge to apply to the cloned actor.

Parameters:

  • network_name (str) – Network name

  • task_name_list (list[str]) – Tasks to rehearse on the network.

  • knowledge (dict[str, Any], optional) – knowledge to give to the cloned actor. Defaults to None.

  • end_task (str, optional) – A task to end on once reached.

Returns:

The cloned actor after rehearsing the network.

Return type:

Actor

set_knowledge(name: str, value: Any, overwrite: bool = False, caller: str | None = None) → None#

Set a knowledge value.

Raises an error if the knowledge exists and overwrite is False.

Parameters:

  • name (str) – The name of the knowledge item.

  • value (Any) – The value to store for the knowledge.

  • overwrite (bool, Optional) – Allow the knowledge to be changed if it exits. Defaults to False.

  • caller (str, Optional) – The name of the object that called the method.

set_task_queue(network_name: str, task_list: list[str]) → None#

Initialize an actor’s empty task queue.

Parameters:

  • network_name (str) – Task Network name

  • task_list (list[str]) – List of task names to queue.

Raises:

SimulationError – _description_

start_network_loop(network_name: str, init_task_name: str | None = None) → None#

Start a task network looping/running on an actor.

If no task name is given, it will default to following the queue.

Parameters:

  • network_name (str) – Network name.

  • init_task_name (str, optional) – Task to start with. Defaults to None.

property state_values: dict[str, Any]#

Get the state names and values.

Returns:

State name:value pairs.

Return type:

dict[str, Any]

property states: tuple[str, ...]#

Get the names of the actor’s states.

Returns:

State names

Return type:

tuple[str]

succeed_knowledge_event(*, name: str, **kwargs: Any) → None#

Succeed and clear an event stored in the actor’s knowledge.

See “create_knowledge_event” for usage example.

Parameters:

  • name (str) – Event knowledge name.

  • **kwargs (Any) – Any payload to send to the event. Defaults to None

suggest_network_name(factory: TaskNetworkFactory) → str#

Deconflict names of task networks by suggesting a new name.

Used for creating multiple parallel task networks.

Parameters:

factory (TaskNetworkFactory) – The factory from which you will create the network.

Returns:

The network name to use

Return type:

str

class TaskData(name: str, process: simpy.events.Process)#

Bases: object

name: str#
process: Process#

upstage.api module#

The elements in the UPSTAGE Application Programmable Interface.

upstage.base module#

Base classes and exceptions for UPSTAGE.

class EarthProtocol(*args, **kwargs)#

Bases: Protocol

Protocol for defining an earth model interface.

distance(loc1: tuple[float, float], loc2: tuple[float, float], units: str) → float#

Get the distance between two lat/lon (degrees) points.

distance_and_bearing(loc1: tuple[float, float], loc2: tuple[float, float], units: str) → tuple[float, float]#

Get the distance between two lat/lon (degrees) points.

geo_linspace(start: tuple[float, float], end: tuple[float, float], num_segments: int) → list[tuple[float, float]]#

Get evenly spaced coordinates between lat/lon pairs.

lla2ecef(locs: list[tuple[float, float, float]]) → list[tuple[float, float, float]]#

Get ECEF coordinates from lat lon alt.

point_from_bearing_dist(point: tuple[float, float], bearing: float, distance: float, distance_units: str = 'nmi') → tuple[float, float]#

Get a lat/lon in degrees from a point, bearing, and distance.

class EnvironmentContext(initial_time: float = 0.0, random_seed: int | None = None, random_gen: Any | None = None)#

Bases: object

A context manager to create a safe, globally (in context) referenceable environment and data.

The environment created is of type simpy.Environment

This also sets context variables for actors, entities, and the stage.

Usage:
>>> with EnvironmentContext(initial_time=0.0) as env:
>>>    env.run(until=3.0)

This context manager is meant to be paired with inheritors of UpstageBase.

that provides access to the context variables created in this manager.

>>> class SimData(UpstageBase):
>>>     ...
>>>
>>> with EnvironmentContext(initial_time=0.0) as env:
>>>     data = SimData()
>>>     assert data.env is env

You may also provide a random seed, and a default Random() will be created with that seed.

>>> with EnvironmentContext(random_seed=1234986) as env:
>>>    UpstageBase().stage.random.uniform(1, 3)
...    2.348057489610457

Or your own RNG:

>>> rng = Random(1234986)
>>> with EnvironmentContext(random_gen=rng) as env:
>>>    UpstageBase().stage.random.uniform(1, 3)
...    2.348057489610457
class MockEnvironment(now: float)#

Bases: object

A fake environment that holds the now property and all-caps attributes.

classmethod mock(env: Environment | MockEnvironment) → MockEnvironment#

Create a mock environment from another environment.

Parameters:

env (SimpyEnv | MockedEnvironment) – The simpy environments

Returns:

The mocked environment (time only)

Return type:

MockEnvironment

classmethod run(until: float | int) → None#

Method stub for playing nice with rehearsal.

Parameters:

until (float | int) – Placeholder

exception MotionAndDetectionError(message: str, time: float | None = None)#

Bases: SimulationError

A simulation error raised during motion detection.

class NamedUpstageEntity#

Bases: UpstageBase

A base class for naming entities, and retrieving them.

This creates a record of every instance of a subclass of this class.

Example

>>> class RocketCar(NamedUpstageEntity, entity_groups=["car", "fast"])
>>>     ...
>>> rc = RocketCar()
>>> assert rc in rc.get_entity_group("car")
exception RulesError#

Bases: UpstageError

Raised by the user when a simulation rule is violated.

class SettableEnv(*args: Any, **kwargs: Any)#

Bases: UpstageBase

A mixin class for allowing the instance’s environment to change.

property env: Environment | MockEnvironment#

Get the relevant environment.

Returns:

Real or mocked environment.

Return type:

SimpyEnv | MockEnvironment

exception SimulationError(message: str, time: float | None = None)#

Bases: UpstageError

Raised when a simulation error occurs.

class StageProtocol(*args, **kwargs)#

Bases: Protocol

Protocol for typing the minimum entries in the Stage.

property altitude_units: str#

Units of altitude.

property distance_units: str#

Units of distance.

property intersection_model: Callable[[tuple[float, float, float], tuple[float, float, float], tuple[float, float, float], float, str, EarthProtocol, float | None, list[int] | None], list[CrossingCondition]]#

Callable for geodetic intersections.

property random: Random#

Random number generator.

property stage_model: EarthProtocol#

Model for geodetics.

property time_unit: str#

Time unit, Treated as ‘hr’ if not set.

class UpstageBase#

Bases: object

A base mixin class for everyone.

Provides access to all context variables created by EnvironmentContext.

>>> with EnvironmentContext(initial_time=0.0) as env:
>>>     data = UpstageBase()
>>>     assert data.env is env
property env: Environment#

Return the environment.

Returns:

SimPy environment.

Return type:

SimpyEnv

get_actors() → list[NamedUpstageEntity]#

Return all actors that the director knows.

Returns:

List of actors in the simulation.

Return type:

list[NamedUpstageEntity]

get_all_entity_groups() → dict[str, list[NamedUpstageEntity]]#

Get all entity groups.

Returns:

Entity group names and associated

entities.

Return type:

dict[str, list[NamedUpstageEntity]]

get_entity_group(group_name: str) → list[NamedUpstageEntity]#

Get a single entity group by name.

Parameters:

group_name (str) – The name of the entity group.

Returns:

List of entities in the group.

Return type:

list[NamedUpstageEntity]

property pretty_now: str#

A well-formatted string of the sim time.

Returns:

The sim time

Return type:

str

property stage: StageProtocol#

Return the stage context variable.

Returns:

The stage, as defined in context.

Return type:

StageProtocol

exception UpstageError#

Bases: Exception

Raised when an UPSTAGE error happens or expectation is not met.

add_stage_variable(varname: str, value: Any) → None#

Add a variable to the stage.

Will fail if it already exists.

Parameters:

  • varname (str) – Name of the variable

  • value (Any) – Value to set it as

class dotdict#

Bases: dict

A dictionary that supports dot notation as well as dictionary access notation.

Usage: d = dotdict({‘val1’:’first’}) set attributes: d.val2 = ‘second’ or d[‘val2’] = ‘second’ get attributes: d.val2 or d[‘val2’] would both produce ‘second’

upstage.constants module#

The constant values used by UPSTAGE.

upstage.data_types module#

Data types for common operations. Currently just locations.

class CartesianLocation(x: float, y: float, z: float = 0.0, *, use_altitude_units: bool = False)#

Bases: Location

A location that can be mapped to a 3-dimensional cartesian space.

copy() → CartesianLocation#

Return a copy of the location.

Returns:

CartesianLocation

straight_line_distance(other: object) → float#

Get the straight line distance between this and another location.

Parameters:

other (object) – The other CartesianLocation point.

class GeodeticLocation(lat: float, lon: float, alt: float = 0.0, *, in_radians: bool = False)#

Bases: Location

A Location that can be mapped to the surface of a spheroid (a.k.a. ellipsoid).

More specifically, a Location representing somewhere on an ellipsoid, with Latitude, Longitude, and Altitude that uses the geodetic datum (or geodetic system). Can be used to define a location on Earth or other planetary bodies.

Units for the horizontal datum (i.e., lat and lon) can be Decimal Degrees or Radians, depending on the value of units, the vertical datum (i.e., alt) is assumed to be in meters.

Subtraction represents a great circle distance, NOT a true 3D straight-line distance.

Speeds used for this location type will represent ground speed, which allows the class to ignore solving the exact altitude change in the path.

Units are an input to the location type.

The ellipsoid model must have a .distance(Location1, Location2) method that looks for .lat and .lon.

altitude is 0.0 by default.

lat (latitude) and lon (longitude) are in degrees by default.

if using radians, set in_radians to True.

copy() → GeodeticLocation#

Copy the location.

Returns:

GeodeticLocation

dist_with_altitude(other: GeodeticLocation) → float#

Get the distance between two points with an altitude component.

Parameters:

other (GeodeticLocation) – The other point

Returns:

Distance - pythagorean of great-circle and altitude

Return type:

float

latlon() → tuple[float, float]#

Return a tuple of just lat/lon as degrees.

Returns:

Latitude and longitude in degrees.

Return type:

tuple[float, float]

straight_line_distance(other: object) → float#

Straight-line distance, using ECEF.

This won’t account for horizon.

Parameters:

other (GeodeticLocation) – The other point

Returns:

Distance

Return type:

float

to_degrees() → GeodeticLocation#

Convert to degrees, if already in degrees, return self.

Returns:

GeodeticLocation

to_radians() → GeodeticLocation#

Convert to radians, if already in radians, return self.

Returns:

GeodeticLocation

class Location#

Bases: UpstageBase

An abstract class for representing a location in a space.

copy() → Location#

Copy the location.

straight_line_distance(other: object) → float#

Straight line distances b/w locations.

Parameters:

other (Location) – Another location

upstage.events module#

Classes for UPSTAGE events that feed to simpy.

class All(*events: BaseEvent | Process)#

Bases: MultiEvent

An event that requires all events to succeed before succeeding.

static aggregation_function(times: list[float]) → float#

Aggregate event times for rehearsal.

Parameters:

times (list[float]) – List of rehearsing times.

Returns:

Aggregated (maximum) time.

Return type:

float

static simpy_equivalent(env: Environment, events: list[Event]) → Event#

Return the SimPy version of the UPSTAGE All event.

Parameters:

  • env (SIM.Environment) – SimPy Environment.

  • events (list[SIM.Event]) – List of events.

Returns:

A simpy AllOf event.

Return type:

SIM.Event

class Any(*events: BaseEvent | Process)#

Bases: MultiEvent

An event that requires one event to succeed before succeeding.

static aggregation_function(times: list[float]) → float#

Aggregation function for rehearsal time.

Parameters:

times (list[float]) – List of rehearsal times

Returns:

Aggregated time (the minimum)

Return type:

float

static simpy_equivalent(env: Environment, events: list[Event]) → Event#

Return the SimPy version of the UPSTAGE Any event.

Parameters:

  • env (SIM.Environment) – SimPy Environment.

  • events (list[SIM.Event]) – List of events.

Returns:

A simpy AnyOf event.

Return type:

SIM.Event

class BaseEvent(*, rehearsal_time_to_complete: float = 0.0)#

Bases: UpstageBase

Base class for framework events.

as_event() → Event#

Convert UPSTAGE event to a simpy Event.

Returns:

The upstage event as a simpy event.

Return type:

SIM.Event

calculate_time_to_complete() → float#

Calculate the time elapsed until the event is triggered.

Returns:

The time until the event triggers.

Return type:

float

cancel() → None#

Cancel an event.

property done_rehearsing: bool#

If the event is done rehearsing.

Returns:

bool

is_complete() → bool#

Is the event complete?

Returns:

If it’s complete or not.

Return type:

bool

property now: float#

Current sim time.

Returns:

sim time

Return type:

float

rehearse() → tuple[float, Any | None]#

Run the event in ‘rehearsal’ mode without changing the real environment.

This is used by the task rehearsal functions.

Returns:

The time to complete and the event’s response.

Return type:

tuple[float, Any | None]

property rehearsing: bool#

If the event is rehearsing.

Returns:

bool

class BaseRequestEvent(rehearsal_time_to_complete: float = 0.0)#

Bases: BaseEvent

Base class for Request Events.

Requests are things like Get and Put that wait in a queue.

cancel() → None#

Cancel the Request.

is_complete() → bool#

Test if the request is finished.

Returns:

bool

class Event(rehearsal_time_to_complete: float = 0.0, auto_reset: bool = True)#

Bases: BaseEvent

An UPSTAGE version of the standard SimPy Event.

Returns a planning factor object on rehearsal for user testing against in rehearsals, in case.

When the event is succeeded, a payload can be added through kwargs.

This Event assumes that it might be long-lasting, and will auto-reset when yielded on.

as_event() → Event#

Get the Event as a simpy type.

This resets the event if allowed.

Returns:

SIM.Event

calculate_time_to_complete() → float#

Return the time to complete.

Returns:

Time to complete estimate.

Return type:

float

cancel() → None#

Cancel the event.

Cancelling doesn’t mean much, since it’s still going to be yielded on.

get_payload() → dict[str, Any]#

Get any payload from the call to succeed().

Returns:

The payload left by the succeed() caller.

Return type:

dict[str, Any]

is_complete() → bool#

Is the event done?

Returns:

bool

rehearse() → tuple[float, Any]#

Run the event in ‘trial’ mode without changing the real environment.

Returns:

The time to complete and the event’s response.

Return type:

tuple[float, Any]

reset() → None#

Reset the event to allow it to be held again.

succeed(**kwargs: Any) → None#

Succeed the event and store any payload.

Parameters:

**kwargs (Any) – key:values to store as payload.

class FilterGet(get_location: FilterStore, filter: Callable[[Any], bool], rehearsal_time_to_complete: float = 0.0)#

Bases: Get

A Get for a FilterStore.

class Get(get_location: Store | Container, *get_args: Any, rehearsal_time_to_complete: float = 0.0, **get_kwargs: Any)#

Bases: BaseRequestEvent

Wrap the simpy Get event.

Event that gets an object from a simpy store or gets an amount from a container.

as_event() → ContainerGet | StoreGet#

Convert get to a simpy Event.

Returns:

ContainerGet | StoreGet

calculate_time_to_complete() → float#

Calculate time elapsed until the event is triggered.

Returns:

Estimated time until the event triggers.

Return type:

float

get_value() → Any#

Get the value returned when the request is complete.

Returns:

The amount or item requested.

Return type:

Any

rehearse() → tuple[float, Any]#

Mock the event to test if it is feasible.

Note

The function does not fully test the conditions to satisfy the get request, but this method can be called as part of a more complex rehearse run.

Returns:

The time it took to do the request Any: The value of the request.

Return type:

float

class MultiEvent(*events: BaseEvent | Process)#

Bases: BaseEvent

A base class for evaluating multiple events.

Note

Subclasses of MultiEvent must define these methods:

  • aggregation_function: Callable[[list[float]], float]

  • simpy_equivalent: simpy.Event

For an example, refer to Any and All.

static aggregation_function(times: list[float]) → float#

Aggregate event times to one single time.

Parameters:

times (list[float]) – Event rehearsal times

Returns:

The aggregated time

Return type:

float

as_event() → Event#

Convert the UPSTAGE event to simpy.

Returns:

typically an Any or All

Return type:

SIM.Event

calc_time_to_complete_with_sub() → tuple[float, dict[BaseEvent, float]]#

Compute time required for MultiEvent and get sub-event times.

Returns:

Aggregate and individual times.

Return type:

tuple[float, dict[BaseEvent, float]]

calculate_time_to_complete() → float#

Compute time required to complete the multi-event.

Parameters:

return_sub_events (bool, Optional) – Whether to return all times or not. Defaults to False.

cancel() → None#

Cancel the multi event and propagate it to the sub-events.

rehearse() → tuple[float, Any]#

Run the event in ‘trial’ mode without changing the real environment.

Returns:

The time to complete and the event’s response.

Return type:

tuple[float, Any]

Note

This is used by the task rehearsal functions.

static simpy_equivalent(env: Environment, events: list[Event]) → Event#

Return the simpy equivalent event.

Parameters:

  • env (SIM.Environment) – The SimPy environment.

  • events (list[BaseEvent]) – Events to turn into multi-event.

Returns:

The aggregate event.

Return type:

SIM.Event

class Put(put_location: Container | Store, put_object: float | int | Any, rehearsal_time_to_complete: float = 0.0)#

Bases: BaseRequestEvent

Wrap the simpy Put event.

This is an event that puts an object into a simpy store or puts an amount into a container.

as_event() → ContainerPut | StorePut#

Convert event to a simpy Event.

Returns:#

simpy.events.Event

Put request as a simpy event.

class ResourceHold(resource: Resource, *resource_args: Any, rehearsal_time_to_complete: float = 0.0, **resource_kwargs: Any)#

Bases: BaseRequestEvent

Wrap the simpy request resource event.

This manages getting and giving back all in one object.

Example

>>> resource = simpy.Resource(env, capacity=1)
>>> hold = ResourceHold(resource)
>>> # yield on the hold to get it
>>> yield hold
>>> # now that you have it, do things..
>>> # give it back
>>> yield hold
>>> ...
as_event() → Request | Release#

Create the simpy event for the right state of Resource usage.

Returns:

The simpy event.

Return type:

Request | Release

calculate_time_to_complete() → float#

Time to complete, based on waiting for getting or giving back.

Returns:

Time

Return type:

float

class Wait(timeout: float | int, rehearsal_time_to_complete: float | int | None = None)#

Bases: BaseEvent

Wait a specified or random uniformly distributed amount of time.

Return a timeout. If time is a list of length 2, choose a random time between the interval given.

Rehearsal time is given by the maximum time of the interval, if given.

Parameters:

timeout (int, float, list, tuple) – Amount of time to wait. If it is a list or a tuple of length 2, a random uniform value between the two values will be used.

as_event() → Timeout#

Cast Wait event as a simpy Timeout event.

Returns:

SIM.Timeout

cancel() → None#

Cancel the timeout.

There’s no real meaning to cancelling a timeout. It sits in simpy’s queue either way.

classmethod from_random_uniform(low: float | int, high: float | int, rehearsal_time_to_complete: float | int | None = None) → Wait#

Create a wait from a random uniform time.

Parameters:

  • low (float) – Lower bounds of random draw

  • high (float) – Upper bounds of random draw

  • rehearsal_time_to_complete (float | int, optional) – The rehearsal time to complete. Defaults to None - meaning the random value drawn.

Returns:

The timeout event

Return type:

Wait

upstage.math_utils module#

This module contains math utility functions to avoid numpy.

upstage.nucleus module#

The file contains the Nucleus features of UPSTAGE.

class NucleusInterrupt(name: str, value: Any)#

Bases: object

A data container for interrupting nucleus events.

class TaskNetworkNucleus(*, actor: Actor)#

Bases: object

The nucleus, for state-based task network signaling.

add_network(network_name: str | TaskNetwork, watch_states: list[str]) → None#

Add a network to the nucleus for state management.

Parameters:

  • network_name (str | TaskNetwork) – A task network that works on this nucleus/actor

  • watch_states (list[str]) – States that - when changed - cause the network to change.

remove_network(network_name: str | TaskNetwork) → None#

Remove a network from nucleus.

Parameters:

network_name (str | TaskNetwork) – A task network that works on this nucleus/actor

send_change(state_name: str, state_value: Any) → None#

Send a change notification for a given state.

Parameters:

  • state_name (str) – The state’s name.

  • state_value (Any) – The value of the state.

upstage.state_sharing module#

States that enable sharing between tasks.

class SharedLinearChangingState(*, default: float | None = None, recording: bool = False)#

Bases: ActiveState[float]

A state whose value changes linearly over time.

Allows for multiple users of the state, keyed on actor and task.

Assumes it’s a non-frozen, floating-point value.

Still activated in the usual way:

>>> class Example(Actor):
>>>     fuel = SharedLinearChangingState()
...
>>> example.activate_state(
>>>     state="fuel",
>>>     task=self,
>>>     rate=actor.fuel_burn,
>>> )
deactivate(instance: Actor, task: Task | None = None) → bool#

Deactivate the state.

Parameters:

  • instance (Actor) – Actor the state is on

  • task (Task) – The task that is stopping its rate.

Returns:

Still active or not

Return type:

bool

upstage.states module#

A state defines the conditions of an actor over time.

class ActiveState(*, default: ST | None = None, frozen: bool = False, valid_types: type | tuple[type, ...] | None = None, recording: bool = False, default_factory: Callable[[], ST] | None = None)#

Bases: State, Generic[ST]

Base class for states that change over time according to some rules.

This class must be subclasses with an implemented active method.

deactivate(instance: Actor, task: Task | None = None) → bool#

Optional method to override that is called when a state is deactivated.

Useful for motion states to deactivate their motion from the motion manager.

Defaults to any deactivation removing active state data.

get_activity_data(instance: Actor) → dict[str, Any]#

Get the data useful for updating active states.

Returns:

A dictionary with the state’s pertinent data. Includes the actor’s

environment current time ('now') and the value of the actor’s state ('state').

Return type:

dict[str, Any]

class CartesianLocationChangingState(*, recording: bool = False)#

Bases: ActiveState[CartesianLocation]

A state that contains the location in 3-dimensional Cartesian space.

Movement is along straight lines in that space.

For activating:
>>> actor.activate_state(
>>>     state=<state name>,
>>>     task=self, # usually
>>>     speed=<speed>,
>>>     waypoints=[
>>>         List of CartesianLocation
>>>     ]
>>> )
deactivate(instance: Actor, task: Task | None = None) → bool#

Deactivate the motion.

Parameters:

  • instance (Actor) – The owning actor

  • task (Task) – The task calling the deactivation.

Returns:

_description_

Return type:

bool

class CommunicationStore(*, mode: str, default: type | None = None, valid_types: type | tuple[type, ...] | None = None)#

Bases: ResourceState[Store]

A State class for communications inputs.

Used for automated finding of communication inputs on Actors by the CommsTransfer code.

Follows the same rules for defaults as ResourceState, except this defaults to a SelfMonitoringStore without any user input.

Only resources inheriting from simpy.Store will work for this state. Capacities are assumed infinite.

The input an Actor needs to receive for a CommunicationStore is a dictionary of:
>>> {
>>>     'kind': <class> (optional)
>>>     'mode': <string> (required)
>>> }

Example

>>> class Worker(Actor):
>>>     walkie = CommunicationStore(mode="UHF")
>>>     intercom = CommunicationStore(mode="loudspeaker")
>>>
>>> worker = Worker(
>>>     name='Billy',
>>>     walkie={'kind': SelfMonitoringStore},
>>> )
class DetectabilityState(*, default: bool = False, recording: bool = False)#

Bases: State[bool]

A state whose purpose is to indicate True or False.

For consideration in the motion manager’s <>LocationChangingState checks.

class GeodeticLocationChangingState(*, recording: bool = False)#

Bases: ActiveState[GeodeticLocation]

A state that contains a location around an ellipsoid that follows great-circle paths.

Requires a distance model class that implements: 1. distance_and_bearing 2. point_from_bearing_dist and outputs objects with .lat and .lon attributes

For activating:

>>> actor.activate_state(
>>>     state=<state name>,
>>>     task=self, # usually
>>>     speed=<speed>,
>>>     waypoints=[
>>>         List of CartesianLocation
>>>     ]
>>> )
deactivate(instance: Actor, task: Task | None = None) → bool#

Deactivate the state.

Parameters:

  • instance (Actor) – The owning actor

  • task (Task) – The task doing the deactivating

Returns:

If the state is all done

Return type:

bool

class LinearChangingState(*, default: ST | None = None, frozen: bool = False, valid_types: type | tuple[type, ...] | None = None, recording: bool = False, default_factory: Callable[[], ST] | None = None)#

Bases: ActiveState[float]

A state whose value changes linearly over time.

When activating:

>>> class Lin(Actor):
>>>     x = LinearChangingState()
>>>
>>> def task(self, actor: Lin):
>>>     actor.activate_state(
>>>         name="x",
>>>         task=self,
>>>         rate=3.2,
>>>     )
class ResourceState(*, default: Any | None = None, valid_types: type | tuple[type, ...] | None = None)#

Bases: State, Generic[T]

A State class for States that are meant to be Stores or Containers.

This should enable easier initialization of Actors with stores/containers or similar objects as states.

No input is needed for the state if you define a default resource class in the class definition and do not wish to modify the default inputs of that class.

The input an Actor needs to receive for a ResourceState is a dictionary of: * ‘kind’: <class> (optional if you provided a default) * ‘capacity’: <numeric> (optional, works on stores and containers) * ‘init’: <numeric> (optional, works on containers) * key:value for any other input expected as a keyword argument by the resource class

Note that the resource class given must accept the environment as the first positional argument. This is to maintain compatibility with simpy.

Example

>>> class Warehouse(Actor):
>>>     shelf = ResourceState[Store](default=Store)
>>>     bucket = ResourceState[Container](
>>>         default=Container,
>>>         valid_types=(Container, SelfMonitoringContainer),
>>>     )
>>>
>>> wh = Warehouse(
>>>     name='Depot',
>>>     shelf={'capacity': 10},
>>>     bucket={'kind': SelfMonitoringContainer, 'init': 30},
>>> )
class State(*, default: ST | None = None, frozen: bool = False, valid_types: type | tuple[type, ...] | None = None, recording: bool = False, default_factory: Callable[[], ST] | None = None)#

Bases: Generic[ST]

The particular condition that something is in at a specific time.

The states are implemented as Descriptors which are associated to upstage.actor.Actor.

Note

The classes that use this descriptor must contain an env attribute.

States are aware

has_default() → bool#

Check if a default exists.

Returns:

bool

property is_recording: bool#

Check if the state is recording.

Returns:

bool

upstage.task module#

Tasks constitute the actions that Actors can perform.

class DecisionTask#

Bases: Task

A task used for decision processes.

make_decision(*, actor: Any) → None#

Define the process this task follows.

rehearse(*, actor: REH_ACTOR, knowledge: dict[str, Any] | None = None, cloned_actor: bool = False, **kwargs: Any) → REH_ACTOR#

Rehearse the task to evaluate its feasibility.

Parameters:

  • actor (Actor) – The actor to rehearse with

  • knowledge (Optional[dict[str, Any]], optional) – Knowledge to add. Defaults to None.

  • cloned_actor (bool, optional) – If the actor is a clone or not. Defaults to False.

  • kwargs (Any) – Kwargs for rehearsal. Kept for consistency to the base class.

Returns:

Cloned actor after rehearsing this task.

Return type:

Actor

rehearse_decision(*, actor: Any) → None#

Define the process this task follows.

run(*, actor: Actor) → Generator[Event, None, None]#

Run the decision task.

Parameters:

actor (Actor) – The actor making decisions

Yields:

Generator[SimpyEvent, None, None] – Generator for SimPy event queue.

task(*, actor: Any) → Generator[BaseEvent | Process, Any, None]#

Define the process this task follows.

class InterruptStates(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)#

Bases: IntFlag

Class that describes how to behave after an interrupt.

END = 0#
IGNORE = 1#
RESTART = 2#
class Task#

Bases: SettableEnv

A Task is an action that can be performed by an Actor.

INTERRUPT#

alias of InterruptStates

clear_actor_knowledge(actor: Actor, name: str) → None#

Clear knowledge from an actor.

Convenience method for passing in the name of task for actor logging.

Parameters:

  • actor (Actor) – The actor to clear knowledge from

  • name (str) – The name of the knowledge

clear_actor_task_queue(actor: Actor) → None#

Clear out the task queue on the network.

Parameters:

actor (Actor) – The actor whose queue will be cleared

clear_marker() → None#

Clear the marker and set that an interrupt ends the task.

static get_actor_knowledge(actor: Actor, name: str, must_exist: bool = False) → Any#

Get knowledge from the actor.

Parameters:

  • actor (Actor) – The actor to get knowledge from.

  • name (str) – Name of the knowledge

  • must_exist (bool, optional) – Raise errors if the knowledge doesn’t exist.

  • False. (Defaults to)

Returns:

The knowledge value, which could be None

Return type:

Any

get_actor_next_task(actor: Actor) → str | None#

Get the next queued task.

Parameters:

actor (Actor) – The actor to get the next task from

Returns:

The next task name (or None if no task)

Return type:

str | None

get_actor_task_queue(actor: Actor) → list[str]#

Get the task queue on the actor.

Parameters:

actor (Actor) – The actor to modify the task queue of

get_marker() → str | None#

Get the current marker.

Returns:

Marker (or None if cleared)

Return type:

str | None

get_marker_time() → float | None#

The time the current marker was set.

Returns:

Marker set time (or None if cleared)

Return type:

float | None

on_interrupt(*, actor: Any, cause: Any) → InterruptStates#

Define any actions to take on the actor if this task is interrupted.

Note

Custom Tasks can overwrite this method so they can handle being interrupted with a custom procedure. By default, interrupt ends the task.

Parameters:

  • actor (Actor) – the actor using the task

  • cause (Any) – Optional data for the interrupt

rehearse(*, actor: REH_ACTOR, knowledge: dict[str, Any] | None = None, cloned_actor: bool = False, **kwargs: Any) → REH_ACTOR#

Rehearse the task to evaluate its feasibility.

Parameters:

  • actor (Actor) – The actor to rehearse in the task

  • knowledge (dict[str, Any], optional) – Knowledge to add to the actor. Defaults to None.

  • cloned_actor (bool, optional) – If the actor is a clone or not. Defaults to False.

  • kwargs (Any) – Optional args to send to the task.

Returns:

The cloned actor with a state reflecting the task flow.

Return type:

Actor

run(*, actor: Actor) → Generator[Event | Process, Any, None]#

Execute the task.

Parameters:

actor (Actor) – The actor using the task

Returns:

Generator[SimpyEvent, Any, None]

set_actor_knowledge(actor: Actor, name: str, value: Any, overwrite: bool = False) → None#

Set knowledge on the actor.

Convenience method for passing in the name of task for actor logging.

Parameters:

  • actor (Actor) – The actor to set knowledge on.

  • name (str) – Name of the knowledge

  • value (Any) – Value of the knowledge

  • overwrite (bool, optional) – Allow overwrite or not. Defaults to False.

set_actor_task_queue(actor: Actor, task_list: list[str]) → None#

Set the task queue on the actor.

This assumes an empty queue.

Parameters:

  • actor (Actor) – The actor to modify the task queue of

  • task_list (list[str]) – The list of task names to queue.

set_marker(marker: str, interrupt_action: ~upstage.task.InterruptStates = <InterruptStates.END: 0>) → None#

Set a marker to help with inspection of interrupts.

The interrupt_action is set for when no on_interrupt is implemented.

Parameters:

  • marker (str) – String for the marker.

  • interrupt_action (InterruptStates, optional) – Action to take on interrupt.

  • InterruptStates.END. (Defaults to)

task(*, actor: Any) → Generator[BaseEvent | Process, Any, None]#

Define the process this task follows.

class TerminalTask#

Bases: Task

A rehearsal-safe task that cannot exit, i.e., it is terminal.

Note

The user can re-implement the log_message method to return a custom message that will be appended to the actor’s log through its log method.

log_message(*, actor: Actor) → str#

A message to save to a log when this task is reached.

Parameters:

actor (Actor) – The actor using this task.

Returns:

A log message

Return type:

str

on_interrupt(*, actor: Actor, cause: Any) → InterruptStates#

Special case interrupt for terminal task.

Parameters:

  • actor (Actor) – The actor

  • cause (Any) – Additional data sent to the interrupt.

task(*, actor: Actor) → Generator[BaseEvent | Process, Any, None]#

The terminal task.

It’s just a long wait.

Parameters:

actor (Actor) – The actor

process(func: Callable[[...], Generator[Event, Any, None]]) → Callable[[...], Process]#

Decorate a simpy process to schedule it as a callable.

Allows users to decorate a generator, and when they want to schedule them as a simpy process, they can simply call it, e.g., instead of calling:

Usage:

>>> from upstage.api import process, Wait
...
>>> @process
>>> def generator(wait_period=1.0, msg="Finished Waiting"):
>>>     # A simple process that periodically prints a statement
>>>     while True:
>>>         yield Wait(wait_period).as_event()
>>>         print(msg)
...
>>> @process
>>> def another_process():
>>>     # Some other process that calls the first one
>>>     generator()
Parameters:

  • func (Callable[..., Generator[BaseEvent, None, None]]) – The process function that is a

  • events. (generator of simpy)

Returns:

The generator as a simpy process.

Return type:

Process

Note

The value of this decorator is that it reduces the chance of a user forgetting to call the generator as a process, which tends to produce behaviors that are difficult to troubleshoot because the code will build and can run, but the simulation will not work schedule the process defined by the generator.

upstage.task_network module#

The task network class, and factory classes.

Bases: object

Type hinting for task link dictionaries.

allowed: Sequence[str]#
default: str | None#
class TaskNetwork(name: str, task_classes: Mapping[str, type[Task]], task_links: Mapping[str, TaskLinks])#

Bases: object

A means to represent, execute, and rehearse interdependent tasks.

is_feasible(curr: str, new: str) → bool#

Determine if a task can follow another one.

Parameters:

  • curr (str) – Current task name

  • new (str) – Potential next task name

Returns:

If the new task can follow the current.

Return type:

bool

loop(*, actor: Actor, init_task_name: str | None = None) → Generator[Process, None, None]#

Start a task network running its loop.

If no initial task name is given, it will default to following the queue.

Parameters:

  • actor (Actor) – The actor to run the loop on.

  • init_task_name (Optional[str], optional) – Optional task to start running.

  • None. (Defaults to)

rehearse_network(*, actor: REH_ACTOR, task_name_list: list[str], knowledge: dict[str, Any] | None = None, end_task: str | None = None) → REH_ACTOR#

Rehearse a path through the task network.

Parameters:

  • actor (Actor) – The actor to perform the task rehearsal withs

  • task_name_list (list[str]) – The tasks to be performed in order

  • knowledge (dict[str, Any], optional) – Knowledge to give to the cloned/rehearsing actor

  • end_task (str, optional) – A task name to end on

Returns:

A copy of the original actor with state changes associated with the network.

Return type:

Actor

class TaskNetworkFactory(name: str, task_classes: Mapping[str, type[Task]], task_links: Mapping[str, TaskLinks])#

Bases: object

A factory for creating task network instances.

classmethod from_ordered_loop(name: str, task_classes: list[type[Task]]) → TaskNetworkFactory#

Create a network factory from a list of tasks that loops.

Parameters:

  • name (str) – Network name

  • task_classes (list[Task]) – The tasks to run in order.

Returns:

The factory for the ordered network.

Return type:

TaskNetworkFactory

classmethod from_ordered_terminating(name: str, task_classes: list[type[Task]]) → TaskNetworkFactory#

Create a network factory from a list of tasks that terminates.

Parameters:

  • name (str) – Network name

  • task_classes (list[Task]) – The tasks to run in order.

Returns:

The factory for the ordered network.

Return type:

TaskNetworkFactory

classmethod from_single_looping(name: str, task_class: type[Task]) → TaskNetworkFactory#

Create a network factory from a single task that loops.

Parameters:

  • name (str) – Network name

  • task_class (Task) – The single task to loop

Returns:

The factory for the single looping network.

Return type:

TaskNetworkFactory

classmethod from_single_terminating(name: str, task_class: type[Task]) → TaskNetworkFactory#

Create a network factory from a single task that terminates.

Parameters:

  • name (str) – Network name

  • task_class (Task) – The single task to terminate after

Returns:

The factory for the single terminating network.

Return type:

TaskNetworkFactory

make_network(other_name: str | None = None) → TaskNetwork#

Create an instance of the task network.

By default, this uses the name defined on instantiation.

Parameters:

other_name (str, optional) – Another name for the network. Defaults to None.

Returns:

TaskNetwork

upstage.utils module#

This module contains utility functions.

Note

Some of the functions included in this module directly support UPSTAGE’s other modules, and some are there for the user’s convenience.

debug_assert(test: bool, msg: str = '') → None#

Coalesces breakpoints for any failing assert to a single line.

Coalesces all potential lines where the code may fail into one single line that can be marked as a breakpoint.

Parameters:

  • test (bool) – The boolean statement to test, i.e., must evaluate to true or false.

  • msg (str) – The message to display if the test is false.

Note

This is necessary because pdb sometimes does not work well when running simpy due to the way that simpy handles exceptions.

This is also helpful when debugging complex behaviors that run the same code multiple times. Instead of manually writing a try/except statement, you can use debug_assert to do that for you, and ignore all of them from a single control point.

Example

>>> # 1. Add a ``debug_assert`` statement in your code, e.g.:
>>> from upstage.utils import debug_assert
>>> ...
>>> bar, foo = 0, 1  # <<< change foo to be less than bar to raise
>>> debug_assert(foo > bar, "foo is not greater than bar")
>>> # 2. Put a break point on the ``raise error`` line to see why the assert failed.
debug_pause(test: bool | None = None) → None#

Call function to pause IDE on debug mode with single breakpoint.

A helper function to pause the execution of the interpreter when running the code from an IDE (e.g., PyCharm).

Parameters:

test (bool, optional) – A boolean statement to pause on when it evaluates to true.

Note

This is necessary because pdb sometimes does not work well when running simpy due to the way that simpy handles exceptions.

Note

Put a break point on the pass line to pause the IDE.

get_caller_info(caller_level: int = 1) → str#

Get information from the object that called the function.

Parameters:

caller_level (str, optional) – The number of frames to go back in the call stack.

get_caller_object(caller_level: int = 2) → Any#

Inspect the stack to see who called you.

Parameters:

caller_level (int, optional) – Number of hops up in the stack. Defaults to 2.

Returns:

The task object

Return type:

Any

iterable_convert(item: T | list[T] | tuple[T, ...]) → list[T]#

Convert single objects or tuples into a list.

Parameters:

item (T | list[T] | tuple[T,...]) – Object, list, or tuple to convert.

Returns:

The list version of the input.

Return type:

list[T]

waypoint_time_and_dist(start: Location, waypoints: Sequence[Location], speed: float) → tuple[float, float]#

Get the time and distance of a series of locations.

Parameters:

  • start (Location) – Starting point

  • waypoints (Sequence[Location]) – Waypoints after the start

  • speed (float) – Travel speed

Returns:

Time and Distance over the waypoints.

Return type:

tuple[float, float]

Module contents#

A framework for modeling and simulating complex systems of systems.

UPSTAGE (i.e., the Universal Platform for Simulating Tasks and Actors with Graphs and Events) is built atop of SimPy, with the intent of simplifying the development process for complex simulations.

On this page
Show Source