Skip to content

World Simulation

Mission simulation tiers — single-sat (Tier 1), multi-sat handover (Tier 2), and network traffic (Tier 3).

Tier 1: Simple World Sim

SimpleWorldSim 

SimpleWorldSim()

Tier 1 world simulation: single satellite to terminal time-series.

Evaluates a snapshot link budget at each time step along a pre-computed satellite pass, applying operational policy checks (minimum elevation) and collecting margin, throughput, and outage statistics.

Examples:

>>> sim = SimpleWorldSim()
>>> result = sim.run(link_inputs, trajectory, OpsPolicy(), env)
>>> result.summary["availability"]
0.95
Source code in src/opensatcom/world/sim.py 
31
32
def __init__(self) -> None:
    self._engine = DefaultLinkEngine()

run 

run(link_inputs, trajectory, ops, env, doppler_hz=None)

Run the Tier 1 simulation over a satellite pass.

Parameters:

Name Type Description Default
link_inputs LinkInputs

Link configuration (terminals, antennas, propagation, RF chain).

 required
trajectory PrecomputedTrajectory

Pre-computed satellite trajectory with elevation, azimuth, and range arrays.

 required
ops OpsPolicy

Operational constraints (min elevation, max scan angle).

 required
env StaticEnvironmentProvider

Environment provider for propagation conditions at each step.

 required
doppler_hz ndarray or None

Pre-computed Doppler shift time series in Hz. If provided, included in the output.

 None

Returns:

Type Description
WorldSimOutputs

Time-series results with margin, throughput, outage mask, and scalar summary statistics.
Source code in src/opensatcom/world/sim.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def run(
    self,
    link_inputs: LinkInputs,
    trajectory: PrecomputedTrajectory,
    ops: OpsPolicy,
    env: StaticEnvironmentProvider,
    doppler_hz: np.ndarray | None = None,
) -> WorldSimOutputs:
    """Run the Tier 1 simulation over a satellite pass.

    Parameters
    ----------
    link_inputs : LinkInputs
        Link configuration (terminals, antennas, propagation, RF chain).
    trajectory : PrecomputedTrajectory
        Pre-computed satellite trajectory with elevation, azimuth,
        and range arrays.
    ops : OpsPolicy
        Operational constraints (min elevation, max scan angle).
    env : StaticEnvironmentProvider
        Environment provider for propagation conditions at each step.
    doppler_hz : numpy.ndarray or None
        Pre-computed Doppler shift time series in Hz. If provided,
        included in the output.

    Returns
    -------
    WorldSimOutputs
        Time-series results with margin, throughput, outage mask,
        and scalar summary statistics.
    """
    pd = trajectory.pass_data
    n_steps = len(pd.times_s)

    margins = np.zeros(n_steps)
    throughputs = np.zeros(n_steps)
    outages = np.zeros(n_steps, dtype=bool)
    modcod_names: list[str] = []
    breakdown_ts: dict[str, np.ndarray] = {}

    for i in range(n_steps):
        elev = float(pd.elev_deg[i])
        az = float(pd.az_deg[i])
        rng = float(pd.range_m[i])
        t = float(pd.times_s[i])

        # Ops policy check
        if elev < ops.min_elevation_deg:
            outages[i] = True
            margins[i] = float("nan")
            throughputs[i] = 0.0
            modcod_names.append("")
            continue

        cond = env.conditions(t, link_inputs.tx_terminal, link_inputs.rx_terminal)
        out = self._engine.evaluate_snapshot(elev, az, rng, link_inputs, cond)

        margins[i] = out.margin_db
        if out.margin_db < 0:
            outages[i] = True

        if out.throughput_mbps is not None:
            throughputs[i] = out.throughput_mbps
        else:
            throughputs[i] = 0.0

        if out.breakdown is not None:
            modcod_names.append(
                out.breakdown.get("selected_modcod", "")  # type: ignore[arg-type]
                if "selected_modcod" in out.breakdown
                else ""
            )
        else:
            modcod_names.append("")

        # Collect breakdown time-series
        if out.breakdown is not None:
            for key, val in out.breakdown.items():
                if key not in breakdown_ts:
                    breakdown_ts[key] = np.full(n_steps, float("nan"))
                breakdown_ts[key][i] = val

    # Compute summary
    valid_mask = ~outages
    n_valid = int(np.sum(valid_mask))
    dt_s = float(pd.times_s[1] - pd.times_s[0]) if n_steps > 1 else 1.0
    outage_seconds = float(np.sum(outages)) * dt_s
    availability = n_valid / n_steps if n_steps > 0 else 0.0

    valid_margins = margins[valid_mask] if n_valid > 0 else np.array([0.0])

    summary: dict[str, float] = {
        "availability": availability,
        "outage_seconds": outage_seconds,
        "margin_db_min": float(np.nanmin(valid_margins)) if n_valid > 0 else 0.0,
        "margin_db_mean": float(np.nanmean(valid_margins)) if n_valid > 0 else 0.0,
        "margin_db_p05": float(np.nanpercentile(valid_margins, 5)) if n_valid > 0 else 0.0,
        "margin_db_p50": float(np.nanpercentile(valid_margins, 50)) if n_valid > 0 else 0.0,
        "margin_db_p95": float(np.nanpercentile(valid_margins, 95)) if n_valid > 0 else 0.0,
        "throughput_mbps_mean": float(np.mean(throughputs[valid_mask])) if n_valid > 0 else 0.0,
    }

    return WorldSimOutputs(
        times_s=pd.times_s,
        elev_deg=pd.elev_deg,
        range_m=pd.range_m,
        margin_db=margins,
        throughput_mbps=throughputs if link_inputs.modem is not None else None,
        selected_modcod=modcod_names if link_inputs.modem is not None else None,
        outages_mask=outages,
        summary=summary,
        breakdown_timeseries=breakdown_ts if breakdown_ts else None,
        az_deg=pd.az_deg,
        doppler_hz=doppler_hz,
    )

