API Docs

alpyne.sim

class alpyne.sim.AnyLogicSim(model_path: str, port: int = 0, py_log_level: int | str | bool = 30, java_log_level: int | str | bool = 30, log_id: str | None = None, auto_lock: bool = True, auto_finish: bool = False, engine_overrides: dict[Literal['units', 'start_time', 'start_date', 'stop_time', 'stop_date', 'seed'], datetime | TimeUnits | int | float | UnitValue] | None = None, config_defaults: dict[str, Any] | None = None, lock_defaults: dict | None = None, java_exe: str | None = None, startup_delay: float = 0.1, **kwargs)[source]

The class for connecting to, communicating with, and controlling an AnyLogic sim model exported via the RL Experiment.

Initialize a connection to the simulation model, with arguments for defining the model setup and operating behavior.

By default, the runs will use the units, start/stop time/date, and RNG seed based on what you set in the RL experiment. This can be overriden by passing a dictionary to engine_overrides.

Parameters:
  • model_path (str) – a relative or absolute path to the exported model zip or extracted model.jar file

  • port (int) – what local port to run the Alpyne app on (0 = find a free one)

  • py_log_level (int | str | bool) – verboseness for this library; expressed as levels from the logging library or a boolean - True defaults to INFO, False to WARNING; defaults to WARNING

  • java_log_level (int | str | bool) – verboseness of java-side logging (writes to alpyne.log); expressed as levels from the logging library or a boolean - True defaults to INFO, False to WARNING; defaults to WARNING

  • log_id (str) – An identifier to put between the log file name and the ‘.log’ extension, useful if you do not want log files to be overridden by subsequent or parallel runs; you can use $p for the port number, $n for a unique number (starts from 1); defaults to None (empty string)

  • auto_lock (bool) – whether to automatically wait for a ‘ready’ state after each call to reset or take_action and return the subsequent RL status instead of None; defaults to False

  • auto_finish (bool) – whether to automatically force the model into a FINISH state once the stop condition (defined in the RL experiment) is met; defaults to False

  • engine_overrides (dict[Literal['units', 'start_time', 'start_date', 'stop_time', 'stop_date', 'seed'], ~datetime.datetime | ~alpyne.outputs.TimeUnits | int | float | ~alpyne.outputs.UnitValue]) – definition for overrides to the engine settings; allowed keys: seed, units, start_time, start_date, stop_time, stop_date; defaults to None

  • config_defaults (dict[str, Any]) – desired default values for the Configuration values; defaults to None

  • lock_defaults (dict) – default values to use when calling lock; flag arg defaults to EngineState.ready(), timeout to 30

  • java_exe (str) – Path to Java executable; if None, uses whatever is associated with the ‘java’ command

  • startup_delay (float) – Seconds to wait between launching the application and sending the first request to verify it’s alive. Too small and it may incorrectly detect as not started up; too large and it adds to startup overhead. Default 0.1

  • kwargs – Internal arguments

Raises:

ModelError – if the app fails to start

Warning

For engine_overrides: (1) do not override the model time units unless ensuring the model does not have ambiguous, unit-generic logic (e.g., calls to time()), as this can affect the logic or outputs; (2) when setting a stop time and date, only the last set value will be used.

schema: SimSchema = None

A class variable (static) defining the model schema, assigned on startup

reset(_config: SimConfiguration | dict | None = None, **config_kwargs) SimStatus | None[source]

Reset the experiment to the beginning with the specified configuration. Any omitted values will use the default (either defined by you in the constructor or else the Java defaults). You should pass an object or keyword arguments (only one necessary).

After applying, the model will auto-run (in the background) to the first call to takeAction or its natural finish condition, whichever comes first.

Parameters:
  • _config (SimConfiguration | dict | None) – A dictionary or dictionary subclass with configuration arguments

  • config_kwargs – Mapping between names defined in the RL Experiment’s Configuration to the desired values

Returns:

Nothing (when auto_wait == False) or the model’s status post-waiting (when auto_wait == True)

Return type:

SimStatus | None

take_action(_action: SimAction | dict | None = None, **action_kwargs) SimStatus | None[source]

Submit an action to the model. You should pass an object or keyword arguments (only one necessary, with the former taking higher precedence).

Parameters:
  • _action (SimAction | dict | None) – The dataclass instance or a dictionary with action arguments

  • action_kwargs – Direct mapping between names defined in the RL Experiment’s Action to the desired values

Returns:

Nothing (when auto_wait == False) or the model’s status (when auto_wait == True)

Return type:

SimStatus | None

observation() SimObservation[source]

Queries the current Observation, regardless of the model’s current state (i.e., it may not be requesting an action yet!). This function is a shorthand for status().observation.

Returns:

The current model observation

Return type:

SimObservation

status() SimStatus[source]

Queries the current status of the model, regardless of its current state (i.e., it may not be requesting an action yet!).

Returns:

The current model status

Return type:

SimStatus

_engine() EngineStatus[source]
Returns:

An immutable object providing engine-level information

Return type:

EngineStatus

lock(flag: EngineState | None = None, timeout: int | None = None) SimStatus[source]

Hang the active thread until the engine is in a given state.

Parameters:
  • flag (EngineState | None) – An encoded indicator for which state(s) to wait for the engine to be in. Defaults to State.ready() (i.e., in PAUSED, FINISHED, or ERROR) unless this object was constructed with different defaults (by you).

  • timeout (int | None) – Maximum wait time, in seconds. Defaults to 30 unless this object was constructed with different defaults (by you).

Returns:

An object providing status information

Raises:

TimeoutError – If timeout elapses before the desired state is reached

