Activities Module

Activity framework and state machine components for modeling complex entity lifecycles.

Activity

Time-based operations with capacity constraints.

class simcraft.activities.activity.Activity[source]

Bases: Generic[T]

Time-based activity with capacity and resource synchronization.

Activities model operations that take time, with support for: - Capacity constraints - External resource requirements - State-based lifecycle management - Statistics collection

The activity lifecycle is: 1. PENDING: Entity is waiting to start 2. READY_TO_START: Resources available, can start 3. PROCESSING: Currently being processed 4. COMPLETED: Processing done 5. READY_TO_FINISH: Can depart (external signal) 6. READY_TO_DEPART: Ready to leave

Parameters:

  • sim (Simulation) – Parent simulation

  • capacity (int) – Maximum concurrent entities

  • processing_time (Union[float, Callable[[T], float]]) – Fixed time or function returning processing time

  • need_external_start (bool) – Whether external signal needed to start

  • need_external_finish (bool) – Whether external signal needed to finish

  • name (str) – Activity name

Examples

>>> def processing_time(job):
...     return job.size * 0.1
...
>>> activity = Activity(
...     sim,
...     capacity=2,
...     processing_time=processing_time,
...     name="Assembly"
... )
>>>
>>> activity.on_departure(lambda job: output_queue.enqueue(job))
>>> activity.request_to_start(job)
__init__(sim, capacity=1, processing_time=1.0, need_external_start=False, need_external_finish=False, name='')[source]

Initialize activity.

Parameters:

  • sim (Simulation)

  • capacity (int)

  • processing_time (Any)

  • need_external_start (bool)

  • need_external_finish (bool)

  • name (str)

Return type:

None

property name: str

Get activity name.

property capacity: int

Get activity capacity.

property stats: ActivityStats

Get activity statistics.

property occupancy: int

Get current occupancy (processing + completed + ready_to_finish).

property available_capacity: int

Get available capacity.

property pending_count: int

Get number of pending entities.

property processing_count: int

Get number of entities in processing.

request_to_start(entity)[source]

Request an entity to start the activity.

Parameters:

entity (T) – Entity requesting to start

Return type:

None

try_start(entity)[source]

External signal that entity can start.

Parameters:

entity (T) – Entity to start

Returns:

True if entity was found and can start

Return type:

bool

try_finish(entity)[source]

External signal that entity can finish.

Parameters:

entity (T) – Entity to finish

Returns:

True if entity was found and can finish

Return type:

bool

get_pending()[source]

Get list of pending entities.

Return type:

List[T]

get_processing()[source]

Get list of entities in processing.

Return type:

List[T]

get_completed()[source]

Get list of completed entities.

Return type:

List[T]

on_start(callback)[source]

Set callback for processing start.

Parameters:

callback (Callable[[T], None])

Return type:

None

on_complete(callback)[source]

Set callback for processing complete.

Parameters:

callback (Callable[[T], None])

Return type:

None

on_departure(callback)[source]

Set callback for entity departure.

Parameters:

callback (Callable[[T], None])

Return type:

None

reset_stats()[source]

Reset activity statistics.

Return type:

None

class simcraft.activities.activity.ActivityState[source]

Bases: Enum

States for activity lifecycle.

PENDING = 1
READY_TO_START = 2
PROCESSING = 3
COMPLETED = 4
READY_TO_FINISH = 5
READY_TO_DEPART = 6
class simcraft.activities.activity.ActivityStats[source]

Bases: object

Statistics for activity performance.

entries

Total entities that entered the activity

Type:

int

completions

Total entities that completed

Type:

int

departures

Total entities that departed

Type:

int

total_processing_time

Sum of all processing times

Type:

float

total_wait_time

Sum of all wait times

Type:

float

entries: int = 0
completions: int = 0
departures: int = 0
total_processing_time: float = 0.0
total_wait_time: float = 0.0
record_entry(time)[source]

Record entity entering waiting state.

Parameters:

time (float)

Return type:

None

record_start(time, wait_time)[source]

Record entity starting processing.

Parameters:
Return type:

None

