I/O and Scenario Management

The pyschedsim module exposes functions and data classes for loading platform definitions, loading and generating workload scenarios, injecting tasks into the engine, and writing scenarios back to disk.

For the full C++ I/O reference, see I/O Library.

Platform Loading

load_platform(engine: Engine, path: str) → None

Load a platform definition from a JSON file and configure the engine.

The platform must not have been finalized before calling this function. Both the new component-based format and the legacy cluster-based format are supported; the loader detects the format automatically. See JSON Formats for a description of both formats.

Parameters:

  • engine – The simulation engine to configure.

  • path – Path to a platform JSON file.

Raises:

LoaderError – If the file cannot be opened or the JSON is invalid.

load_platform_from_string(engine: Engine, json: str) → None

Load a platform definition from a JSON string.

Identical to load_platform() but reads from an in-memory string instead of a file. Useful for testing or dynamically constructed configurations.

Parameters:

  • engine – The simulation engine to configure.

  • json – A JSON string in platform format.

Raises:

LoaderError – If the JSON is invalid or missing required fields.

Scenario Loading

Data Classes

class ScenarioData

Container for a complete task set with job arrivals.

tasks: list[TaskParams]

The list of tasks in this scenario. Each element describes one periodic real-time task and its concrete job instances.

class TaskParams

Parameters for one periodic real-time task.

id: int

Task identifier. 1-based by convention; preserved through load and inject operations.

period: float

Inter-arrival period in seconds. Must be positive.

relative_deadline: float

Relative deadline in seconds. Defaults to period (implicit deadline) when loading from JSON if the field is absent.

wcet: float

Worst-case execution time in seconds. Must be positive.

jobs: list[JobParams]

Concrete job instances for this task, sorted by arrival time.

class JobParams

Parameters for a single job instance.

arrival: float

Absolute arrival time in seconds.

duration: float

Actual execution demand for this job in seconds. Must be positive. May exceed wcet — the loader does not enforce this constraint.

Loading Functions

load_scenario(path: str) → ScenarioData

Load a scenario (task set with job arrivals) from a JSON file.

The returned ScenarioData object owns its data and can be modified before injection. See JSON Formats for the scenario file format.

Parameters:

path – Path to a scenario JSON file.

Returns:

A ScenarioData populated from the file.

Raises:

LoaderError – If the file cannot be opened or the JSON is invalid.

load_scenario_from_string(json: str) → ScenarioData

Load a scenario from a JSON string.

Identical to load_scenario() but reads from an in-memory string.

Parameters:

json – A JSON string in scenario format.

Returns:

A ScenarioData populated from the string.

Raises:

LoaderError – If the JSON is invalid or missing required fields.

Scenario Injection and Scheduling

Ordering constraints

  • inject_scenario() must be called before platform.finalize() because it adds tasks to the platform.

  • schedule_arrivals() can be called either before or after platform.finalize().

  • The scheduler and allocator must be created after finalize.

inject_scenario(engine: Engine, scenario: ScenarioData) → list[Task]

Create Task objects in the platform from scenario data.

For each TaskParams in scenario, a task is added to the platform with the corresponding period, relative_deadline, and wcet. The task IDs from the scenario are preserved.

Must be called before platform.finalize().

Parameters:

  • engine – The simulation engine.

  • scenario – A loaded or generated ScenarioData.

Returns:

A list of Task references, one per task in scenario, in the same order as scenario.tasks.

schedule_arrivals(engine: Engine, task: Task, jobs: list[JobParams]) → None

Schedule job arrivals for a task.

For each JobParams in jobs, a job arrival event is registered in the engine’s event queue at the specified time with the given execution demand.

Can be called before or after platform.finalize().

Parameters:

  • engine – The simulation engine.

  • task – The task to schedule arrivals for.

  • jobs – The list of job parameters to schedule.

Writing Scenarios

write_scenario(scenario: ScenarioData, path: str) → None

Write a scenario to a JSON file.

The output always uses the canonical wcet field (never utilization). Jobs are written in ascending arrival-time order.

Parameters:

  • scenario – The ScenarioData to serialize.

  • path – Output file path. The file is created or overwritten.

Scenario Generation

class WeibullJobConfig

Parameters controlling the Weibull distribution used to sample job execution times relative to the task WCET.

success_rate: float

Fraction of jobs that complete within their WCET, in (0, 1]. A value of 1.0 means all jobs finish at or before their WCET; lower values introduce overruns.