Return type:

SimStatus

outputs(*names: str) list[int | float | str | bool | datetime | UnitValue | StatisticsDiscrete | StatisticsContinuous | DataSet | HistogramSimpleData | HistogramSmartData | Histogram2DData] | dict[str, int | float | str | bool | datetime | UnitValue | StatisticsDiscrete | StatisticsContinuous | DataSet | HistogramSimpleData | HistogramSmartData | Histogram2DData] | None[source]

Retrieves the values of any analysis-related objects in the top-level agent (if any exist), as detailed further in the AnyLogic Help article Data analysis items and its subpages.

Each of the analysis-related objects have a type of the same name implemented in this library. Output objects for plain scalar types are converted to their natural python equivalent (e.g., String -> str, double -> float); types units attached make use of the custom UnitValue type.

Parameters:

names (str) – The names (as defined in the sim) of objects to query the current value for; passing nothing will get everything

Returns:

The desired values; provided as a list when explicit names are used (in the specified order), otherwise in a dictionary keyed by the names

Return type:

list[int | float | str | bool | datetime | UnitValue | StatisticsDiscrete | StatisticsContinuous | DataSet | HistogramSimpleData | HistogramSmartData | Histogram2DData] | dict[str, int | float | str | bool | datetime | UnitValue | StatisticsDiscrete | StatisticsContinuous | DataSet | HistogramSimpleData | HistogramSmartData | Histogram2DData] | None

alpyne.data

class alpyne.data.SimConfiguration(*args, **kwargs)[source]

A subclass of UserDict describing the desired Configuration, as defined in the RL experiment, for the sim to use when resetting.

Usage of this class adds validation to ensure only the pre-defined fields can be set.

class alpyne.data.SimObservation(*args, **kwargs)[source]

A subclass of UserDict describing the received Observation, as defined in the RL experiment, received from the sim.

class alpyne.data.SimAction(*args, **kwargs)[source]

A subclass of UserDict describing the desired Action, as defined in the RL experiment, for the sim to use when submitting an action.

Usage of this class adds validation to ensure only the pre-defined fields can be set.

class alpyne.data.SimStatus(state: EngineState, observation: SimObservation, stop: bool, sequence_id: int, episode_num: int, step_num: int, time: float, date: datetime | int, progress: float, message: str | None)[source]

A report of the current simulation model’s status

Parameters:
  • state (EngineState) – The current state of the model’s engine; this matches the value of what AnyLogic reports (e.g., from getEngine().getState())

  • observation (SimObservation) – A dictionary-subclass-typed object mapping field names with values, as defined in the RL Experiment

  • stop (bool) – The value of the RL Experiment’s “Simulation run stop condition” field; indicates whether the episode should be terminated (e.g., fail or success condition)

  • sequence_id (int) – A counter of how many actionable requests (i.e., resets + actions) have been taken

  • episode_num (int) – A counter of how many resets have been taken

  • step_num (int) – A counter of how many actions have been taken

  • time (float) – The model time, in the engine’s set time units

  • date (datetime | int) – The model date

  • progress (float) – If the engine has a set stop time/date, this is a value between 0-1 indicating it’s percent completion; otherwise (i.e., set to run indefinitely), this will be -1

  • message (str | None) – An informal message from the underlying Alpyne app, used to report potential reasons for the current state of the model; usually set when some stopping scenario has occurred. It may be None. May change in the future.

class alpyne.data.EngineStatus(state: EngineState, engine_events: int, engine_steps: int, next_engine_step: float, next_engine_event: float, time: float, date: datetime | str, progress: float, message: str | None, settings: dict[Literal['units', 'start_time', 'start_date', 'stop_time', 'stop_date', 'seed'], int | float | datetime | TimeUnits])[source]

A report for the status of the underlying AnyLogic engine driving the simulation model.

Warning

This object is not currently part of the public API and is intended for debugging purposes only. It may be refactored or removed in the future.

Parameters:
  • state (EngineState) – The current state of the model’s engine; this matches the value of what AnyLogic reports (e.g., from getEngine().getState())

  • engine_events (int) – The number of currently scheduled events in the model (both by the user and the engine)

  • engine_steps (int) – A counter of how many of events have been executed by the engine

  • next_engine_step (float) – The time (in the engine’s time units) which will be after the engine’s next step() execution. If the model is about to finish, returns negative infinity. Special cases: in system-dynamics models the result may depend on the selected accuracy in system-dynamics models some conditional events may occur based on the value obtained from numerical solver within the step.

  • next_engine_event (float) – The time (in the engine’s time units) of the next event scheduled

  • time (float) – The current model (logical) time, in the engine’s time units

  • date (datetime | str) – The current model date

  • progress (float) – The progress of the simulation: the part of model time simulated so far in case the stop time is set, or -1 if it is not set

  • message (str | None) – An informal message from the underlying Alpyne app, used to report potential reasons for the current state of the model; usually set when some stopping scenario has occurred. It may be None. May change in the future.

  • settings (dict[Literal['units', 'start_time', 'start_date', 'stop_time', 'stop_date', 'seed'], int | float | ~datetime.datetime | ~alpyne.outputs.TimeUnits]) – A dictionary mapping the engine setting key names to the values currently in use by the model

class alpyne.data.FieldData(name: str, type: str, value: Any, units: str | None = None)[source]

Represents a single data element with a name, type, value, and (optional) units.

Used to describe the space information in the schema object or basic input/output types.

Two properties - py_type and py_value - are available to convert the Java type and value (respectively) to Python native equivalents (e.g., Python’s datetime.datetime for Java’s Date type; UnitValue objects for Outputs with a unit-type).

