Simulation

dsbx.Sim

Simulation kernel for dynamic scheduling environments.

This package provides a higher-level simulator (DynaSchedSim) that consumes event streams generated by dsbx.Gen and exposes a rich snapshot interface for downstream solvers and agents.

class dsbx.Sim.Snapshot(**data)

Bases: BaseModel

Top-level snapshot returned by the simulator.

It aims to be sufficiently rich for OR/MILP, RL, and LLM-based agents, while remaining serialization-friendly (JSON-compatible).

Parameters:

  • time (float)

  • prev_decision_time (float)

  • horizon (float)

  • lookahead_horizon (float)

  • scenario_id (str | None)

  • seed (int | None)

  • config_hash (str | None)

  • plant (Dict[str, Any] | None)

  • scale (Dict[str, Any] | None)

  • targets (Dict[str, Any] | None)

  • dynamics (Dict[str, Any] | None)

  • dynamic_scenarios (Dict[str, Any] | None)

  • jobs (List[JobSnapshot])

  • machines (List[MachineSnapshot])

  • pending_events (List[EventSnapshot])

  • system_stats (SystemStats)

  • stability_stats (StabilityStats)

time: float
prev_decision_time: float
horizon: float
lookahead_horizon: float
scenario_id: str | None
seed: int | None
config_hash: str | None
plant: Dict[str, Any] | None
scale: Dict[str, Any] | None
targets: Dict[str, Any] | None
dynamics: Dict[str, Any] | None
dynamic_scenarios: Dict[str, Any] | None
jobs: List[JobSnapshot]
machines: List[MachineSnapshot]
pending_events: List[EventSnapshot]
system_stats: SystemStats
stability_stats: StabilityStats
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class dsbx.Sim.DynaSchedSim(model, events)

Bases: object

Core simulator for dynamic scheduling instances.

The simulator consumes event streams generated by dsbx.Gen, maintains explicit job and machine state, and exposes rich snapshots at decision points.

The implementation follows the same time-advance style as the legacy LLM scheduling environment: time is the decision reference time, job and machine availability controls when the next operation can start, dynamic events become active when the clock reaches their timestamp, and snapshots keep planned start/end times for optimization and visualization.

Parameters:
reset()

Reset internal time/state and return the initial snapshot.

This does not regenerate events; it assumes the caller provides a fresh event list when creating a new simulator instance.

Return type:

Snapshot

step_action(action)

Apply a single scheduling decision and return the new snapshot.

Parameters:

action (Dict[str, Any]) – A dictionary describing the scheduling decision. By convention, it must contain at least: - job_id (str) - machine_group (str) It may optionally include: - machine_id (str): specific machine choice within the group.

Return type:

Snapshot

advance_to_next_decision_point()

Advance simulation time to the next moment where a decision is needed.

If there are already ready operations at the current time, this is a no-op and simply returns the current snapshot.

Return type:

Snapshot

export_light_state()
Return type:

Dict[str, Any]

estimate_action_score(action)

Estimate the quality of an action based on its immediate end time.

The logic matches LLMEnv.estimate_action_score: estimate the completion time of the action on the selected machine, return the negative end time so earlier is better, and add a small priority correction.

Return type:

float

Parameters:

action (Dict[str, Any])

quick_rollout_score(action, steps=1)

Approximate makespan impact of an action via a short, greedy rollout.

The behavior matches LLMEnv.quick_rollout_score: execute the action on a lightweight local copy, greedily schedule several subsequent ready operations, and return the negative estimated completion time without modifying the main simulator state.

Return type:

float

Parameters:

  • action (Dict[str, Any])

  • steps (int)

export_snapshot()

Build a rich Snapshot object from the current SystemState.

Return type:

Snapshot

get_ready_operations()

Return a shallow copy of the current ready-operations list.

Downstream components (e.g. environments, agents) should prefer this method over the internal _get_ready_operations() helper.

Return type:

List[OperationState]

get_action_timing(action)

Return an estimated timing profile for action under the current state.

The result is computed using the same instantaneous semantics as estimate_action_score(), but instead of a scalar score it returns a dictionary with:

  • job_id: job identifier (str)

  • op_index: index of the operation being scheduled (int)

  • machine_id: selected concrete machine id (str)

  • start_time: earliest feasible start time given all constraints

  • end_time: corresponding end time on the selected machine

  • proc_time_raw: realized processing time before speed adjustment

  • machine_speed: speed of the selected machine

  • proc_time_effective: actual processing time on the machine

The simulation state is not mutated.

Return type:

Optional[Dict[str, float]]

Parameters:

action (Dict[str, Any])

get_gantt()

Return an immutable copy of the internal Gantt list.

Return type:

List[Dict[str, Any]]

get_events()

Return the underlying event list (useful for debugging/vis).

Return type:

List[Union[ArrivalEvent, DueDateEvent, BreakdownEvent, PriorityChangeEvent, OrderCancellationEvent, MachineRepairCompletionEvent, ProcessTimeChangeEvent, PreventiveMaintenanceEvent, RouteChangeEvent, DueDateChangeEvent]]

dsbx.Sim.load_events_from_jsonl(events_file)

Load all events from an events.jsonl file.

Parameters:

events_file (Path) – Path to events.jsonl file

Return type:

List[Union[ArrivalEvent, DueDateEvent, BreakdownEvent, PriorityChangeEvent, OrderCancellationEvent, MachineRepairCompletionEvent, ProcessTimeChangeEvent, PreventiveMaintenanceEvent, RouteChangeEvent, DueDateChangeEvent]]

Returns:

List of Event objects, sorted by time

dsbx.Sim.load_instance_from_directory(run_dir)

Load InputModel and events from a run directory.

Looks for input.json and events.jsonl in the given directory.

Parameters:

run_dir (Path) – Path to the run directory (e.g., runs/cold_start/my_instance)

Return type:

Tuple[InputModel, List[Union[ArrivalEvent, DueDateEvent, BreakdownEvent, PriorityChangeEvent, OrderCancellationEvent, MachineRepairCompletionEvent, ProcessTimeChangeEvent, PreventiveMaintenanceEvent, RouteChangeEvent, DueDateChangeEvent]]]

Returns:

Tuple of (InputModel, events list)

Example

>>> model, events = load_instance_from_directory(Path("runs/cold_start/my_instance"))
>>> from dsbx.Sim import DynaSchedSim
>>> sim = DynaSchedSim(model, events)
>>> snapshot = sim.reset()
dsbx.Sim.find_instance_directory(events_file)

Find the run directory containing an events.jsonl file.

Parameters:

events_file (Path) – Path to events.jsonl (can be absolute or relative)

Return type:

Path

Returns:

Path to the directory containing events.jsonl