record_completion(time, processing_time)[source]

Record entity completing processing.

Parameters:
Return type:

None

record_departure(time)[source]

Record entity departing.

Parameters:

time (float)

Return type:

None

property average_processing: float

Get time-average number in processing.

property average_waiting: float

Get time-average number waiting.

property average_processing_time: float

Get average processing time per entity.

property average_wait_time: float

Get average wait time per entity.

__init__(entries=0, completions=0, departures=0, total_processing_time=0.0, total_wait_time=0.0, _area_processing=0.0, _area_waiting=0.0, _last_change_time=0.0, _current_processing=0, _current_waiting=0)
Parameters:

  • entries (int)

  • completions (int)

  • departures (int)

  • total_processing_time (float)

  • total_wait_time (float)

  • _area_processing (float)

  • _area_waiting (float)

  • _last_change_time (float)

  • _current_processing (int)

  • _current_waiting (int)

Return type:

None

class simcraft.activities.activity.ParallelActivity[source]

Bases: Generic[T]

Parallel activities with shared waiting queue.

Useful for modeling parallel processing stations that share a common input queue.

Parameters:

  • sim (Simulation) – Parent simulation

  • num_parallel (int) – Number of parallel activities

  • processing_time (Any) – Processing time per activity

Examples

>>> parallel = ParallelActivity(sim, num_parallel=3, processing_time=5.0)
>>> parallel.request_to_start(job)  # Goes to first available
__init__(sim, num_parallel, processing_time=1.0, name='')[source]

Initialize parallel activities.

Parameters:
Return type:

None

request_to_start(entity)[source]

Request entity to start on any available activity.

Parameters:

entity (T) – Entity to process

Return type:

None

on_departure(callback)[source]

Set callback for entity departure.

Parameters:

callback (Callable[[T], None])

Return type:

None

property total_capacity: int

Get total capacity across all activities.

property available_capacity: int

Get total available capacity.

property pending_count: int

Get number of entities waiting.

State Machine

State machine framework for complex entity behavior.

class simcraft.activities.state_machine.State[source]

Bases: object

A state in a state machine.

name

State identifier

Type:

str

on_enter

Called when entering the state

Type:

Optional[Callable]

on_exit

Called when exiting the state

Type:

Optional[Callable]

on_stay

Called while staying in the state

Type:

Optional[Callable]

is_initial

Whether this is an initial state

Type:

bool

is_final

Whether this is a final state

Type:

bool

metadata

Additional state metadata

Type:

Dict

name: str
on_enter: Callable[[Any], None] | None = None
on_exit: Callable[[Any], None] | None = None
on_stay: Callable[[Any], None] | None = None
is_initial: bool = False
is_final: bool = False
metadata: Dict[str, Any]
__init__(name, on_enter=None, on_exit=None, on_stay=None, is_initial=False, is_final=False, metadata=<factory>)
Parameters:
Return type:

None

class simcraft.activities.state_machine.Transition[source]

Bases: object

A transition between states.

source

Source state name

Type:

str

target

Target state name

Type:

str

trigger

Event that triggers the transition

Type:

str

guard

Condition that must be true for transition

Type:

Optional[Callable]

action

Action to perform during transition

Type:

Optional[Callable]

priority

Priority when multiple transitions match (higher = first)

Type:

int

source: str
target: str
trigger: str = ''
guard: Callable[[Any], bool] | None = None
action: Callable[[Any], None] | None = None
priority: int = 0
can_fire(context)[source]

Check if transition can fire.

Parameters:

context (Any)

Return type:

bool

fire(context)[source]

Execute transition action.

Parameters:

context (Any)

Return type:

None

__init__(source, target, trigger='', guard=None, action=None, priority=0)
Parameters:
Return type:

None

class simcraft.activities.state_machine.StateMachine[source]

Bases: Generic[T]

Flexible state machine for entity lifecycle management.

Supports hierarchical states, guards, actions, and timed transitions.

Parameters:

  • sim (Simulation) – Parent simulation

  • name (str) – State machine name

Examples

