utils package
Submodules
utils.graph_generator module
- utils.graph_generator.fat_tree(k: int = 4, qpu_edge_dist=1.0, edge_aggregate_dist=3.0, aggregate_core_dist=6.0, L_dep: float = 50.0) tuple[list, list, dict, dict, list, float][source]
Generates a fat-tree topology with k pods. Each pod contains k/2 edge switches and k/2 aggregate switches. The core layer has (k/2)^2 core switches. Each edge switch connects to k/2 hosts (QPUs). See: https://arxiv.org/pdf/2601.01353
- Parameters:
k (int, optional) – Number of pods.
qpu_edge_dist (float, optional) – Distance between QPUs and edge
switches. (core)
edge_aggregate_dist (float, optional) – Distance between edge and
switches.
aggregate_core_dist (float, optional) – Distance between aggregate and
switches.
L_dep (float, optional) – Depolarization length in km for edge
fidelities.
- Returns:
- A tuple containing:
List of nodes in the graph.
List of edges in the graph.
Dictionary mapping edges to their fidelities.
Dictionary mapping edges to their p_gen rates.
List of QPUs (hosts) in the graph.
Diameter of the graph.
- Return type:
tuple[list, list, dict, dict, list, float]
- utils.graph_generator.generate_waxman_graph(n: int = 48, alpha: float = 0.2, beta: float = 0.3, rng: Generator | None = None, max_retries: int = 5000, max_avg_degree: float = 3.0, max_hops: int = 8) tuple[list, list, dict, dict, float, float][source]
Generates a Waxman graph with constraints on connectivity, average degree, and diameter.
- Parameters:
n (int, optional) – Number of nodes in the graph.
(float (alpha) – ): Controls the influence of distance on edge
optional – ): Controls the influence of distance on edge
distant (probability. Higher alpha values lead to fewer edges between)
nodes.
beta (float, optional) – Controls the overall density of the graph.
edges. (Higher beta values lead to more)
rng (np.random.Generator | None, optional) – Random number generator.
max_retries (int, optional) – Maximum number of retries to generate a
(connected (valid graph)
max_avg_degree (with average degree <=)
:param : :param and diameter <= max_hops).: :param max_avg_degree: Maximum average degree of the graph. :type max_avg_degree: float, optional :param max_hops: Maximum diameter of the graph. Defaults to 8. :type max_hops: int, optional
- Returns:
- A tuple containing:
List of nodes in the graph.
List of edges in the graph.
Dictionary mapping edges to their fidelities.
Dictionary mapping edges to their p_gen rates.
Average degree of the graph.
Diameter of the graph.
- Return type:
tuple[list, list, dict, dict, float, float]
utils.helper module
- utils.helper.all_simple_paths(paths: Dict[Tuple[str, str], List[Tuple[float, Tuple[str, ...]]]], src: str, dst: str) List[Tuple[float, Tuple[str, ...]]][source]
- utils.helper.app_params_sim(paths: dict[str, list[str]], app_specs: dict[str, dict[str, Any]], p_packet: float, memory: int, p_swap: float, time_slot_duration: float) dict[str, dict[str, float | int]][source]
Prepare application parameters for simulation.
- Parameters:
paths (dict[str, list[str]]) – Paths for each application in the
network.
app_specs (dict[str, dict[str, Any]]) – Application metadata produced by
requirements. (generate_n_apps containing network endpoints and)
p_packet (float) – Probability of a packet being generated.
memory (int) – Number of independent link-generation trials per slot.
p_swap (float) – Probability of swapping an EPR pair in a single trial.
time_slot_duration (float) – Duration of a time slot in seconds.
- Returns:
Mapping of application name to the parameters required by the simulator when instantiating PGAs.
- Return type:
dict[str, dict[str, float | int]]
- utils.helper.compute_edge_fidelities(G: Graph, distances: Dict[Tuple, float], L_dep: float = 50.0) Dict[Tuple, float][source]
- utils.helper.compute_edge_rates(G: Graph, distances: Dict[Tuple, float], attenuation: float = 0.2) Dict[Tuple, float][source]
- utils.helper.compute_link_utilization(link_busy: Dict[Tuple[str, str], float], min_start: float, max_completion: float) Dict[Tuple[str, str], Dict[str, float]][source]
- utils.helper.fidelity_bounds(bounds: Dict[Tuple[str, str], Tuple[float, float]], src: str, dst: str) Tuple[float, float][source]
- utils.helper.generate_n_apps(end_nodes: list[str], bounds: dict[tuple, tuple], n_apps: int, inst_range: int, epr_range: tuple[int, int], deadline_range: tuple[float, float], list_policies: list[str], rng: Generator) Dict[str, Dict[str, Any]][source]
Generates a specified number of applications with random parameters.
- Parameters:
end_nodes (list) – List of end nodes in the network.
n_apps (int) – Number of applications to generate.
inst_range (int) – Poisson lambda for the number of instances
application. (each)
epr_range (tuple[int, int]) – Range (min, max) for the number of EPR
application.
deadline_range (tuple[float, float]) – Range (min, max) for the relative
application (deadline budget of each)
fidelity_range (tuple[float, float]) – Range (min, max) for the minimum
application.
list_policies (list[str], optional) – List of policies to assign to
application.
rng (np.random.Generator) – Random number generator for reproducibility.
- Returns:
Mapping of application name to its metadata, including endpoints, number of instances, requested EPR pairs, deadline budget, and policy.
- Return type:
Dict[str, Dict[str, Any]]
- utils.helper.gml_data(gml_file: str) Tuple[list, list, dict[tuple, float], dict, float][source]
Extracts nodes, edges, distances, fidelities, and diameter from a GML file.
- Parameters:
gml_file (str) – Path to the GML file.
- Returns:
List of nodes. edges (list): List of edges (source, target). distances (dict[tuple, float]): Dict mapping directed edges to distances. fidelities (dict[tuple, float]): Dict mapping directed edges to fidelities. rates (dict[tuple, float]): Dict mapping directed edges to rates. diameter (float): Diameter of the graph.
- Return type:
nodes (list)
- utils.helper.parallelizable_tasks(paths_for_each_apps: dict[str, List[str] | None]) dict[str, set[str]][source]
Find parallelizable applications based on shared links of a quantum network.
- Parameters:
paths_for_each_apps (dict[str, List[str]]) –
A dictionary where keys are application names and values are list of nodes describing the route used to run the applications, e.g. on a linear chain Alice-Rob-Bob:
- {
‘A’: [‘Alice’, ‘Rob’], ‘B’: [‘Rob’, ‘Bob’],
}
- Returns:
- A dictionary where keys are application names and
values are sets of applications that can run in parallel with key application, based on shared links. From the example above, the output would be A and B can run in parallel since they do not share any links:
- {
‘A’: {‘B’}, ‘B’: {‘A’},
}
- Return type:
dict[str, set[str]]
- utils.helper.prepare_run_dir(output_dir: str, ppacket_values: Iterable[float], keep_seed_outputs: bool = True) tuple[str, str][source]
- utils.helper.save_results(df: DataFrame, pga_names: List[str], pga_release_times: Dict[str, float], app_specs: Dict[str, Dict[str, Any]], n_edges: int, durations: Dict[str, float] | None = None, pga_network_paths: Dict[str, List[str]] | None = None, link_utilization: Dict[Tuple[str, str], Dict[str, float]] | None = None, link_waiting: Dict[Tuple[str, str], Dict[str, float | int]] | None = None, admitted_apps: int | None = None, total_apps: int | None = None, app_e2e_fidelities: Dict[str, float] | None = None, single_path_share: float = nan, two_path_share: float = nan, avg_deg: float = nan, output_dir: str = 'results', save_csv: bool = True, verbose: bool = True, routing_decision_cpt: int | None = None, routing_decision_runtime: float | None = None, warmup: float | None = None, end_time: float | None = None, defer_counts: Dict[str, int] | None = None) Dict[str, float][source]
Save the results of PGA scheduling and execution to a CSV file and print a summary of the results.
- Parameters:
df (DataFrame) – DataFrame containing PGA results with columns: - pga: PGA identifier - arrival_time: Time when the PGA arrived - start_time: Time when the PGA started execution - burst_time: Total time required for the PGA to complete - completion_time: Time when the PGA completed execution - turnaround_time: Total time from arrival to completion - waiting_time: Total time the PGA waited before execution - status: Status of the PGA (e.g., “completed”, “failed”) - deadline: Deadline for the PGA (if applicable) - src_node: Source node of the PGA - dst_node: Destination node of the PGA - instances: Number of instances for the PGA - epr_pairs: Number of EPR pairs for the PGA - policy: Scheduling policy used for the PGA
pga_names (List) – List of all PGA names that should be present in the results.
pga_release_times (Dict) – Dictionary mapping PGA names to their relative release times, used to fill in missing PGAs.
app_specs (Dict) – Metadata for each application including endpoints, instances, requested EPR pairs, deadline budget, and policy.
n_edges (int) – Number of edges in the network graph.
durations (Dict | None) – Optional mapping of deterministic PGA durations per application.
pga_network_paths (Dict | None) – Length of network paths per application.
link_utilization (Dict) – Dictionary mapping links to busy time and utilization metrics.
link_waiting (Dict | None) – Dictionary mapping links to waiting metrics (total waiting time and number of PGAs that waited).
output_dir (str) – Directory where the results CSV file will be saved.
save_csv (bool) – Whether to save results to CSV files.
verbose (bool) – Whether to print summary statistics to stdout.
- Returns:
- Dictionary containing summary metrics including
admission_rate, makespan, throughput, completion ratios, and utilization statistics.
- Return type:
Dict[str, float]
- utils.helper.track_link_waiting(waiting_time: float, wait_acc: Dict[Tuple[str, str], Dict[str, float]], blocking_links: List[Tuple[str, str]] | None = None) None[source]
Track waiting time statistics per link.
- Parameters:
waiting_time (float) – Waiting time per PGA.
wait_acc (Dict[Tuple[str, str], Dict[str, float]]) – Accumulator
link. (for waiting time statistics per)
blocking_links (List[Tuple[str, str]] | None) – The specific link(s)
waiting (that caused the)
links. (time is distributed equally among these)
utils.parallel_simulations module
- utils.parallel_simulations.run_parallel_sims(tasks: list[tuple[Any, ...]], max_workers: int, show_progress: bool, default_kwargs: dict[str, Any], run_dir: str) list[dict[str, Any]][source]
- utils.parallel_simulations.run_ppacket_parallel_simulations(ppacket_values: Sequence[float], arrival_rate_values: Sequence[float], simulations_per_point: int, seed_start: int, run_dir: str, default_kwargs: dict, keep_seed_outputs: bool, inst_range_values: Sequence[int] | None = None, max_workers: int | None = None, show_progress: bool = True, raw_csv_path: str | None = None) DataFrame[source]
- utils.parallel_simulations.run_ppacket_sweep_to_csv(ppacket_values: Sequence[float], arrival_rate_values: Sequence[float], simulations_per_point: int, seed_start: int = 0, config: str = 'configurations/network/Dumbbell.gml', output_dir: str = 'results', simulation_kwargs: dict | None = None, keep_seed_outputs: bool = False, inst_range_values: Sequence[int] | None = None, max_workers: int | None = None, show_progress: bool = True) tuple[DataFrame, str][source]
utils.plots module
- utils.plots.build_metric_specs(v_label: str, save_path: str, run_dir: str, sch_suffix: str, x: str = 'p_packet', group_column: str | None = None, group_labels: dict | None = None, group_palette: Sequence[str] | None = None, group_palette_map: dict[str, Any] | None = None, individual_values: Sequence[int | float] | None = None, group_line_styles: dict[str, str] | None = None, create_individual: bool = False) list[dict[str, Any]][source]
- utils.plots.plot_metrics_vs_load(path: str | Sequence[str], save_path: str | None = None, figsize: tuple[float, float] = (7, 4.5), dpi: int = 600, group_column: str | None = None, gp_labels: dict | None = None, group_palette: Sequence[str] | None = None, p_packet_values: Sequence[float] | None = None, scheduler: str | None = None, create_individual: bool = False, multi: bool = False, x_var: str = 'arrival_rate') DataFrame[source]
Example usage: df = plot_metrics_vs_load(
- path=[
“1.csv”, “2.csv”, “3.csv”, “4.csv”,
], multi=True, gp_labels={
“1”: “Precomputed”, “2”: “Proactive”, “3”: “Hybrid”, “4”: “Reactive”,
},
)
- utils.plots.plot_metrics_vs_ppacket(raw_csv_path: str, save_path: str | None = None, figsize: tuple[float, float] = (7, 4.5), dpi: int = 600, gp_column: str | None = None, gp_labels: dict | None = None, group_palette: Sequence[str] | None = None, n_apps_values: Sequence[int] | None = None, scheduler: str | None = None, create_individual: bool = False) DataFrame[source]