Tier 2: Multi-Sat World Sim

MultiSatWorldSim 

MultiSatWorldSim(handover_policy=None)

Tier 2 world simulation: multiple satellites with handover.

Evaluates link budget for each satellite at each timestep, applies handover policy to select serving satellite, and produces combined time-series output.

Parameters:

Name Type Description Default
handover_policy HandoverPolicy | None

Handover policy. Defaults to hysteresis-based with values from OpsPolicy.

 None
Source code in src/opensatcom/world/multisim.py 
65
66
67
def __init__(self, handover_policy: HandoverPolicy | None = None) -> None:
    self._engine = DefaultLinkEngine()
    self._policy = handover_policy

run 

run(link_inputs, trajectories, ops, env)

Run multi-satellite simulation.

Parameters:

Name Type Description Default
link_inputs LinkInputs

Base link inputs (shared antenna, RF chain, scenario).

 required
trajectories dict[str, PrecomputedTrajectory]

Satellite trajectories keyed by satellite ID.

 required
ops OpsPolicy

Operational policy with min elevation and handover hysteresis.

 required
env StaticEnvironmentProvider

Environment conditions provider.

 required

Returns:

Type Description
MultiSatWorldSimOutputs

Combined time-series output including per-satellite contact times and handover events.
Source code in src/opensatcom/world/multisim.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def run(
    self,
    link_inputs: LinkInputs,
    trajectories: dict[str, PrecomputedTrajectory],
    ops: OpsPolicy,
    env: StaticEnvironmentProvider,
) -> MultiSatWorldSimOutputs:
    """Run multi-satellite simulation.

    Parameters
    ----------
    link_inputs : LinkInputs
        Base link inputs (shared antenna, RF chain, scenario).
    trajectories : dict[str, PrecomputedTrajectory]
        Satellite trajectories keyed by satellite ID.
    ops : OpsPolicy
        Operational policy with min elevation and handover hysteresis.
    env : StaticEnvironmentProvider
        Environment conditions provider.

    Returns
    -------
    MultiSatWorldSimOutputs
        Combined time-series output including per-satellite contact
        times and handover events.
    """
    if not trajectories:
        raise ValueError("At least one satellite trajectory required")

    sat_ids = list(trajectories.keys())
    n_sats = len(sat_ids)

    # Use first trajectory to determine time grid
    first_traj = trajectories[sat_ids[0]]
    times_s = first_traj.pass_data.times_s
    n_steps = len(times_s)

    # Set up handover policy
    policy = self._policy or HandoverPolicy(
        hysteresis_db=3.0,
        hysteresis_s=ops.handover_hysteresis_s,
    )
    policy.reset(initial_sat_idx=0)

    # Pre-evaluate all satellites at all timesteps
    # margins[sat_idx][step_idx] and elevations
    all_margins = np.full((n_sats, n_steps), float("nan"))
    all_elevations = np.full((n_sats, n_steps), 0.0)
    all_throughputs = np.full((n_sats, n_steps), 0.0)
    all_visible = np.zeros((n_sats, n_steps), dtype=bool)

    for s_idx, sat_id in enumerate(sat_ids):
        traj = trajectories[sat_id]
        pd = traj.pass_data
        for i in range(n_steps):
            elev = float(pd.elev_deg[i])
            az = float(pd.az_deg[i])
            rng = float(pd.range_m[i])
            t = float(pd.times_s[i])

            all_elevations[s_idx, i] = elev

            if elev < ops.min_elevation_deg:
                all_visible[s_idx, i] = False
                continue

            all_visible[s_idx, i] = True
            cond = env.conditions(
                t, link_inputs.tx_terminal, link_inputs.rx_terminal
            )
            out = self._engine.evaluate_snapshot(
                elev, az, rng, link_inputs, cond
            )
            all_margins[s_idx, i] = out.margin_db
            if out.throughput_mbps is not None:
                all_throughputs[s_idx, i] = out.throughput_mbps

    # Run handover decision at each timestep
    selected_margins = np.zeros(n_steps)
    selected_throughputs = np.zeros(n_steps)
    selected_elevations = np.zeros(n_steps)
    selected_ranges = np.zeros(n_steps)
    outages = np.zeros(n_steps, dtype=bool)
    selected_sat_ids: list[str] = []
    handover_times: list[float] = []
    decisions: list[HandoverDecision] = []

    for i in range(n_steps):
        t = float(times_s[i])
        step_metrics = (
            [float(all_margins[s, i]) for s in range(n_sats)]
            if policy.metric == "margin"
            else [float(all_elevations[s, i]) for s in range(n_sats)]
        )
        step_visible = [bool(all_visible[s, i]) for s in range(n_sats)]

        decision = policy.evaluate(t, sat_ids, step_metrics, step_visible)
        decisions.append(decision)

        s_idx = decision.selected_sat_idx
        selected_sat_ids.append(decision.selected_sat_id)

        if decision.is_handover:
            handover_times.append(t)

        if not all_visible[s_idx, i]:
            outages[i] = True
            selected_margins[i] = float("nan")
            selected_throughputs[i] = 0.0
        else:
            margin = all_margins[s_idx, i]
            selected_margins[i] = margin
            selected_throughputs[i] = all_throughputs[s_idx, i]
            if margin < 0:
                outages[i] = True

        traj = trajectories[sat_ids[s_idx]]
        selected_elevations[i] = float(traj.pass_data.elev_deg[i])
        selected_ranges[i] = float(traj.pass_data.range_m[i])

    # Compute summary
    valid_mask = ~outages
    n_valid = int(np.sum(valid_mask))
    dt_s = float(times_s[1] - times_s[0]) if n_steps > 1 else 1.0
    outage_seconds = float(np.sum(outages)) * dt_s
    availability = n_valid / n_steps if n_steps > 0 else 0.0

    valid_margins_arr = (
        selected_margins[valid_mask] if n_valid > 0 else np.array([0.0])
    )

    summary: dict[str, float] = {
        "availability": availability,
        "outage_seconds": outage_seconds,
        "margin_db_min": float(np.nanmin(valid_margins_arr)) if n_valid > 0 else 0.0,
        "margin_db_mean": float(np.nanmean(valid_margins_arr)) if n_valid > 0 else 0.0,
        "margin_db_p05": (
            float(np.nanpercentile(valid_margins_arr, 5)) if n_valid > 0 else 0.0
        ),
        "margin_db_p50": (
            float(np.nanpercentile(valid_margins_arr, 50)) if n_valid > 0 else 0.0
        ),
        "margin_db_p95": (
            float(np.nanpercentile(valid_margins_arr, 95)) if n_valid > 0 else 0.0
        ),
        "throughput_mbps_mean": (
            float(np.mean(selected_throughputs[valid_mask])) if n_valid > 0 else 0.0
        ),
        "n_handovers": float(len(handover_times)),
    }

    # Per-satellite contact time
    per_sat_contact: dict[str, float] = {sid: 0.0 for sid in sat_ids}
    for i, sid in enumerate(selected_sat_ids):
        if not outages[i]:
            per_sat_contact[sid] += dt_s

    base = WorldSimOutputs(
        times_s=times_s,
        elev_deg=selected_elevations,
        range_m=selected_ranges,
        margin_db=selected_margins,
        throughput_mbps=(
            selected_throughputs if link_inputs.modem is not None else None
        ),
        selected_modcod=None,
        outages_mask=outages,
        summary=summary,
    )

    return MultiSatWorldSimOutputs(
        base=base,
        selected_sat_id=selected_sat_ids,
        handover_times_s=handover_times,
        n_handovers=len(handover_times),
        per_sat_contact_s=per_sat_contact,
    )