>>> sm = StateMachine(sim, name="OrderProcess")
>>>
>>> # Define states
>>> sm.add_state("created", is_initial=True)
>>> sm.add_state("processing")
>>> sm.add_state("shipped")
>>> sm.add_state("delivered", is_final=True)
>>>
>>> # Define transitions
>>> sm.add_transition("created", "processing", trigger="start")
>>> sm.add_transition("processing", "shipped",
...                   trigger="complete",
...                   guard=lambda ctx: ctx.payment_verified)
>>> sm.add_transition("shipped", "delivered", trigger="arrive")
>>>
>>> # Create instance and process
>>> order = Order(id=1)
>>> instance = sm.create_instance(order)
>>> instance.trigger("start")  # -> processing
>>> instance.trigger("complete")  # -> shipped (if payment verified)
__init__(sim, name='')[source]

Initialize state machine.

Parameters:
Return type:

None

property name: str

Get state machine name.

property states: List[str]

Get list of state names.

property initial_state: str | None

Get initial state name.

add_state(name, on_enter=None, on_exit=None, is_initial=False, is_final=False, **metadata)[source]

Add a state to the machine.

Parameters:

  • name (str) – State identifier

  • on_enter (Optional[Callable]) – Called when entering state

  • on_exit (Optional[Callable]) – Called when exiting state

  • is_initial (bool) – Whether this is the initial state

  • is_final (bool) – Whether this is a final state

  • **metadata – Additional state metadata

Returns:

The created state

Return type:

State

add_transition(source, target, trigger='', guard=None, action=None, priority=0)[source]

Add a transition between states.

Parameters:

  • source (str) – Source state name

  • target (str) – Target state name

  • trigger (str) – Event trigger name (empty for automatic)

  • guard (Optional[Callable]) – Condition for transition

  • action (Optional[Callable]) – Action to perform during transition

  • priority (int) – Priority (higher = checked first)

Returns:

The created transition

Return type:

Transition

add_timed_transition(source, target, duration, action=None)[source]

Add a timed transition that fires after a duration.

Parameters:

  • source (str) – Source state name

  • target (str) – Target state name

  • duration (Union[float, Callable]) – Time to wait or function returning time

  • action (Optional[Callable]) – Action to perform

Returns:

The created transition

Return type:

Transition

get_state(name)[source]

Get a state by name.

Parameters:

name (str) – State name

Returns:

The state or None

Return type:

Optional[State]

get_transitions_from(state)[source]

Get all transitions from a state.

Parameters:

state (str) – Source state name

Returns:

Transitions from the state

Return type:

List[Transition]

create_instance(context)[source]

Create a new state machine instance.

Parameters:

context (T) – Context object (entity) for this instance

Returns:

New instance in initial state

Return type:

StateMachineInstance

on_state_enter(callback)[source]

Set global callback for state entry.

Parameters:

callback (Callable[[str, T], None])

Return type:

None

on_state_exit(callback)[source]

Set global callback for state exit.

Parameters:

callback (Callable[[str, T], None])

Return type:

None

on_transition(callback)[source]

Set global callback for transitions.

Parameters:

callback (Callable[[str, str, T], None])

Return type:

None

class simcraft.activities.state_machine.StateMachineInstance[source]

Bases: Generic[T]

Instance of a state machine for a specific entity.

Parameters:

  • machine (StateMachine) – Parent state machine definition

  • context (T) – Context object (entity)

__init__(machine, context)[source]

Initialize instance.

Parameters:
Return type:

None

property current_state: str | None

Get current state name.

property context: T

Get context object.

property history: List[Tuple[float, str, str]]

Get transition history (time, from_state, to_state).

property is_in_final_state: bool

Check if in a final state.

trigger(event)[source]

Trigger an event to potentially cause a transition.

Parameters:

event (str) – Event name

Returns:

True if a transition occurred

Return type:

bool

can_trigger(event)[source]

Check if an event can cause a transition.

Parameters:

event (str) – Event name

Returns:

True if a transition would occur

Return type:

bool

force_state(state)[source]

Force transition to a state (bypassing guards).

Parameters:

state (str) – Target state name

Return type:

None