Source code for ra2ce.analysis.losses.multi_link_origin_destination

import copy
from pathlib import Path

import networkx as nx
import numpy as np
import pandas as pd
from geopandas import GeoDataFrame, read_feather, read_file

from ra2ce.analysis.analysis_base import AnalysisBase
from ra2ce.analysis.analysis_config_data.analysis_config_data import (
    AnalysisSectionLosses,
)
from ra2ce.analysis.analysis_input_wrapper import AnalysisInputWrapper
from ra2ce.analysis.analysis_result.analysis_result_wrapper import AnalysisResultWrapper
from ra2ce.analysis.losses.analysis_losses_protocol import AnalysisLossesProtocol
from ra2ce.analysis.losses.optimal_route_origin_destination import (
    OptimalRouteOriginDestination,
)
from ra2ce.network.graph_files.graph_file import GraphFile
from ra2ce.network.hazard.hazard_names import HazardNames
from ra2ce.network.network_config_data.network_config_data import (
    OriginsDestinationsSection,
)




[docs]
class MultiLinkOriginDestination(AnalysisBase, AnalysisLossesProtocol):
    analysis: AnalysisSectionLosses
    graph_file_hazard: GraphFile
    input_path: Path
    static_path: Path
    output_path: Path
    hazard_names: HazardNames
    origins_destinations: OriginsDestinationsSection
    _analysis_input: AnalysisInputWrapper

    def __init__(
        self,
        analysis_input: AnalysisInputWrapper,
    ) -> None:
        self.analysis = analysis_input.analysis
        self.graph_file_hazard = analysis_input.graph_file_hazard
        self.input_path = analysis_input.input_path
        self.static_path = analysis_input.static_path
        self.output_path = analysis_input.output_path
        self.hazard_names = analysis_input.hazard_names
        self.origins_destinations = analysis_input.origins_destinations
        self._analysis_input = analysis_input



[docs]
    @staticmethod
    def extract_od_nodes_from_graph(
        graph: nx.MultiGraph,
    ) -> list[tuple[str, str]]:
        """
        Extracts all Origin - Destination nodes from the graph, prevents from entries
        with list of nodes for a node.

        Args:
            graph (nx.MultiGraph): Graph containing origin-destination nodes.

        Returns:
            list[tuple[str, str]]]: List containing tuples of origin - destination node combinations.
        """
        # NOTE: this code is duplicated in optimal_route_origin_destination.py
        _od_nodes = []
        for n, v in graph.nodes(data=True):
            if "od_id" not in v:
                continue
            _o_node_list = list(map(lambda x: (n, x), v["od_id"].split(",")))
            _od_nodes.extend(_o_node_list)
        return _od_nodes



    def _get_origin_destination_pairs(
        self, graph: nx.MultiGraph
    ) -> list[tuple[tuple[str, str], tuple[str, str]]]:
        # NOTE: this code is duplicated in optimal_route_origin_destination.py
        od_path = self.static_path.joinpath(
            "output_graph", "origin_destination_table.feather"
        )
        od = read_feather(od_path)
        od_pairs = [
            (a, b)
            for a in od.loc[od["o_id"].notnull(), "o_id"]
            for b in od.loc[od["d_id"].notnull(), "d_id"]
        ]
        all_nodes = self.extract_od_nodes_from_graph(graph)

        def _find_node(name: str, nodes: list[tuple[str, str]]) -> tuple[str, str]:
            # it is possible that there are multiple origins/destinations at the same 'entry-point' in the road
            return next(
                (n, n_name)
                for n, n_name in nodes
                if n_name == name or name in n_name.split(",")
            )

        return [
            (_find_node(aa, all_nodes), _find_node(bb, all_nodes))
            for aa, bb in od_pairs
        ]












[docs]
    def execute(self) -> AnalysisResultWrapper:
        _output_path = self.output_path.joinpath(self.analysis.analysis.config_value)
        gdf = self.multi_link_origin_destination(
            self.graph_file_hazard.get_graph(), self.analysis
        )
        self._analysis_input.graph_file = self._analysis_input.graph_file_hazard
        _orod_result_wrapper = OptimalRouteOriginDestination(
            self._analysis_input
        ).execute()
        (disruption_impact_df, gdf_ori,) = self.multi_link_origin_destination_impact(
            gdf, _orod_result_wrapper.get_single_result()
        )
        try:
            assert self.origins_destinations.region
            (
                regional_impact_df,
                regional_impact_summary_df,
            ) = self.multi_link_origin_destination_regional_impact(gdf_ori)
            impact_csv_path = _output_path.joinpath(
                (self.analysis.name.replace(" ", "_") + "_regional_impact.csv"),
            )
            regional_impact_df.to_csv(impact_csv_path, index=False)
            impact_csv_path = _output_path.joinpath(
                (self.analysis.name.replace(" ", "_") + "_regional_impact_summary.csv"),
            )
            regional_impact_summary_df.to_csv(impact_csv_path)
        except Exception:
            pass
        impact_csv_path = _output_path.joinpath(
            (self.analysis.name.replace(" ", "_") + "_impact.csv"),
        )
        del gdf_ori["geometry"]
        gdf_ori.to_csv(impact_csv_path, index=False)
        impact_csv_path = _output_path.joinpath(
            (self.analysis.name.replace(" ", "_") + "_impact_summary.csv"),
        )
        disruption_impact_df.to_csv(impact_csv_path, index=False)

        return self.generate_result_wrapper(gdf)