MultiSatWorldSimOutputs dataclass 

MultiSatWorldSimOutputs(base, selected_sat_id, handover_times_s, n_handovers, per_sat_contact_s)

Outputs from multi-satellite simulation, extends WorldSimOutputs.

Parameters:

Name Type Description Default
base WorldSimOutputs

Base world simulation outputs (time-series margins, throughput, etc.).

 required
selected_sat_id list[str]

Satellite ID selected at each timestep.

 required
handover_times_s list[float]

Simulation times (seconds) at which handovers occurred.

 required
n_handovers int

Total number of handovers during the simulation.

 required
per_sat_contact_s dict[str, float]

Cumulative contact time (seconds) per satellite ID.

 required

Tier 3: Network World Sim

NetworkWorldSim 

NetworkWorldSim(scheduler='proportional_fair')

Tier 3 world simulation: multi-satellite with traffic scheduling.

At each timestep: 1. Evaluate link per satellite (via MultiSatWorldSim) 2. Compute available capacity from link margin + modem 3. Schedule traffic across users based on demand and capacity 4. Record per-user throughput and satisfaction metrics

Parameters:

Name Type Description Default
scheduler str

Scheduler type: "proportional_fair" (default) or "round_robin".

 'proportional_fair'
Source code in src/opensatcom/world/network_sim.py 
66
67
68
69
70
71
72
def __init__(self, scheduler: str = "proportional_fair") -> None:
    self._scheduler: ProportionalFairScheduler | RoundRobinScheduler
    if scheduler == "round_robin":
        self._scheduler = RoundRobinScheduler()
    else:
        self._scheduler = ProportionalFairScheduler()
    self._engine = DefaultLinkEngine()