Parameters:
  • name (str) – The user-defined field name

  • type (str) – The “simple” (Java) class name

  • value (Any) – The default value that will be used if omitted

  • units (str) – The AnyLogic unit type constant, when relevant (None otherwise)

class alpyne.data.SimSchema(_schema_def: dataclasses.InitVar[dict], inputs: dict[str, FieldData] | None = None, outputs: dict[str, FieldData] | None = None, configuration: dict[str, FieldData] | None = None, engine_settings: dict[str, FieldData] | None = None, observation: dict[str, FieldData] | None = None, action: dict[str, FieldData] | None = None)[source]

Contains information describing each of the possible data-holding objects. Each attribute is a dictionary mapping the field name to its data. These are provided as a reference for what’s available and is not intended to be modified.

Parameters:
  • _schema_def (dataclasses.InitVar[dict]) – A pseudo-field used only for initializing the schema

  • inputs (dict[str, FieldData]) – Parameters of the top-level agent

  • outputs (dict[str, FieldData]) – Analysis objects (Output, DataSet, HistogramData, etc.)

  • configuration (dict[str, FieldData]) – Defined data fields in the Configuration section of the RL experiment

  • engine_settings (dict[str, FieldData]) – Represents various engine settings (model units, random seed, start and stop time/date)

  • observation (dict[str, FieldData]) – Defined data fields in the Observation section of the RL experiment

  • action (dict[str, FieldData]) – Defined data fields in the Action section of the RL experiment.

Note

The inputs are provided for information about the model and not currently able to be assigned (v1.0.0)

class alpyne.data.EngineSettings(**override_kwargs: dict[Literal['units', 'start_time', 'start_date', 'stop_time', 'stop_date', 'seed'], datetime | TimeUnits | int | float | UnitValue | None])[source]

Settings to use for the underlying AnyLogic engine to run the simulation model with; contains fields for the time units, start/stop time/date, and RNG seed.

Parameters:

override_kwargs (dict[Literal['units', 'start_time', 'start_date', 'stop_time', 'stop_date', 'seed'], ~datetime.datetime | ~alpyne.outputs.TimeUnits | int | float | ~alpyne.outputs.UnitValue | None]) – Desired mapping between setting name and value to override in the sim’s RL experiment

property stop_time

The time (in the units) to set the model’s engine to a FINISHED state, preventing further events or actions from being taken.

property stop_date: datetime | None

The date to set the model’s engine to a FINISHED state, preventing further events or actions from being taken.

alpyne.outputs

class alpyne.outputs.StatisticsDiscrete(count: int = 0, mean: float = 0.0, confidence: float = inf, min: float = inf, max: float = -inf, deviation: float = 0.0, sum: float = 0.0)[source]

Bases: _Statistics

Parameters:
  • count (int)

  • mean (float)

  • confidence (float)

  • min (float)

  • max (float)

  • deviation (float)

  • sum (float)

class alpyne.outputs.StatisticsContinuous(count: int = 0, mean: float = 0.0, confidence: float = inf, min: float = inf, max: float = -inf, deviation: float = 0.0, integral: float = 0.0)[source]

Bases: _Statistics

Parameters:
  • count (int)

  • mean (float)

  • confidence (float)

  • min (float)

  • max (float)

  • deviation (float)

  • integral (float)

class alpyne.outputs.DataSet(xmin: float = inf, xmean: float = 0.0, xmedian: float = 0.0, xmax: float = -inf, ymin: float = inf, ymean: float = 0.0, ymedian: float = 0.0, ymax: float = -inf, plainDataTable: list[list[float]] = <factory>)[source]

Bases: _AnalysisObject

Parameters:
  • xmin (float)

  • xmean (float)

  • xmedian (float)

  • xmax (float)

  • ymin (float)

  • ymean (float)

  • ymedian (float)

  • ymax (float)

  • plainDataTable (list[list[float]])

class alpyne.outputs.HistogramSmartData(count: int = 0, lowerBound: float = 0.0, intervalWidth: float = 0.1, hits: list[int] = <factory>, statistics: alpyne.outputs._Statistics = <factory>)[source]

Bases: _AnalysisObject

Parameters:
  • count (int)

  • lowerBound (float)

  • intervalWidth (float)

  • hits (list[int])

  • statistics (_Statistics)

class alpyne.outputs.HistogramSimpleData(count: int = 0, lowerBound: float = 0.0, intervalWidth: float = 0.1, hits: list[int] = <factory>, statistics: alpyne.outputs._Statistics = <factory>, hitsOutLow: float = 0.0, hitsOutHigh: float = 0.0)[source]

Bases: HistogramSmartData

Parameters:
  • count (int)

  • lowerBound (float)

  • intervalWidth (float)

  • hits (list[int])

  • statistics (_Statistics)

  • hitsOutLow (float)

  • hitsOutHigh (float)

class alpyne.outputs.Histogram2DData(hits: list[list[int]] = <factory>, hitsOutLow: list[int] = <factory>, hitsOutHigh: list[int] = <factory>, xMin: float = inf, xMax: float = -inf, yMin: float = inf, yMax: float = -inf)[source]

Bases: _AnalysisObject

Parameters:
  • hits (list[list[int]])

  • hitsOutLow (list[int])

  • hitsOutHigh (list[int])

  • xMin (float)

  • xMax (float)

  • yMin (float)

  • yMax (float)

enum alpyne.outputs.AmountUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

