Core API¶

The core module covers the simulation engine, hardware model, tasks, jobs, and the supporting enumerations and exceptions.

import pyschedsim as sim

engine = sim.Engine()
platform = engine.platform
cpu  = platform.add_processor_type("cpu", 1.0)
clock = platform.add_clock_domain(500.0, 2000.0)
power = platform.add_power_domain([(0, 0, 0.0, 100.0), (1, 0, 0.001, 10.0)])
platform.add_processor(cpu, clock, power)

For the equivalent C++ types, see Core Library.

Enumerations¶

ProcessorState¶

class ProcessorState¶

Represents the current execution state of a processor.

Idle¶: The processor has no job assigned and is not in a low-power state.

ContextSwitching¶: The processor is performing a context switch between jobs.

Running¶: The processor is actively executing a job.

Sleep¶: The processor is in a C-state (DPM sleep).

Changing¶: The processor’s clock domain is undergoing a DVFS frequency transition.

CStateScope¶

class CStateScope¶

Controls whether a C-state affects a single processor or the entire power domain.

PerProcessor¶: Value 0. The C-state is entered and exited per individual processor.

DomainWide¶: Value 1. The C-state is entered only when all processors in the power domain are idle, and all processors wake up together.

Exceptions¶

exception AlreadyFinalizedError¶

Inherits from RuntimeError.

Raised when attempting to modify a platform (add processors, tasks, etc.) after finalize() has been called.

exception InvalidStateError¶

Inherits from RuntimeError.

Raised when an operation is attempted in an invalid simulation state (e.g., scheduling on an un-finalized engine).

exception HandlerAlreadySetError¶

Inherits from RuntimeError.

Raised when a job-arrival handler is registered on a scheduler or allocator that already has a handler installed.

exception OutOfRangeError¶

Inherits from ValueError.

Raised when an index argument (e.g., platform.processor(idx)) is out of range.

exception AdmissionError¶

Inherits from ValueError.

Raised when a task cannot be admitted to a scheduler because the admission test fails (utilization bound exceeded).

exception LoaderError¶

Inherits from IOError.

Raised when a JSON platform or scenario file cannot be parsed or is structurally invalid.

Engine¶

class Engine¶

The central event-driven simulation loop. Engine owns the event priority queue and the Platform, advancing simulation time by dispatching events in chronological order.

Properties

time: float¶: Current simulation time in seconds (read-only).

platform: Platform¶: Reference to the platform owned by this engine (read-only).

context_switch_enabled: bool¶: Whether context switching is enabled (read-only).

energy_tracking_enabled: bool¶: Whether energy tracking is enabled (read-only). Energy tracking must be enabled before finalize() is called.

is_finalized: bool¶: Whether finalize() has been called (read-only).

Methods

run() → None¶: Run the simulation until the event queue is empty.

run(until: float) → None¶: Run until simulation time reaches until (seconds), then pause.

finalize() → None¶: Finalize the engine. Delegates to platform.finalize() internally. Must be called before creating schedulers.

schedule_job_arrival(task: Task, arrival_time: float, exec_time: float) → None¶: Schedule a single job arrival for task at absolute time arrival_time (seconds) with actual execution time exec_time (seconds).

enable_context_switch(enabled: bool) → None¶: Enable or disable context switching. Must be called before finalize().

enable_energy_tracking(enabled: bool) → None¶: Enable or disable energy tracking. Must be called before finalize().

processor_energy(proc_id: int) → float¶: Return the cumulative energy consumed by processor proc_id in mJ.

clock_domain_energy(cd_id: int) → float¶: Return the cumulative energy consumed by clock domain cd_id in mJ.

power_domain_energy(pd_id: int) → float¶: Return the cumulative energy consumed by power domain pd_id in mJ.

total_energy() → float¶: Return the total system energy consumed across all processors in mJ.

set_trace_writer(writer: TraceWriter | None) → None¶: Attach a trace writer to record simulation events. Pass None to detach. The writer can be set any time before engine.run().