run 

run(link_inputs, trajectories, ops, env, traffic_profile)

Run network simulation with traffic scheduling.

Parameters:

Name Type Description Default
link_inputs LinkInputs

Base link inputs (antenna, RF chain, scenario).

 required
trajectories dict[str, PrecomputedTrajectory]

Satellite trajectories keyed by satellite ID.

 required
ops OpsPolicy

Operational policy with min elevation and handover hysteresis.

 required
env StaticEnvironmentProvider

Environment conditions provider.

 required
traffic_profile TrafficProfile

Time-varying traffic demand profile for all users.

 required

Returns:

Type Description
NetworkSimOutputs

Network-level outputs including per-user throughput allocation and satisfaction metrics.
Source code in src/opensatcom/world/network_sim.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def run(
    self,
    link_inputs: LinkInputs,
    trajectories: dict[str, PrecomputedTrajectory],
    ops: OpsPolicy,
    env: StaticEnvironmentProvider,
    traffic_profile: TrafficProfile,
) -> NetworkSimOutputs:
    """Run network simulation with traffic scheduling.

    Parameters
    ----------
    link_inputs : LinkInputs
        Base link inputs (antenna, RF chain, scenario).
    trajectories : dict[str, PrecomputedTrajectory]
        Satellite trajectories keyed by satellite ID.
    ops : OpsPolicy
        Operational policy with min elevation and handover hysteresis.
    env : StaticEnvironmentProvider
        Environment conditions provider.
    traffic_profile : TrafficProfile
        Time-varying traffic demand profile for all users.

    Returns
    -------
    NetworkSimOutputs
        Network-level outputs including per-user throughput allocation
        and satisfaction metrics.
    """
    # First run the multi-sat sim to get link-level results
    multi_sim = MultiSatWorldSim()
    multi_out = multi_sim.run(link_inputs, trajectories, ops, env)

    base_out = multi_out.base
    n_steps = len(base_out.times_s)

    # Determine capacity from margin/throughput at each step
    total_capacity = np.zeros(n_steps)
    bw_hz = link_inputs.scenario.bandwidth_hz

    # Re-evaluate each timestep to get actual Eb/N0 for capacity estimation
    for i in range(n_steps):
        if base_out.outages_mask[i]:
            total_capacity[i] = 0.0
            continue

        # Use throughput if modem is available, else estimate from Eb/N0
        if base_out.throughput_mbps is not None and base_out.throughput_mbps[i] > 0:
            total_capacity[i] = base_out.throughput_mbps[i]
        else:
            # Compute actual Eb/N0 from margin + required value
            actual_ebn0_db = base_out.margin_db[i] + link_inputs.scenario.required_value
            if np.isnan(actual_ebn0_db):
                total_capacity[i] = 0.0
            else:
                ebn0_lin = 10.0 ** (actual_ebn0_db / 10.0)
                capacity_bps = bw_hz * np.log2(1.0 + ebn0_lin)
                total_capacity[i] = capacity_bps / 1e6

    # Collect all user IDs from traffic profile
    sample_demands = traffic_profile.demands_at(float(base_out.times_s[0]))
    user_ids = [d.user_id for d in sample_demands]

    per_user_throughput: dict[str, np.ndarray] = {
        uid: np.zeros(n_steps) for uid in user_ids
    }

    # Schedule traffic at each timestep
    for i in range(n_steps):
        t_s = float(base_out.times_s[i])
        demands = traffic_profile.demands_at(t_s)
        cap = total_capacity[i]

        allocation = self._scheduler.allocate(demands, cap)
        for uid, allocated in allocation.items():
            if uid in per_user_throughput:
                per_user_throughput[uid][i] = allocated

    # Compute satisfaction: fraction of demand met over simulation
    user_satisfaction: dict[str, float] = {}
    for uid in user_ids:
        total_demand = 0.0
        total_served = 0.0
        for i in range(n_steps):
            t_s = float(base_out.times_s[i])
            demands = traffic_profile.demands_at(t_s)
            for d in demands:
                if d.user_id == uid:
                    total_demand += d.demand_mbps
                    total_served += per_user_throughput[uid][i]
                    break
        user_satisfaction[uid] = total_served / total_demand if total_demand > 0 else 0.0

    return NetworkSimOutputs(
        base=multi_out,
        per_user_throughput_mbps=per_user_throughput,
        total_capacity_mbps=total_capacity,
        user_satisfaction=user_satisfaction,
    )

NetworkSimOutputs dataclass 

NetworkSimOutputs(base, per_user_throughput_mbps, total_capacity_mbps, user_satisfaction)

Outputs from network-level simulation, extends MultiSatWorldSimOutputs.

Parameters:

Name Type Description Default
base MultiSatWorldSimOutputs

Underlying multi-satellite simulation outputs.

 required
per_user_throughput_mbps dict[str, ndarray]

Allocated throughput (Mbps) time-series per user ID.

 required
total_capacity_mbps ndarray

Total available link capacity (Mbps) at each timestep.

 required