LITER = <AmountUnits.LITER: NumericUnitAttrs(conversion_factor=0.001, symbol='L')>
OIL_BARREL = <AmountUnits.OIL_BARREL: NumericUnitAttrs(conversion_factor=0.158987295, symbol='barrels')>
CUBIC_METER = <AmountUnits.CUBIC_METER: NumericUnitAttrs(conversion_factor=1.0, symbol='m3')>
KILOGRAM = <AmountUnits.KILOGRAM: NumericUnitAttrs(conversion_factor=1.0, symbol='kg')>
TON = <AmountUnits.TON: NumericUnitAttrs(conversion_factor=1000.0, symbol='ton')>
enum alpyne.outputs.TimeUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

MILLISECOND = <TimeUnits.MILLISECOND: NumericUnitAttrs(conversion_factor=0.001, symbol='ms')>
SECOND = <TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>
MINUTE = <TimeUnits.MINUTE: NumericUnitAttrs(conversion_factor=60.0, symbol='min')>
HOUR = <TimeUnits.HOUR: NumericUnitAttrs(conversion_factor=3600.0, symbol='hr')>
DAY = <TimeUnits.DAY: NumericUnitAttrs(conversion_factor=86400.0, symbol='day')>
WEEK = <TimeUnits.WEEK: NumericUnitAttrs(conversion_factor=604800.0, symbol='wk')>
MONTH = <TimeUnits.MONTH: NumericUnitAttrs(conversion_factor=2592000.0, symbol='mn')>
YEAR = <TimeUnits.YEAR: NumericUnitAttrs(conversion_factor=31536000.0, symbol='yr')>
enum alpyne.outputs.LengthUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

MILLIMETER = <LengthUnits.MILLIMETER: NumericUnitAttrs(conversion_factor=0.001, symbol='mm')>
CENTIMETER = <LengthUnits.CENTIMETER: NumericUnitAttrs(conversion_factor=0.01, symbol='cm')>
METER = <LengthUnits.METER: NumericUnitAttrs(conversion_factor=1.0, symbol='m')>
KILOMETER = <LengthUnits.KILOMETER: NumericUnitAttrs(conversion_factor=1000.0, symbol='km')>
INCH = <LengthUnits.INCH: NumericUnitAttrs(conversion_factor=0.0254, symbol='in')>
FOOT = <LengthUnits.FOOT: NumericUnitAttrs(conversion_factor=0.3048, symbol='ft')>
YARD = <LengthUnits.YARD: NumericUnitAttrs(conversion_factor=0.9144, symbol='yd')>
MILE = <LengthUnits.MILE: NumericUnitAttrs(conversion_factor=1609.344, symbol='m')>
NAUTICAL_MILE = <LengthUnits.NAUTICAL_MILE: NumericUnitAttrs(conversion_factor=1853.184, symbol='nm')>
enum alpyne.outputs.AngleUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

TURN = <AngleUnits.TURN: NumericUnitAttrs(conversion_factor=6.283185307179586, symbol='turn')>
RADIAN = <AngleUnits.RADIAN: NumericUnitAttrs(conversion_factor=1.0, symbol='rad')>
DEGREE = <AngleUnits.DEGREE: NumericUnitAttrs(conversion_factor=0.017453292519943295, symbol='deg')>
enum alpyne.outputs.AreaUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

SQ_MILLIMETER = <AreaUnits.SQ_MILLIMETER: SpaceUnitAttrs(length_unit=<LengthUnits.MILLIMETER: NumericUnitAttrs(conversion_factor=0.001, symbol='mm')>, symbol='mm2')>
SQ_CENTIMETER = <AreaUnits.SQ_CENTIMETER: SpaceUnitAttrs(length_unit=<LengthUnits.CENTIMETER: NumericUnitAttrs(conversion_factor=0.01, symbol='cm')>, symbol='cm2')>
SQ_METER = <AreaUnits.SQ_METER: SpaceUnitAttrs(length_unit=<LengthUnits.METER: NumericUnitAttrs(conversion_factor=1.0, symbol='m')>, symbol='m2')>
SQ_KILOMETER = <AreaUnits.SQ_KILOMETER: SpaceUnitAttrs(length_unit=<LengthUnits.KILOMETER: NumericUnitAttrs(conversion_factor=1000.0, symbol='km')>, symbol='km2')>
SQ_INCH = <AreaUnits.SQ_INCH: SpaceUnitAttrs(length_unit=<LengthUnits.INCH: NumericUnitAttrs(conversion_factor=0.0254, symbol='in')>, symbol='in2')>
SQ_FOOT = <AreaUnits.SQ_FOOT: SpaceUnitAttrs(length_unit=<LengthUnits.FOOT: NumericUnitAttrs(conversion_factor=0.3048, symbol='ft')>, symbol='ft2')>
SQ_YARD = <AreaUnits.SQ_YARD: SpaceUnitAttrs(length_unit=<LengthUnits.YARD: NumericUnitAttrs(conversion_factor=0.9144, symbol='yd')>, symbol='yard2')>
SQ_MILE = <AreaUnits.SQ_MILE: SpaceUnitAttrs(length_unit=<LengthUnits.MILE: NumericUnitAttrs(conversion_factor=1609.344, symbol='m')>, symbol='mile2')>
SQ_NAUTICAL_MILE = <AreaUnits.SQ_NAUTICAL_MILE: SpaceUnitAttrs(length_unit=<LengthUnits.NAUTICAL_MILE: NumericUnitAttrs(conversion_factor=1853.184, symbol='nm')>, symbol='nautmile2')>
enum alpyne.outputs.RateUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

