Skip to content

dtos

cosmica.experimental_packet_routing.dtos

__all__ module-attribute 

__all__ = ['PacketRoutingResult', 'PacketRoutingSetting']

logger module-attribute 

logger: Logger = getLogger(__name__)

PacketRoutingResult dataclass 

PacketRoutingResult(
    *,
    all_graphs_after_simulation: list[Graph] = list[
        Graph
    ](),
    node_knowledge_known_by_each_node: dict[
        Node, NodeKnowledge
    ] = dict[Node, NodeKnowledge](),
    comm_data_demand_list: list[CommDataDemand] = list[
        CommDataDemand
    ](),
    comm_data_lsa_list: list[CommDataLSA] = list[
        CommDataLSA
    ]()
)

all_graphs_after_simulation class-attribute instance-attribute 

all_graphs_after_simulation: list[Graph] = field(
    default_factory=list[Graph]
)

comm_data_demand_list class-attribute instance-attribute 

comm_data_demand_list: list[CommDataDemand] = field(
    default_factory=list[CommDataDemand]
)

comm_data_lsa_list class-attribute instance-attribute 

comm_data_lsa_list: list[CommDataLSA] = field(
    default_factory=list[CommDataLSA]
)

node_knowledge_known_by_each_node class-attribute instance-attribute 

node_knowledge_known_by_each_node: dict[
    Node, NodeKnowledge
] = field(default_factory=dict[Node, NodeKnowledge])

load classmethod 

load(
    graphs_path: Path | None = None,
    node_knowledge_path: Path | None = None,
    comm_data_demand_path: Path | None = None,
    comm_data_lsa_path: Path | None = None,
) -> Self

指定されたパスからpickleファイルを読み込み.

Source code in src/cosmica/experimental_packet_routing/dtos.py 
 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
@classmethod
def load(
    cls,
    graphs_path: Path | None = None,
    node_knowledge_path: Path | None = None,
    comm_data_demand_path: Path | None = None,
    comm_data_lsa_path: Path | None = None,
) -> Self:
    """指定されたパスからpickleファイルを読み込み."""
    all_graphs_after_simulation = []
    if graphs_path is not None and graphs_path.exists():
        with graphs_path.open("rb") as f:
            all_graphs_after_simulation = pickle.load(f)  # noqa: S301
        logger.info(f"Loaded all graphs after simulation: {len(all_graphs_after_simulation)}")
    else:
        logger.info("No graph data to load.")

    node_knowledge_known_by_each_node = {}
    if node_knowledge_path is not None and node_knowledge_path.exists():
        with node_knowledge_path.open("rb") as f:
            node_knowledge_known_by_each_node = pickle.load(f)  # noqa: S301
        logger.info(f"Loaded network information: {len(node_knowledge_known_by_each_node)}")
    else:
        logger.info("No node knowledge data to load.")

    comm_data_demand_list = []
    if comm_data_demand_path is not None and comm_data_demand_path.exists():
        with comm_data_demand_path.open("rb") as f:
            comm_data_demand_list = pickle.load(f)  # noqa: S301
        logger.info(f"Loaded comm data demand: {len(comm_data_demand_list)}")
    else:
        logger.info("No comm data demand to load.")

    comm_data_lsa_list = []
    if comm_data_lsa_path is not None and comm_data_lsa_path.exists():
        with comm_data_lsa_path.open("rb") as f:
            comm_data_lsa_list = pickle.load(f)  # noqa: S301
        logger.info(f"Loaded comm data lsa: {len(comm_data_lsa_list)}")
    else:
        logger.info("No comm data lsa to load.")

    return cls(
        all_graphs_after_simulation=all_graphs_after_simulation,
        node_knowledge_known_by_each_node=node_knowledge_known_by_each_node,
        comm_data_demand_list=comm_data_demand_list,
        comm_data_lsa_list=comm_data_lsa_list,
    )

save_all_graphs_after_simulation 

save_all_graphs_after_simulation(save_path: Path) -> None
Source code in src/cosmica/experimental_packet_routing/dtos.py 
44
45
46
47
48
49
50
def save_all_graphs_after_simulation(self, save_path: Path) -> None:
    if self.all_graphs_after_simulation is not None:
        with save_path.open("wb") as f:
            pickle.dump(self.all_graphs_after_simulation, f)
        logger.info(f"Saved the simulation results of graph to {save_path}")
    else:
        logger.info("No graph data to save.")

save_comm_data_demand_list 

save_comm_data_demand_list(save_path: Path) -> None
Source code in src/cosmica/experimental_packet_routing/dtos.py 
60
61
62
63
64
65
66
def save_comm_data_demand_list(self, save_path: Path) -> None:
    if self.comm_data_demand_list is not None:
        with save_path.open("wb") as f:
            pickle.dump(self.comm_data_demand_list, f)
        logger.info(f"Saved the simulation results of comm data demand to {save_path}")
    else:
        logger.info("No comm data demand to save.")

save_comm_data_lsa_list 

save_comm_data_lsa_list(save_path: Path) -> None
Source code in src/cosmica/experimental_packet_routing/dtos.py 
68
69
70
71
72
73
74
def save_comm_data_lsa_list(self, save_path: Path) -> None:
    if self.comm_data_lsa_list is not None:
        with save_path.open("wb") as f:
            pickle.dump(self.comm_data_lsa_list, f)
        logger.info(f"Saved the simulation results of comm data lsa to {save_path}")
    else:
        logger.info("No comm data lsa to save.")

save_node_knowledge_known_by_each_node 

save_node_knowledge_known_by_each_node(
    save_path: Path,
) -> None
Source code in src/cosmica/experimental_packet_routing/dtos.py 
52
53
54
55
56
57
58
def save_node_knowledge_known_by_each_node(self, save_path: Path) -> None:
    if self.node_knowledge_known_by_each_node is not None:
        with save_path.open("wb") as f:
            pickle.dump(self.node_knowledge_known_by_each_node, f)
        logger.info(f"Saved the simulation results of node knowledge to {save_path}")
    else:
        logger.info("No node knowledge data to save.")

PacketRoutingSetting dataclass 

PacketRoutingSetting(
    *,
    time: NDArray[datetime64],
    nodes_dict: dict[NodeGID, Node],
    demands: list[Demand],
    backup_case: BackupCaseType,
    hop_limit: int,
    packet_size: int
)

backup_case instance-attribute 

backup_case: BackupCaseType

demands instance-attribute 

demands: list[Demand]

hop_limit instance-attribute 

hop_limit: int

nodes_dict instance-attribute 

nodes_dict: dict[NodeGID, Node]

packet_size instance-attribute 

packet_size: int

time instance-attribute 

time: NDArray[datetime64]

load classmethod 

load(load_path: Path) -> PacketRoutingSetting
Source code in src/cosmica/experimental_packet_routing/dtos.py 
140
141
142
143
144
145
@classmethod
def load(cls, load_path: Path) -> PacketRoutingSetting:
    with load_path.open("rb") as f:
        return pickle.load(f)  # noqa: S301
    logger.info(f"Loaded the packet routing setting from {load_path}")
    return None

save 

save(save_path: Path) -> None
Source code in src/cosmica/experimental_packet_routing/dtos.py 
135
136
137
138
def save(self, save_path: Path) -> None:
    with save_path.open("wb") as f:
        pickle.dump(self, f)
    logger.info(f"Saved the packet routing setting to {save_path}")