user_satisfaction dict[str, float]

Fraction of total demand served over the simulation, per user ID (0.0 = none served, 1.0 = fully served).

 required

Providers

StaticEnvironmentProvider 

StaticEnvironmentProvider(conditions)

Returns fixed propagation conditions for all timesteps.

Parameters:

Name Type Description Default
conditions PropagationConditions

Static conditions returned at every time step.

 required
Source code in src/opensatcom/world/providers.py 
21
22
def __init__(self, conditions: PropagationConditions) -> None:
    self._conditions = conditions

conditions 

conditions(t_s, terminal_a, terminal_b)

Return the static propagation conditions.

Parameters:

Name Type Description Default
t_s float

Simulation time in seconds (unused).

 required
terminal_a Terminal

First terminal (unused).

 required
terminal_b Terminal

Second terminal (unused).

 required

Returns:

Type Description
PropagationConditions

The fixed conditions provided at construction.
Source code in src/opensatcom/world/providers.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def conditions(
    self, t_s: float, terminal_a: Terminal, terminal_b: Terminal
) -> PropagationConditions:
    """Return the static propagation conditions.

    Parameters
    ----------
    t_s : float
        Simulation time in seconds (unused).
    terminal_a : Terminal
        First terminal (unused).
    terminal_b : Terminal
        Second terminal (unused).

    Returns
    -------
    PropagationConditions
        The fixed conditions provided at construction.
    """
    return self._conditions

PrecomputedTrajectory 

PrecomputedTrajectory(pass_data)

Trajectory provider from pre-computed pass data.

Stores az/el/range arrays directly — no ECEF conversion needed. The states_ecef method returns dummy states (the WorldSim uses the az/el/range directly via get_geometry).

Source code in src/opensatcom/world/providers.py 
87
88
def __init__(self, pass_data: PrecomputedPassData) -> None:
    self.pass_data = pass_data

from_arrays classmethod 

from_arrays(times_s, elev_deg, az_deg, range_m)

Construct from raw numpy arrays.

Parameters:

Name Type Description Default
times_s ndarray

Time stamps in seconds.

 required
elev_deg ndarray

Elevation angles in degrees.

 required
az_deg ndarray

Azimuth angles in degrees.

 required
range_m ndarray

Slant ranges in metres.

 required

Returns:

Type Description
PrecomputedTrajectory

Trajectory wrapping the provided arrays.
Source code in src/opensatcom/world/providers.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@classmethod
def from_arrays(
    cls,
    times_s: np.ndarray,
    elev_deg: np.ndarray,
    az_deg: np.ndarray,
    range_m: np.ndarray,
) -> PrecomputedTrajectory:
    """Construct from raw numpy arrays.

    Parameters
    ----------
    times_s : numpy.ndarray
        Time stamps in seconds.
    elev_deg : numpy.ndarray
        Elevation angles in degrees.
    az_deg : numpy.ndarray
        Azimuth angles in degrees.
    range_m : numpy.ndarray
        Slant ranges in metres.

    Returns
    -------
    PrecomputedTrajectory
        Trajectory wrapping the provided arrays.
    """
    return cls(PrecomputedPassData(times_s, elev_deg, az_deg, range_m))

states_ecef 

states_ecef(t0_s, t1_s, dt_s)

Return dummy ECEF states (not used by SimpleWorldSim).

Parameters:

Name Type Description Default
t0_s float

Start time in seconds.

 required
t1_s float

End time in seconds.

 required
dt_s float

Time step in seconds.

 required

Returns:

Type Description
list of StateECEF

Placeholder states with zero position vectors.
Source code in src/opensatcom/world/providers.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def states_ecef(
    self, t0_s: float, t1_s: float, dt_s: float
) -> list[StateECEF]:
    """Return dummy ECEF states (not used by SimpleWorldSim).

    Parameters
    ----------
    t0_s : float
        Start time in seconds.
    t1_s : float
        End time in seconds.
    dt_s : float
        Time step in seconds.

    Returns
    -------
    list of StateECEF
        Placeholder states with zero position vectors.
    """
    return [
        StateECEF(t_s=t, r_m=np.zeros(3))
        for t in self.pass_data.times_s
    ]

get_geometry 

get_geometry(idx)

Get geometry for a given timestep index.

Parameters:

Name Type Description Default
idx int

Time step index into the pass data arrays.

 required

Returns:

Type Description
tuple of (float, float, float)

(elev_deg, az_deg, range_m) at the given index.
Source code in src/opensatcom/world/providers.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def get_geometry(self, idx: int) -> tuple[float, float, float]:
    """Get geometry for a given timestep index.

    Parameters
    ----------
    idx : int
        Time step index into the pass data arrays.

    Returns
    -------
    tuple of (float, float, float)
        ``(elev_deg, az_deg, range_m)`` at the given index.
    """
    return (
        float(self.pass_data.elev_deg[idx]),
        float(self.pass_data.az_deg[idx]),
        float(self.pass_data.range_m[idx]),
    )

PrecomputedPassData dataclass 

PrecomputedPassData(times_s, elev_deg, az_deg, range_m)