PER_MILLISECOND = <RateUnits.PER_MILLISECOND: RateUnitAttrs(time_unit=<TimeUnits.MILLISECOND: NumericUnitAttrs(conversion_factor=0.001, symbol='ms')>, symbol='per ms')>
PER_SECOND = <RateUnits.PER_SECOND: RateUnitAttrs(time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='per sec')>
PER_MINUTE = <RateUnits.PER_MINUTE: RateUnitAttrs(time_unit=<TimeUnits.MINUTE: NumericUnitAttrs(conversion_factor=60.0, symbol='min')>, symbol='per min')>
PER_HOUR = <RateUnits.PER_HOUR: RateUnitAttrs(time_unit=<TimeUnits.HOUR: NumericUnitAttrs(conversion_factor=3600.0, symbol='hr')>, symbol='per hr')>
PER_DAY = <RateUnits.PER_DAY: RateUnitAttrs(time_unit=<TimeUnits.DAY: NumericUnitAttrs(conversion_factor=86400.0, symbol='day')>, symbol='per day')>
PER_WEEK = <RateUnits.PER_WEEK: RateUnitAttrs(time_unit=<TimeUnits.WEEK: NumericUnitAttrs(conversion_factor=604800.0, symbol='wk')>, symbol='per wk')>
PER_MONTH = <RateUnits.PER_MONTH: RateUnitAttrs(time_unit=<TimeUnits.MONTH: NumericUnitAttrs(conversion_factor=2592000.0, symbol='mn')>, symbol='per month')>
PER_YEAR = <RateUnits.PER_YEAR: RateUnitAttrs(time_unit=<TimeUnits.YEAR: NumericUnitAttrs(conversion_factor=31536000.0, symbol='yr')>, symbol='per year')>
enum alpyne.outputs.AccelerationUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

MPS_SQ = <AccelerationUnits.MPS_SQ: VelocityUnitAttrs(spacial_unit=<LengthUnits.METER: NumericUnitAttrs(conversion_factor=1.0, symbol='m')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='mps2')>
FPS_SQ = <AccelerationUnits.FPS_SQ: VelocityUnitAttrs(spacial_unit=<LengthUnits.FOOT: NumericUnitAttrs(conversion_factor=0.3048, symbol='ft')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='fps2')>
enum alpyne.outputs.SpeedUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

MPS = <SpeedUnits.MPS: VelocityUnitAttrs(spacial_unit=<LengthUnits.METER: NumericUnitAttrs(conversion_factor=1.0, symbol='m')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='meters per second')>
KPH = <SpeedUnits.KPH: VelocityUnitAttrs(spacial_unit=<LengthUnits.KILOMETER: NumericUnitAttrs(conversion_factor=1000.0, symbol='km')>, time_unit=<TimeUnits.HOUR: NumericUnitAttrs(conversion_factor=3600.0, symbol='hr')>, symbol='kilometers per hour')>
FPS = <SpeedUnits.FPS: VelocityUnitAttrs(spacial_unit=<LengthUnits.FOOT: NumericUnitAttrs(conversion_factor=0.3048, symbol='ft')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='feet per second')>
FPM = <SpeedUnits.FPM: VelocityUnitAttrs(spacial_unit=<LengthUnits.FOOT: NumericUnitAttrs(conversion_factor=0.3048, symbol='ft')>, time_unit=<TimeUnits.MINUTE: NumericUnitAttrs(conversion_factor=60.0, symbol='min')>, symbol='feet per minute')>
MPH = <SpeedUnits.MPH: VelocityUnitAttrs(spacial_unit=<LengthUnits.MILE: NumericUnitAttrs(conversion_factor=1609.344, symbol='m')>, time_unit=<TimeUnits.HOUR: NumericUnitAttrs(conversion_factor=3600.0, symbol='hr')>, symbol='miles per hour')>
KN = <SpeedUnits.KN: VelocityUnitAttrs(spacial_unit=<LengthUnits.NAUTICAL_MILE: NumericUnitAttrs(conversion_factor=1853.184, symbol='nm')>, time_unit=<TimeUnits.HOUR: NumericUnitAttrs(conversion_factor=3600.0, symbol='hr')>, symbol='knots')>
enum alpyne.outputs.FlowRateUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

LITER_PER_SECOND = <FlowRateUnits.LITER_PER_SECOND: VelocityUnitAttrs(spacial_unit=<AmountUnits.LITER: NumericUnitAttrs(conversion_factor=0.001, symbol='L')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='liter per second')>
OIL_BARREL_PER_SECOND = <FlowRateUnits.OIL_BARREL_PER_SECOND: VelocityUnitAttrs(spacial_unit=<AmountUnits.OIL_BARREL: NumericUnitAttrs(conversion_factor=0.158987295, symbol='barrels')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='oil barrel per second')>
CUBIC_METER_PER_SECOND = <FlowRateUnits.CUBIC_METER_PER_SECOND: VelocityUnitAttrs(spacial_unit=<AmountUnits.CUBIC_METER: NumericUnitAttrs(conversion_factor=1.0, symbol='m3')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='meter3 per second')>
KILOGRAM_PER_SECOND = <FlowRateUnits.KILOGRAM_PER_SECOND: VelocityUnitAttrs(spacial_unit=<AmountUnits.KILOGRAM: NumericUnitAttrs(conversion_factor=1.0, symbol='kg')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='kilogram per second')>
TON_PER_SECOND = <FlowRateUnits.TON_PER_SECOND: VelocityUnitAttrs(spacial_unit=<AmountUnits.TON: NumericUnitAttrs(conversion_factor=1000.0, symbol='ton')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='ton per second')>
enum alpyne.outputs.RotationSpeedUnits(value)[source]

