utils package

Submodules

utils.graph_generator module

utils.graph_generator.clos(n_spine: int = 4, n_leaf: int = 4, hosts_per_leaf: int = 4, qpu_leaf_dist: float = 0.1, leaf_spine_dist: float = 0.3, coherence: float = 0.02) → tuple[list, list, dict, dict, list, float][source]

Generates a leaf-spine (two-tier Clos).

Parameters:

  • n_spine (int, optional) – Number of spine switches.

  • n_leaf (int, optional) – Number of leaf switches.

  • hosts_per_leaf (int, optional) – Number of QPUs (hosts) attached to

  • switch. (each leaf)

  • qpu_leaf_dist (float, optional) – Distance between QPUs and leaf

  • switches.

  • leaf_spine_dist (float, optional) – Distance between leaf and spine

  • switches.

  • coherence (float, optional) – Memory coherence time in seconds used to

  • fidelities. (compute edge)

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.dragonfly(a: int = 4, h: int = 2, p: int = 2, qpu_router_dist: float = 0.1, intra_group_dist: float = 0.3, global_dist: float = 0.6, coherence: float = 0.02) → tuple[list, list, dict, dict, list, float][source]

Generates a dragonfly topology with g = a * h + 1 groups. Each group contains a routers connected in a complete graph, each router hosts p QPUs and has h global links, so every pair of groups is connected by exactly one global link. See: https://doi.org/10.1145/1394608.1382129

Parameters:

  • a (int, optional) – Number of routers per group.

  • h (int, optional) – Number of global links per router.

  • p (int, optional) – Number of QPUs (hosts) per router.

  • qpu_router_dist (float, optional) – Distance between QPUs and routers.

  • intra_group_dist (float, optional) – Distance between routers within

  • group. (a)

  • global_dist (float, optional) – Distance between routers of different

  • groups.

  • coherence (float, optional) – Memory coherence time in seconds used to

  • fidelities. (compute edge)

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.fat_tree(k: int = 4, qpu_edge_dist=0.1, edge_aggregate_dist=0.3, aggregate_core_dist=0.6, coherence: float = 0.02) → 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.

  • coherence (float, optional) – Memory coherence time in seconds used to

  • fidelities. (compute edge)

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: numpy.random.Generator = None, max_retries: int = 5000, max_avg_degree: float = 3.0, max_hops: int = 8, coherence: float = 0.02) → 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 :param coherence: Memory coherence time in seconds used to :type coherence: float, optional :param compute 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.

  • 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, coherence: float = 0.02) → 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.

  • coherence (float) – Coherence time in seconds of a generated pair.

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.build_default_sim_args(config: str, args: dict | None) → dict[source]
utils.helper.compute_edge_fidelities(G: networkx.Graph, distances: Dict[Tuple, float], T_coh: float = 0.02, c_fiber: float = 200000.0, F0: float = 0.95) → Dict[Tuple, float][source]
utils.helper.compute_edge_probs(G: networkx.Graph, distances: Dict[Tuple, float], attenuation: float = 0.2, coupling_efficiency: float = 0.2) → Dict[Tuple, float][source]
utils.helper.count_edge_disjoint_paths(feasible_paths: List[Tuple[float, Tuple[str, ...]]]) → int[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], rng: numpy.random.Generator, manual_pairs: list[tuple[str, str]] | None = None) → 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. (pairs for 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 deadline

  • k (budget multiplier)

  • duration (derived later as deadline_budget = k * fastest-path PGA)

  • so

  • path. (every app is feasible on its best)

  • 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, and deadline budget multiplier.

Return type:

Dict[str, Dict[str, Any]]

utils.helper.gml_data(gml_file: str, coherence: float = 0.02) → 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.

  • coherence (float, optional) – Memory coherence time in seconds used to

  • fidelities. (compute edge)

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.ppacket_dirname(value: float) → str[source]
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: pandas.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, app_request_rows: List[Dict[str, Any]] | None = None, 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, multi_path_apps: Iterable[str] | 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

  • 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, and deadline budget.

  • 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]

Track waiting time statistics per link.

Parameters:

  • waiting_time (float) – Wait incurred by a single deferral.

  • 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)

Note

block_events counts blocking episodes (one per deferral per link), not distinct PGAs, so a single PGA deferred repeatedly contributes multiple counts.

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) → pandas.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[pandas.DataFrame, str][source]
utils.parallel_simulations.simulate_one_ppacket(args: tuple) → dict[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, overlay_multipath: bool = False, y_start_zero: bool = False) → list[dict[str, Any]][source]
utils.plots.main()[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', overlay_multipath: bool = False, out_dir: str | None = None) → pandas.DataFrame[source]

Example usage: df = plot_metrics_vs_load(

path=[

“1.csv”, “2.csv”, “3.csv”, “4.csv”, “5.csv”,

], multi=True, gp_labels={

“1”: “Precomputed”, “2”: “Proactive”, “3”: “Hybrid”, “4”: “Reactive (wc)”, “5”: “Reactive (nwc)”,

}, overlay_multipath=True,

)

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, overlay_multipath: bool = False) → pandas.DataFrame[source]
utils.plots.render_plot(spec: dict[str, Any], raw_data: pandas.DataFrame, figsize: tuple[float, float], dpi: int) → None[source]
utils.plots.set_plot_theme(dpi: int) → None[source]

utils.plots_graph module

utils.plots_graph.plot_graph_from_gml(gml_file: str) → None[source]

Plot a graph from a GML file.

Parameters:

gml_file (str) – Path to the GML file.

utils.run_scenarios module

utils.run_scenarios.main()[source]

Module contents