Pre-baked pass data: arrays of time, elevation, azimuth, range.

Parameters:

Name Type Description Default
times_s ndarray

Time stamps in seconds, shape (N,).

 required
elev_deg ndarray

Elevation angles in degrees, shape (N,).

 required
az_deg ndarray

Azimuth angles in degrees, shape (N,).

 required
range_m ndarray

Slant ranges in metres, shape (N,).

 required

Handover

HandoverPolicy 

HandoverPolicy(hysteresis_db=3.0, hysteresis_s=5.0, metric='margin')

Hysteresis-based handover policy.

Selects the satellite with the best metric (margin or elevation). A handover from the current satellite only occurs if a candidate exceeds the current satellite's metric by hysteresis_db and the hysteresis timer (hysteresis_s) has elapsed since the last handover.

Parameters:

Name Type Description Default
hysteresis_db float

Minimum margin advantage (dB) for a candidate to trigger handover.

 3.0
hysteresis_s float

Minimum time (seconds) between handovers.

 5.0
metric str

Selection metric: "margin" or "elevation".

 'margin'
Source code in src/opensatcom/world/handover.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    hysteresis_db: float = 3.0,
    hysteresis_s: float = 5.0,
    metric: str = "margin",
) -> None:
    if metric not in ("margin", "elevation"):
        raise ValueError(f"metric must be 'margin' or 'elevation', got '{metric}'")
    self.hysteresis_db = hysteresis_db
    self.hysteresis_s = hysteresis_s
    self.metric = metric
    self._current_sat_idx: int = 0
    self._last_handover_t: float = -np.inf

reset 

reset(initial_sat_idx=0)

Reset handover state for a new simulation run.

Parameters:

Name Type Description Default
initial_sat_idx int

Index of the initially selected satellite (default 0).

 0

Returns:

Type Description
None
Source code in src/opensatcom/world/handover.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def reset(self, initial_sat_idx: int = 0) -> None:
    """Reset handover state for a new simulation run.

    Parameters
    ----------
    initial_sat_idx : int, optional
        Index of the initially selected satellite (default 0).

    Returns
    -------
    None
    """
    self._current_sat_idx = initial_sat_idx
    self._last_handover_t = -np.inf

evaluate 

evaluate(t_s, sat_ids, metrics, visible)

Evaluate handover decision at one timestep.

Parameters:

Name Type Description Default
t_s float

Current simulation time in seconds.

 required
sat_ids list[str]

Satellite identifiers.

 required
metrics list[float]

Per-satellite metric values (margin_db or elevation_deg).

 required
visible list[bool]

Per-satellite visibility flags (above min elevation).

 required

Returns:

Type Description
HandoverDecision
Source code in src/opensatcom/world/handover.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def evaluate(
    self,
    t_s: float,
    sat_ids: list[str],
    metrics: list[float],
    visible: list[bool],
) -> HandoverDecision:
    """Evaluate handover decision at one timestep.

    Parameters
    ----------
    t_s : float
        Current simulation time in seconds.
    sat_ids : list[str]
        Satellite identifiers.
    metrics : list[float]
        Per-satellite metric values (margin_db or elevation_deg).
    visible : list[bool]
        Per-satellite visibility flags (above min elevation).

    Returns
    -------
    HandoverDecision
    """
    n_sats = len(sat_ids)
    if n_sats == 0:
        raise ValueError("No satellites to evaluate")

    # Filter to visible satellites
    visible_indices = [i for i in range(n_sats) if visible[i]]

    if not visible_indices:
        # No visible satellites — keep current (will be outage)
        return HandoverDecision(
            selected_sat_idx=self._current_sat_idx,
            selected_sat_id=sat_ids[self._current_sat_idx],
            is_handover=False,
            margin_db=float("nan"),
            all_margins_db=list(metrics),
        )

    # Find best visible candidate
    best_idx = max(visible_indices, key=lambda i: metrics[i])
    best_metric = metrics[best_idx]

    # Current satellite metric (if visible)
    current_visible = self._current_sat_idx in visible_indices
    current_metric = metrics[self._current_sat_idx] if current_visible else -np.inf

    # Handover logic
    time_since_last = t_s - self._last_handover_t
    timer_ok = time_since_last >= self.hysteresis_s

    if not current_visible:
        # Must handover — current satellite not visible
        self._current_sat_idx = best_idx
        self._last_handover_t = t_s
        is_handover = True
    elif best_idx != self._current_sat_idx and timer_ok:
        advantage = best_metric - current_metric
        if advantage > self.hysteresis_db:
            self._current_sat_idx = best_idx
            self._last_handover_t = t_s
            is_handover = True
        else:
            is_handover = False
    else:
        is_handover = False

    return HandoverDecision(
        selected_sat_idx=self._current_sat_idx,
        selected_sat_id=sat_ids[self._current_sat_idx],
        is_handover=is_handover,
        margin_db=metrics[self._current_sat_idx],
        all_margins_db=list(metrics),
    )

HandoverDecision dataclass 

HandoverDecision(selected_sat_idx, selected_sat_id, is_handover, margin_db, all_margins_db)