Bases: _UnitEnum

An enumeration.

Valid values are as follows:

RPM = <RotationSpeedUnits.RPM: VelocityUnitAttrs(spacial_unit=<AngleUnits.TURN: NumericUnitAttrs(conversion_factor=6.283185307179586, symbol='turn')>, time_unit=<TimeUnits.MINUTE: NumericUnitAttrs(conversion_factor=60.0, symbol='min')>, symbol='rotations per minute')>
RAD_PER_SECOND = <RotationSpeedUnits.RAD_PER_SECOND: VelocityUnitAttrs(spacial_unit=<AngleUnits.RADIAN: NumericUnitAttrs(conversion_factor=1.0, symbol='rad')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='radians per second')>
DEG_PER_SECOND = <RotationSpeedUnits.DEG_PER_SECOND: VelocityUnitAttrs(spacial_unit=<AngleUnits.DEGREE: NumericUnitAttrs(conversion_factor=0.017453292519943295, symbol='deg')>, time_unit=<TimeUnits.SECOND: NumericUnitAttrs(conversion_factor=1.0, symbol='sec')>, symbol='degrees per second')>
class alpyne.outputs.UnitValue(value: int | float, unit: _UnitEnum | str)[source]

Bases: object

A custom type to represent some numerical value with units.

Parameters:
  • value (int | float)

  • unit (_UnitEnum | str)

alpyne.constants

alpyne.constants.DATE_PATTERN_LOOKUP = {'^[A-Za-z]{3}, \\d{2} [A-Za-z]{3} \\d{4} \\d{2}:\\d{2}:\\d{2}[ A-Z]{0,4}$': '%a, %d %b %Y %H:%M:%S %Z', '^\\d{4}-\\d{2}-\\d{2}$': '%Y-%m-%d', '^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}$': '%Y-%m-%dT%H:%M:%S', '^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d{1,6}$': '%Y-%m-%dT%H:%M:%S.%f', '^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d{1,6}[-+](?:\\d{2}:*)+(?:\\.\\d+)*$': '%Y-%m-%dT%H:%M:%S.%f%:z', '^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d{1,6}[-+]\\d{2,6}\\.*\\d*$': '%Y-%m-%dT%H:%M:%S.%f%z'}

Constant mapping between a regex for valid date/time pattern and the equivalent datetime.strftime/strptime pattern

alpyne.constants.TYPE_LOOKUP = {'ArrayList': <class 'list'>, 'Boolean': <class 'bool'>, 'DataSet': <class 'alpyne.outputs.DataSet'>, 'Date': <class 'datetime.datetime'>, 'Double': <class 'float'>, 'Float': <class 'float'>, 'HashMap': <class 'dict'>, 'HashSet': <class 'list'>, 'Histogram2DData': <class 'alpyne.outputs.Histogram2DData'>, 'HistogramSimpleData': <class 'alpyne.outputs.HistogramSimpleData'>, 'HistogramSmartData': <class 'alpyne.outputs.HistogramSmartData'>, 'Integer': <class 'int'>, 'LinkedHashMap': <class 'dict'>, 'LinkedHashSet': <class 'list'>, 'LinkedList': <class 'list'>, 'Long': <class 'int'>, 'Map': <class 'dict'>, 'StatisticsContinuous': <class 'alpyne.outputs.StatisticsContinuous'>, 'StatisticsDiscrete': <class 'alpyne.outputs.StatisticsDiscrete'>, 'String': <class 'str'>, 'TimeUnits': <enum 'TimeUnits'>, 'TreeMap': <class 'dict'>, 'TreeSet': <class 'list'>, 'boolean': <class 'bool'>, 'double': <class 'float'>, 'float': <class 'float'>, 'int': <class 'int'>, 'long': <class 'int'>}

Constant mapping between simple names of Java classes to the Python equivalent

enum alpyne.constants.JavaLogLevel(value)[source]

Bases: Enum

Represents the log level to use in the Alpyne Java application.

Valid values are as follows:

SEVERE = <JavaLogLevel.SEVERE: 'SEVERE'>
WARNING = <JavaLogLevel.WARNING: 'WARNING'>
INFO = <JavaLogLevel.INFO: 'INFO'>
CONFIG = <JavaLogLevel.CONFIG: 'CONFIG'>
FINE = <JavaLogLevel.FINE: 'FINE'>
FINER = <JavaLogLevel.FINER: 'FINER'>
FINEST = <JavaLogLevel.FINEST: 'FINEST'>
flag alpyne.constants.EngineState(value)[source]

Bases: Flag

Represents the state of the sim’s engine.

Valid values are as follows:

IDLE = <EngineState.IDLE: 1>
PAUSED = <EngineState.PAUSED: 2>
RUNNING = <EngineState.RUNNING: 4>
FINISHED = <EngineState.FINISHED: 8>
ERROR = <EngineState.ERROR: 16>
PLEASE_WAIT = <EngineState.PLEASE_WAIT: 32>

alpyne.typing

alpyne.typing.EngineSettingKeys

Valid keys related to the engine settings; used when passing the dictionary with override values in the AnyLogicSim’s constructor keyword argument

alias of Literal[‘units’, ‘start_time’, ‘start_date’, ‘stop_time’, ‘stop_date’, ‘seed’]

alpyne.typing.Number

Describes any valid number type

alias of int | float

alpyne.typing.OutputType

Describes the types from calling outputs(); includes all analysis and Output objects (anything with units are described by UnitValue)

alias of int | float | str | bool | datetime | UnitValue | StatisticsDiscrete | StatisticsContinuous | DataSet | HistogramSimpleData | HistogramSmartData | Histogram2DData