Note

The C++ methods run(stop_condition), request_stop(), and the property stop_requested rely on a callable stop predicate and are not available in the Python bindings. Use run(until) to stop at a specific time, or run() to drain the event queue.

Platform¶

class Platform¶

Container for all hardware resources: processor types, clock domains, power domains, processors, and tasks. Factory methods must be called before finalize().

Properties

processor_type_count: int¶: Number of registered processor types (read-only).

processor_count: int¶: Number of registered processors (read-only).

clock_domain_count: int¶: Number of registered clock domains (read-only).

power_domain_count: int¶: Number of registered power domains (read-only).

task_count: int¶: Number of registered tasks (read-only).

reference_performance: float¶: The reference performance value computed at finalization (read-only). Equal to the maximum performance among all processor types.

is_finalized: bool¶: Whether finalize() has been called (read-only).

Methods

add_processor_type(name: str, performance: float, context_switch_delay: float = 0.0) → ProcessorType¶

Parameters:

name – Human-readable label for the processor type.
performance – Relative performance (1.0 = reference speed).
context_switch_delay – Context-switch overhead in seconds (default 0.0).

Returns:

A reference to the newly created ProcessorType.

add_clock_domain(freq_min: float, freq_max: float, transition_delay: float = 0.0) → ClockDomain¶

Parameters:

freq_min – Minimum frequency in MHz.
freq_max – Maximum frequency in MHz.
transition_delay – DVFS transition delay in seconds (default 0.0; set a non-zero value for realistic DVFS modelling).

Returns:

A reference to the newly created ClockDomain.

add_power_domain(cstates: list[tuple[int, int, float, float]]) → PowerDomain¶

Parameters:

cstates –

List of C-state descriptors. Each element is a 4-tuple (level, scope, wake_latency_s, power_mW) where:

level (int) — C-state index (0 = active/C0, 1+ = sleep).
scope (int) — 0 for PerProcessor, 1 for DomainWide. Pass an integer, not a CStateScope enum value.
wake_latency_s (float) — Wake-up latency in seconds.
power_mW (float) — Idle power draw in mW.

Returns:

A reference to the newly created PowerDomain.

Example:

power = platform.add_power_domain([
    (0, 0, 0.0,   100.0),   # C0: active, 100 mW
    (1, 0, 0.001,  10.0),   # C1: sleep, 1 ms wake, 10 mW
])

add_processor(type: ProcessorType, clock_domain: ClockDomain, power_domain: PowerDomain) → Processor¶

Add a processor to the platform.

Parameters:

type – The processor type (returned by add_processor_type).
clock_domain – The clock domain this processor belongs to.
power_domain – The power domain this processor belongs to.

Returns:

A reference to the newly created Processor.

add_task(period: float, relative_deadline: float, wcet: float) → Task¶

Add a periodic task with an automatically assigned ID.

Parameters:

period – Task period in seconds.
relative_deadline – Relative deadline in seconds.
wcet – Worst-case execution time in seconds.

Returns:

A reference to the newly created Task.

add_task(id: int, period: float, relative_deadline: float, wcet: float) → Task¶

Add a periodic task with an explicit ID.

Parameters:

id – Explicit task identifier (must be unique).
period – Task period in seconds.
relative_deadline – Relative deadline in seconds.
wcet – Worst-case execution time in seconds.

Returns:

A reference to the newly created Task.

finalize() → None¶: Lock the hardware configuration and compute derived values (reference performance, etc.). Must be called before creating schedulers. Raises AlreadyFinalizedError if called more than once.

processor_type(idx: int) → ProcessorType¶: Return the processor type at index idx. Raises OutOfRangeError if idx is out of range.

processor(idx: int) → Processor¶: Return the processor at index idx. Raises OutOfRangeError if idx is out of range.