Result of a handover evaluation at one timestep.

Parameters:

Name Type Description Default
selected_sat_idx int

Index of the selected satellite in the candidate list.

 required
selected_sat_id str

Identifier of the selected satellite.

 required
is_handover bool

True if a handover occurred at this timestep.

 required
margin_db float

Link margin (dB) of the selected satellite.

 required
all_margins_db list[float]

Metric values for all candidate satellites at this timestep.

 required

Schedulers

ProportionalFairScheduler 

ProportionalFairScheduler(alpha=1.0)

Allocates capacity proportional to demand and channel quality.

Proportional fair: each user gets capacity proportional to their demand weighted by priority, subject to total capacity constraint.

Parameters:

Name Type Description Default
alpha float

Fairness parameter. alpha=1.0 is pure proportional fair.

 1.0
Source code in src/opensatcom/world/scheduler.py 
20
21
def __init__(self, alpha: float = 1.0) -> None:
    self.alpha = alpha

allocate 

allocate(users, capacity_mbps)

Allocate capacity across users.

Returns dict mapping user_id to allocated Mbps. Total allocation never exceeds capacity_mbps.

Source code in src/opensatcom/world/scheduler.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def allocate(
    self, users: list[TrafficDemand], capacity_mbps: float
) -> dict[str, float]:
    """Allocate capacity across users.

    Returns dict mapping user_id to allocated Mbps.
    Total allocation never exceeds capacity_mbps.
    """
    if not users or capacity_mbps <= 0:
        return {u.user_id: 0.0 for u in users}

    # Weighted demand: demand * (priority + 1) ^ alpha
    weights = {}
    for u in users:
        w = u.demand_mbps * ((u.priority + 1) ** self.alpha)
        weights[u.user_id] = max(w, 0.0)

    total_weight = sum(weights.values())
    if total_weight <= 0:
        return {u.user_id: 0.0 for u in users}

    # Proportional allocation
    allocation: dict[str, float] = {}
    total_demand = sum(u.demand_mbps for u in users)

    if total_demand <= capacity_mbps:
        # Enough capacity for everyone
        for u in users:
            allocation[u.user_id] = u.demand_mbps
    else:
        # Proportional fair split
        for u in users:
            share = weights[u.user_id] / total_weight
            allocated = share * capacity_mbps
            # Don't allocate more than demanded
            allocation[u.user_id] = min(allocated, u.demand_mbps)

    return allocation

RoundRobinScheduler

Simple round-robin scheduler: equal share to each user.

Each user gets capacity / n_users, capped at their demand. Remaining capacity from users whose demand is below their equal share is not redistributed.

allocate 

allocate(users, capacity_mbps)

Allocate capacity equally across users.

Parameters:

Name Type Description Default
users list[TrafficDemand]

List of user traffic demands to schedule.

 required
capacity_mbps float

Total available capacity in Mbps.

 required

Returns:

Type Description
dict[str, float]

Mapping of user_id to allocated throughput in Mbps.
Source code in src/opensatcom/world/scheduler.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def allocate(
    self, users: list[TrafficDemand], capacity_mbps: float
) -> dict[str, float]:
    """Allocate capacity equally across users.

    Parameters
    ----------
    users : list[TrafficDemand]
        List of user traffic demands to schedule.
    capacity_mbps : float
        Total available capacity in Mbps.

    Returns
    -------
    dict[str, float]
        Mapping of user_id to allocated throughput in Mbps.
    """
    if not users or capacity_mbps <= 0:
        return {u.user_id: 0.0 for u in users}

    equal_share = capacity_mbps / len(users)
    allocation: dict[str, float] = {}

    # First pass: allocate min(equal_share, demand)
    remaining = capacity_mbps
    unsatisfied = []
    for u in users:
        alloc = min(equal_share, u.demand_mbps)
        allocation[u.user_id] = alloc
        remaining -= alloc
        if alloc < equal_share and remaining > 0:
            pass  # User doesn't need full share
        elif alloc >= u.demand_mbps:
            remaining += 0  # Already handled
        else:
            unsatisfied.append(u.user_id)

    return allocation

Traffic

TrafficDemand dataclass 

TrafficDemand(user_id, demand_mbps, priority=0, lat_deg=0.0, lon_deg=0.0, alt_m=0.0)

Single user traffic demand at a point in time.

Parameters:

Name Type Description Default
user_id str

Unique identifier for the user.

 required
demand_mbps float

Requested throughput in Mbps.

 required
priority int

Scheduling priority (higher values = higher priority, default 0).

 0
lat_deg float

User latitude in degrees (default 0.0).

 0.0
lon_deg float

User longitude in degrees (default 0.0).

 0.0
alt_m float

User altitude in meters (default 0.0).

 0.0

TrafficProfile

Base class for time-varying traffic demand profiles.

demands_at 

demands_at(t_s)

Return traffic demands at a given simulation time.

Parameters:

Name Type Description Default
t_s float

Simulation time in seconds.

 required

Returns:

Type Description
list[TrafficDemand]