alpyne.env

class alpyne.env.AlpyneEnv(sim: AnyLogicSim)[source]

Bases: Env

A mostly-complete implementation of an environment using the Gymnasium interface, setup to handle most of the routine communication with a provided instance of the AnyLogicSim.

To use it, you’ll need to either create a subclass or use the provided make_alpyne_env() function.

When subclassing, you are required to implement the _calc_reward function and extend the __init__ function to assign the observation_space and action_space class attributes.

You may need to override the other “private” functions if their logic do not match your scenario.

Parameters:

sim (AnyLogicSim)

__init__(sim: AnyLogicSim)[source]
Parameters:

sim (AnyLogicSim)

_get_config() dict | None[source]

Called at the start of each new episode. If a Configuration is returned, that will take highest priority. Otherwise, if None is returned, it will use the default configuration (either defined by you in the sim constructor or Java defaults).

Return type:

dict | None

_get_obs(status: SimStatus) ObsType[source]

Convert the current status of your model to the defined observation type.

Raises:

NotImplementedError – If your observation space deviates from the assumptions (thus requiring you to implement them)

Parameters:

status (SimStatus)

Return type:

ObsType

_to_action(act: ActType) dict[source]

Convert the action received by the code/library using this environment to an action instance to be passed to the sim.

Parameters:

act (ActType)

Return type:

dict

_get_info(status: SimStatus) dict[str, Any] | None[source]

Provide some (technically optional) auxiliary diagnostic information

Parameters:

status (SimStatus)

Return type:

dict[str, Any] | None

_is_terminal(status: SimStatus) bool[source]

Whether the sim is considered ‘terminal’ by gymnasium standards - in summary, the episode is ended due to conditions explicit to the model definition (e.g., some “win” or “lose” condition, completion of full work day or other times on a model with a finite time horizon).

For a more complete/correct definition, see: https://gymnasium.farama.org/tutorials/gymnasium_basics/handling_time_limits/

For a simulation-based definition of finite vs infinite time horizons, see: https://rossetti.github.io/RossettiArenaBook/ch3-finiteVsInfinite.html

The default assumption is based on the value of the ‘stop’ condition (i.e., “Simulation run stop condition” in the RL experiment) or the simulation being in the “FINISHED” or “ERROR” state (e.g., from the model having called finish(), the stop time/date being met, logical error).

Parameters:

status (SimStatus)

Return type:

bool

_is_truncated(status: SimStatus) bool[source]

Whether the sim is considered ‘truncated’ by gymnasium standards - in summary, the episode is ended due to conditions not explicit to the model definition (e.g., artificial time limit on a model with an infinite time horizon).

For a more complete/correct definition, see: https://gymnasium.farama.org/tutorials/gymnasium_basics/handling_time_limits/

For a simulation-based definition of finite vs infinite time horizons, see: https://rossetti.github.io/RossettiArenaBook/ch3-finiteVsInfinite.html

The default assumption is the model is never truncated (i.e., False).

Parameters:

status (SimStatus)

Return type:

bool

reset(*, seed: int | None = None, options: dict[str, Any] | None = None) tuple[ObsType, dict[str, Any]][source]

Resets the simulation model, advancing to the first stopping point in the sim.

Parameters:
  • seed (int | None) – The seed that is used to initialize the PRNG (specifically the attribute np_random)

  • options (dict[str, Any] | None) – Optional settings. Keys matching those in alpyne.typing.EngineSettingKeys() will override the current sim’s engine_settings and keys matching ones in the configuration will override the values in both the default configuration settings and _get_config().

Returns:

The observation and info at the next stopping point

Return type:

tuple[ObsType, dict[str, Any]]

step(action: ActType) tuple[ObsType, SupportsFloat, bool, bool, dict[str, Any]][source]

Submit an action to the simulation model and run to the next stopping point (e.g., the next call to takeAction or any terminating condition).

Parameters:

action (ActType) – The desired data to pass to the sim, in the format defined by your action_space

Returns:

The current observation, reward, terminal flag, truncated flag, optional information dictionary

Return type:

tuple[ObsType, SupportsFloat, bool, bool, dict[str, Any]]

alpyne.env.make(sim: AnyLogicSim, observation_space: ObsType, action_space: ActType, _calc_reward: Callable[[SimStatus], SupportsFloat], **kwargs) AlpyneEnv[source]

A helper function to use in lieu of subclassing alpyne.env.AlpyneEnv. Using this will define a class named “CustomAlpyneEnv”.