compression_rate: float

Lower bound for generated execution times as a fraction of WCET, in [0, 1]. For example, 0.5 prevents durations below half the task WCET.

from_utilizations(utilizations: list[float], success_rate: float = 1.0, compression_rate: float = 1.0, seed: int = 42, num_hyperperiods: int = 1, min_jobs: int = 0) → ScenarioData

Generate a random task set from a list of per-task utilization values.

One task is created per entry in utilizations. Periods are selected uniformly from the fixed 100–500 microsecond grid, WCETs are computed as period * utilization, and job execution times are sampled from a Weibull distribution controlled by success_rate and compression_rate. The legacy num_hyperperiods name controls how many nominal 500 microsecond generation windows are produced.

Parameters:

  • utilizations – List of per-task utilization values, each in (0, 1].

  • success_rate – Fraction of jobs completing within WCET, in (0, 1] (default 1.0).

  • compression_rate – Minimum duration as a fraction of WCET, in [0, 1] (default 1.0).

  • seed – RNG seed for reproducibility (default 42).

  • num_hyperperiods – Number of nominal generation windows (default 1).

  • min_jobs – Minimum total job count across all tasks. Period selection is retried up to 1,000 times when necessary (default 0, disabled).

Returns:

A ScenarioData with one task per utilization entry.

Note

The C++ API provides additional scenario generation functions (generate_scenario, uunifast_discard) that are not exposed in the Python bindings. However, from_utilizations() covers most common use cases. See I/O Library for the full C++ I/O reference.

Examples

Full Workflow: Load, Inject, and Schedule

The typical three-step process for running a scenario from a JSON file:

import pyschedsim as sim

engine = sim.Engine()

# Step 1: Load the hardware platform
sim.load_platform(engine, "platforms/exynos5422LITTLE.json")

# Step 2: Load the scenario (task set with job arrivals)
scenario = sim.load_scenario("scenario.json")

# Step 3: Inject tasks into the platform -- must be BEFORE finalize
tasks = sim.inject_scenario(engine, scenario)

# Step 4: Schedule job arrivals for every task
for i, tp in enumerate(scenario.tasks):
    sim.schedule_arrivals(engine, tasks[i], tp.jobs)

# Step 5: Finalize the platform and build the scheduler
engine.platform.finalize()

procs = sim.get_all_processors(engine)
scheduler = sim.EdfScheduler(engine, procs)
for task in tasks:
    scheduler.add_server(task)

allocator = sim.SingleSchedulerAllocator(engine, scheduler)

writer = sim.MemoryTraceWriter()
engine.set_trace_writer(writer)
engine.run()

metrics = writer.compute_metrics()
print(f"Completed: {metrics.completed_jobs}/{metrics.total_jobs}")
print(f"Deadline misses: {metrics.deadline_misses}")

Programmatic Generation with from_utilizations

When you do not have a pre-recorded scenario file, generate one directly from a list of per-task utilizations:

import pyschedsim as sim

# Three tasks with utilizations 0.3, 0.25, and 0.2
scenario = sim.from_utilizations(
    [0.3, 0.25, 0.2],
    success_rate=0.95,      # 5 % of jobs may overrun
    compression_rate=0.5,  # durations are at least 50 % of WCET
    seed=42,
    num_hyperperiods=100,
    min_jobs=700,
)

print(f"Generated {len(scenario.tasks)} tasks")
for tp in scenario.tasks:
    print(f"  Task {tp.id}: period={tp.period:.4f}s  "
          f"wcet={tp.wcet:.4f}s  jobs={len(tp.jobs)}")

# Inject and simulate as usual
engine = sim.Engine()
sim.load_platform(engine, "platforms/exynos5422LITTLE.json")
tasks = sim.inject_scenario(engine, scenario)
for i, tp in enumerate(scenario.tasks):
    sim.schedule_arrivals(engine, tasks[i], tp.jobs)

engine.platform.finalize()

Writing a Modified Scenario

Scenarios can be serialised back to JSON after loading or programmatic construction:

import pyschedsim as sim

# Load, optionally modify, then save
scenario = sim.load_scenario("original.json")

# Example: trim jobs beyond 1 second
for tp in scenario.tasks:
    tp.jobs = [j for j in tp.jobs if j.arrival < 1.0]

sim.write_scenario(scenario, "trimmed.json")

See Also