List of per-user traffic demands at time t_s.
Source code in src/opensatcom/world/traffic.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def demands_at(self, t_s: float) -> list[TrafficDemand]:
    """Return traffic demands at a given simulation time.

    Parameters
    ----------
    t_s : float
        Simulation time in seconds.

    Returns
    -------
    list[TrafficDemand]
        List of per-user traffic demands at time ``t_s``.
    """
    raise NotImplementedError

ConstantTrafficProfile 

ConstantTrafficProfile(demands)

Bases: TrafficProfile

Traffic profile with constant demands over time.

Parameters:

Name Type Description Default
demands list[TrafficDemand]

Fixed list of user demands.

 required
Source code in src/opensatcom/world/traffic.py 
64
65
def __init__(self, demands: list[TrafficDemand]) -> None:
    self._demands = demands

demands_at 

demands_at(t_s)

Return constant traffic demands regardless of time.

Parameters:

Name Type Description Default
t_s float

Simulation time in seconds (unused).

 required

Returns:

Type Description
list[TrafficDemand]

Copy of the fixed demand list.
Source code in src/opensatcom/world/traffic.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def demands_at(self, t_s: float) -> list[TrafficDemand]:
    """Return constant traffic demands regardless of time.

    Parameters
    ----------
    t_s : float
        Simulation time in seconds (unused).

    Returns
    -------
    list[TrafficDemand]
        Copy of the fixed demand list.
    """
    return list(self._demands)

TimeVaryingTrafficProfile 

TimeVaryingTrafficProfile(base_demands, pattern='ramp', ramp_factor=2.0, burst_period_s=60.0, burst_multiplier=3.0, burst_duration_s=10.0, t_start_s=0.0, t_end_s=600.0)

Bases: TrafficProfile

Traffic profile with time-varying patterns.

Supports ramp and burst patterns per user.

Parameters:

Name Type Description Default
base_demands list[TrafficDemand]

Base demand levels.

 required
pattern str

Pattern type: "ramp" (linear increase), "burst" (periodic spikes).

 'ramp'
ramp_factor float

For "ramp": max multiplier at end of simulation.

 2.0
burst_period_s float

For "burst": period in seconds between bursts.

 60.0
burst_multiplier float

For "burst": demand multiplier during burst.

 3.0
burst_duration_s float

For "burst": duration of each burst.

 10.0
t_start_s float

Simulation start time for ramp scaling.

 0.0
t_end_s float

Simulation end time for ramp scaling.

 600.0
Source code in src/opensatcom/world/traffic.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def __init__(
    self,
    base_demands: list[TrafficDemand],
    pattern: str = "ramp",
    ramp_factor: float = 2.0,
    burst_period_s: float = 60.0,
    burst_multiplier: float = 3.0,
    burst_duration_s: float = 10.0,
    t_start_s: float = 0.0,
    t_end_s: float = 600.0,
) -> None:
    self._base = base_demands
    self._pattern = pattern
    self._ramp_factor = ramp_factor
    self._burst_period_s = burst_period_s
    self._burst_multiplier = burst_multiplier
    self._burst_duration_s = burst_duration_s
    self._t_start = t_start_s
    self._t_end = t_end_s

demands_at 

demands_at(t_s)

Return scaled traffic demands based on the configured pattern.

For "ramp" pattern, demand scales linearly from 1.0 at t_start_s to ramp_factor at t_end_s. For "burst" pattern, demand is multiplied by burst_multiplier during periodic burst windows.

Parameters:

Name Type Description Default
t_s float

Simulation time in seconds.

 required

Returns:

Type Description
list[TrafficDemand]

Demand list with demand_mbps scaled by the time-dependent pattern multiplier.
Source code in src/opensatcom/world/traffic.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def demands_at(self, t_s: float) -> list[TrafficDemand]:
    """Return scaled traffic demands based on the configured pattern.

    For ``"ramp"`` pattern, demand scales linearly from 1.0 at
    ``t_start_s`` to ``ramp_factor`` at ``t_end_s``.  For ``"burst"``
    pattern, demand is multiplied by ``burst_multiplier`` during
    periodic burst windows.

    Parameters
    ----------
    t_s : float
        Simulation time in seconds.

    Returns
    -------
    list[TrafficDemand]
        Demand list with ``demand_mbps`` scaled by the time-dependent
        pattern multiplier.
    """
    if self._pattern == "ramp":
        duration = self._t_end - self._t_start
        if duration <= 0:
            scale = 1.0
        else:
            frac = min(max((t_s - self._t_start) / duration, 0.0), 1.0)
            scale = 1.0 + (self._ramp_factor - 1.0) * frac
    elif self._pattern == "burst":
        phase = (t_s % self._burst_period_s)
        scale = self._burst_multiplier if phase < self._burst_duration_s else 1.0
    else:
        scale = 1.0

    return [
        TrafficDemand(
            user_id=d.user_id,
            demand_mbps=d.demand_mbps * scale,
            priority=d.priority,
            lat_deg=d.lat_deg,
            lon_deg=d.lon_deg,
            alt_m=d.alt_m,
        )
        for d in self._base
    ]