Parameters:
  • sim (AnyLogicSim) – The instantiated AnyLogicSim object for the target sim

  • observation_space (ObsType) – The gymnasium definition for the observation space

  • action_space (ActType) – The gymnasium definition for the action space

  • _calc_reward (Callable[[SimStatus], SupportsFloat]) – A callable taking the status as input and returning the reward from the previous action

  • kwargs – Names of other functions from :class:`AlpyneEnv to callables with the respective signatures and returns

Returns:

An instantiated instance of an alpyne.env.AlpyneEnv subclass

Return type:

AlpyneEnv

alpyne.utils

alpyne.utils.next_num(start=0, step=1)[source]

Helper function to return a unique number every time it’s called. Useful for easily generating unique seeds for the engine settings (e.g., {seed=next_unique_num}). The arguments only apply the first time the function is called, so changing after the first will have no effect.

Parameters:
  • start – The first number to return; where counting starts

  • step – How much to increment each time the function is called

Returns:

The next updated number in the series

class alpyne.utils.AlpyneJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: JSONEncoder

A custom encoder to convert Alpyne classes to JSON, in a format expected by the alpyne app.

To use it, pass a reference to the class to the cls argument of json.dump or json.dumps). For example, json.dumps(my_object, cls=AlpyneJSONEncoder)

Constructor for JSONEncoder, with sensible defaults.

If skipkeys is false, then it is a TypeError to attempt encoding of keys that are not str, int, float or None. If skipkeys is True, such items are simply skipped.

If ensure_ascii is true, the output is guaranteed to be str objects with all incoming non-ASCII characters escaped. If ensure_ascii is false, the output can contain non-ASCII characters.

If check_circular is true, then lists, dicts, and custom encoded objects will be checked for circular references during encoding to prevent an infinite recursion (which would cause an RecursionError). Otherwise, no such check takes place.

If allow_nan is true, then NaN, Infinity, and -Infinity will be encoded as such. This behavior is not JSON specification compliant, but is consistent with most JavaScript based encoders and decoders. Otherwise, it will be a ValueError to encode such floats.

If sort_keys is true, then the output of dictionaries will be sorted by key; this is useful for regression tests to ensure that JSON serializations can be compared on a day-to-day basis.

If indent is a non-negative integer, then JSON array elements and object members will be pretty-printed with that indent level. An indent level of 0 will only insert newlines. None is the most compact representation.

If specified, separators should be an (item_separator, key_separator) tuple. The default is (’, ‘, ‘: ‘) if indent is None and (‘,’, ‘: ‘) otherwise. To get the most compact JSON representation, you should specify (‘,’, ‘:’) to eliminate whitespace.

If specified, default is a function that gets called for objects that can’t otherwise be serialized. It should return a JSON encodable version of the object or raise a TypeError.

default(o)[source]

Overridden method to handle classes used by Alpyne.

class alpyne.utils.AlpyneJSONDecoder(*, object_hook=None, parse_float=None, parse_int=None, parse_constant=None, strict=True, object_pairs_hook=None)[source]

Bases: JSONDecoder

A custom decoder to convert JSON to Alpyne classes.

To use it, pass a reference to the class to the cls argument of json.load or json.loads). For example, json.loads(my_json_str, cls=AlpyneJSONDecoder)

object_hook, if specified, will be called with the result of every JSON object decoded and its return value will be used in place of the given dict. This can be used to provide custom deserializations (e.g. to support JSON-RPC class hinting).

object_pairs_hook, if specified will be called with the result of every JSON object decoded with an ordered list of pairs. The return value of object_pairs_hook will be used instead of the dict. This feature can be used to implement custom decoders. If object_hook is also defined, the object_pairs_hook takes priority.

parse_float, if specified, will be called with the string of every JSON float to be decoded. By default this is equivalent to float(num_str). This can be used to use another datatype or parser for JSON floats (e.g. decimal.Decimal).

parse_int, if specified, will be called with the string of every JSON int to be decoded. By default this is equivalent to int(num_str). This can be used to use another datatype or parser for JSON integers (e.g. float).

parse_constant, if specified, will be called with one of the following strings: -Infinity, Infinity, NaN. This can be used to raise an exception if invalid JSON numbers are encountered.

If strict is false (true is the default), then control characters will be allowed inside strings. Control characters in this context are those with character codes in the 0-31 range, including '\t' (tab), '\n', '\r' and '\0'.

decode(s, **kwargs)[source]

Return the Python representation of s (a str instance containing a JSON document).

alpyne.utils.resolve_model_jar(model_loc: str) Tuple[Path, TemporaryDirectory][source]

Validate the provided location of a supposed model. It handles location validation, unzipping zip files to a temporary directory, and other cases.

Parameters:

model_loc (str) – The user-specified location of their model

Returns:

A (possibly updated) location of the model.jar and the TemporaryDirectory object, if one was created

Raises:

ValueError – If the location is ambiguous or couldn’t be properly resolved

Return type:

Tuple[Path, TemporaryDirectory]

alpyne.utils.histogram_outputs_to_fake_dataset(lower_bound: float, interval_width: float, hits: List[int]) Tuple[List[float], List[float]][source]

Convert statistics from a histogram output to a format usable by plotting libraries (e.g., matplotlib). This recreates the histogram in a way that visually appears the same; however, it’s not (necessarily) statistically similar.

Parameters:
  • lower_bound (float) – The start/minimum X value

  • interval_width (float) – How large each “bin” is

  • hits (List[int]) – A list of hits in each bin

Returns:

A tuple consisting of two lists - for sample data and bins - to use in a plotting library

Example:
>>> histogram_outputs_to_fake_dataset(-0.5, 0.1, [1, 0, 2, 4, 1])
([-0.5, -0.3, -0.3, -0.2, -0.2, -0.2, -0.2, -0.1], [-0.5, -0.4, -0.3, -0.2, -0.1, 0])
Return type:

Tuple[List[float], List[float]]

alpyne.utils.limit(lower: int | float, value: int | float, upper: int | float) int | float[source]

Convenience function to constrain a given value between a lower and upper bound.

Parameters:
  • lower (int | float)

  • value (int | float)

  • upper (int | float)

Return type:

int | float

alpyne.utils.get_resources_path() Path[source]

Convenience method to return the resources directory in this project

Return type:

Path

alpyne.utils.get_wildcard_paths(model_dir: str) List[str][source]

Build wildcard references to the passed directory and all sub-directories

Parameters:

model_dir (str)

Return type:

List[str]

alpyne.utils.shorten_by_relativeness(paths: List[str]) List[str][source]

Update any paths where relative reference would take up less characters

Parameters:

paths (List[str])

Return type:

List[str]