clock_domain(idx: int) → ClockDomain¶: Return the clock domain at index idx. Raises OutOfRangeError if idx is out of range.

power_domain(idx: int) → PowerDomain¶: Return the power domain at index idx. Raises OutOfRangeError if idx is out of range.

task(idx: int) → Task¶: Return the task at index idx. Raises OutOfRangeError if idx is out of range.

Hardware Model¶

ProcessorType¶

class ProcessorType¶

Describes a class of processors sharing the same performance and context-switch characteristics. Created via Platform.add_processor_type().

id: int¶: Unique identifier assigned at creation (read-only).

performance: float¶: Relative performance factor (read-only). The reference performance is the maximum performance value across all types in the platform.

Processor¶

class Processor¶

A single processing unit. Created via Platform.add_processor().

Properties

id: int¶: Unique identifier assigned at creation (read-only).

state: ProcessorState¶: Current execution state (read-only). See ProcessorState.

current_cstate_level: int¶: Current C-state level (read-only). 0 means active (C0); higher values indicate deeper sleep states.

Methods

type() → ProcessorType¶: Return the ProcessorType of this processor.

clock_domain() → ClockDomain¶: Return the ClockDomain this processor belongs to.

power_domain() → PowerDomain¶: Return the PowerDomain this processor belongs to.

ClockDomain¶

class ClockDomain¶

A group of processors sharing a voltage/frequency rail. Created via Platform.add_clock_domain().

Properties

id: int¶: Unique identifier assigned at creation (read-only).

frequency: float¶: Current operating frequency in MHz (read-only).

freq_min: float¶: Minimum frequency in MHz (read-only).

freq_max: float¶: Maximum frequency in MHz (read-only).

is_locked: bool¶: Whether the frequency is locked (read-only). A locked clock domain ignores DVFS frequency-change requests.

is_transitioning: bool¶: Whether a DVFS frequency transition is currently in progress (read-only).

freq_eff: float¶: Effective frequency target in MHz (read-only). This is the frequency that has been requested but may not yet be reached if a transition is in progress.

Methods

set_freq_eff(freq: float) → None¶: Request a frequency change to freq (MHz). The actual transition takes transition_delay seconds if non-zero.

get_processors() → list[Processor]¶: Return all processors belonging to this clock domain.

set_power_coefficients(coeffs: list[float]) → None¶

Set polynomial coefficients for the power model P(f) = a0 + a1*f + a2*f² + a3*f³, where f is in MHz and P is in mW.

Parameters:: coeffs – A list of four coefficients [a0, a1, a2, a3].

PowerDomain¶

class PowerDomain¶

A group of processors sharing power-management state (C-states). Created via Platform.add_power_domain().

id: int¶: Unique identifier assigned at creation (read-only).

Tasks and Jobs¶

Task¶

class Task¶

A periodic real-time task. Created via Platform.add_task().

id: int¶: Unique identifier (read-only).

period: float¶: Task period in seconds (read-only).

relative_deadline: float¶: Relative deadline in seconds (read-only).

wcet: float¶: Worst-case execution time in seconds (read-only).

Job¶

class Job¶

A single execution instance of a task. Jobs are created internally by the engine when a job-arrival event fires.

Properties

remaining_work: float¶: Remaining execution time in seconds (read-only).

total_work: float¶: Total execution time assigned to this job instance in seconds (read-only).

absolute_deadline: float¶: Absolute deadline in seconds (read-only).

is_complete: bool¶: Whether the job has finished executing (read-only).

Methods

task() → Task¶: Return the Task this job belongs to.

Helper Functions¶

get_all_processors(engine: Engine) → list[Processor]¶: Return a list of all processors in the engine’s platform, ordered by ID. Equivalent to iterating platform.processor(i) for i in range(platform.processor_count).

get_all_tasks(engine: Engine) → list[Task]¶: Return a list of all tasks in the engine’s platform, ordered by ID. Equivalent to iterating platform.task(i) for i in range(platform.task_count).