-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
13 changed files
with
242 additions
and
205 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
36 changes: 28 additions & 8 deletions
36
ra2ce/analysis/losses/weighing_analysis/length_weighing_analysis.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,19 +1,39 @@ | ||
| from typing import Any | ||
|
|
||
| import geopandas as gpd | ||
| import numpy as np | ||
|
|
||
| from ra2ce.analysis.analysis_config_data.enums.weighing_enum import WeighingEnum | ||
| from ra2ce.analysis.losses.weighing_analysis.weighing_analysis_protocol import ( | ||
| WeighingAnalysisProtocol, | ||
| ) | ||
| from ra2ce.network.network_config_data.enums.road_type_enum import RoadTypeEnum | ||
| from ra2ce.network.networks_utils import get_avgspeed_per_road_type | ||
|
|
||
|
|
||
| class LengthWeighingAnalysis(WeighingAnalysisProtocol): | ||
| weighing_data: dict | ||
| edge_data: dict[str, Any] | ||
| avgspeed_dict: dict[str, float] | ||
|
|
||
| def calculate_distance(self) -> float: | ||
| return np.nan | ||
| def __init__(self, gdf_graph: gpd.GeoDataFrame | None) -> None: | ||
| self.avgspeed_dict = { | ||
| _road_type.config_value: get_avgspeed_per_road_type(gdf_graph, _road_type) | ||
| for _road_type in RoadTypeEnum | ||
| } | ||
|
|
||
| def calculate_alternative_distance(self, alt_dist: float) -> float: | ||
| return alt_dist | ||
| def _calculate_distance(self, time: float) -> float: | ||
| _avgspeed = self.edge_data.get("avgspeed", None) # km/h | ||
| if not _avgspeed: | ||
| _avgspeed = self.avgspeed_dict[self.edge_data["highway"]] | ||
| return round(time * _avgspeed * 1e3, 1) # m | ||
|
|
||
| def extend_graph(self, gdf_graph: gpd.GeoDataFrame | dict) -> None: | ||
| return | ||
| def get_current_value(self) -> float: | ||
| _dist = self.edge_data.get(WeighingEnum.LENGTH.config_value, None) # h | ||
| if _dist: | ||
| return round(_dist, 1) | ||
| _time = self.edge_data.get(WeighingEnum.TIME.config_value, 0) # m | ||
| _dist = self._calculate_distance(_time) | ||
| self.edge_data[WeighingEnum.LENGTH.config_value] = _dist | ||
| return _dist | ||
|
|
||
| def calculate_alternative_value(self, alt_dist: float) -> float: | ||
| return alt_dist |
68 changes: 30 additions & 38 deletions
68
ra2ce/analysis/losses/weighing_analysis/time_weighing_analysis.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,47 +1,39 @@ | ||
| from typing import Any | ||
|
|
||
| import geopandas as gpd | ||
| import numpy as np | ||
|
|
||
| from ra2ce.analysis.analysis_config_data.enums.weighing_enum import WeighingEnum | ||
| from ra2ce.analysis.losses.weighing_analysis.weighing_analysis_protocol import ( | ||
| WeighingAnalysisProtocol, | ||
| ) | ||
| from ra2ce.network.network_config_data.enums.road_type_enum import RoadTypeEnum | ||
| from ra2ce.network.networks_utils import get_avgspeed_per_road_type | ||
|
|
||
|
|
||
| class TimeWeighingAnalysis(WeighingAnalysisProtocol): | ||
| time_list: list | ||
| weighing_data: dict | ||
|
|
||
| def __init__(self) -> None: | ||
| self.time_list = [] | ||
|
|
||
| def _calculate_time(self) -> float: | ||
| length = self.weighing_data.get("length", None) | ||
| avgspeed = self.weighing_data.get("avgspeed", None) | ||
| if length and avgspeed: | ||
| _calculated_time = round( | ||
| (length * 1e-3) / avgspeed, | ||
| 3, | ||
| ) # in hours and avg speed in km/h | ||
| self.weighing_data[WeighingEnum.TIME.config_value] = _calculated_time | ||
| return round(_calculated_time, 3) | ||
| else: | ||
| return np.nan | ||
|
|
||
| def calculate_distance(self) -> float: | ||
| self.time_list.append(self._calculate_time()) | ||
| return self.time_list[-1] | ||
|
|
||
| def calculate_alternative_distance(self, alt_dist: float) -> float: | ||
| avgspeed = self.weighing_data.get("avgspeed", None) | ||
| if avgspeed: | ||
| alt_time = (alt_dist * 1e-3) / avgspeed # in hours | ||
| self.time_list.append(self._calculate_time()) | ||
| return alt_time | ||
| else: | ||
| return np.nan | ||
|
|
||
| def extend_graph(self, gdf_graph: gpd.GeoDataFrame | dict) -> None: | ||
| if isinstance(gdf_graph, gpd.GeoDataFrame): | ||
| gdf_graph[WeighingEnum.TIME.config_value] = self.time_list | ||
| elif isinstance(gdf_graph, dict): | ||
| gdf_graph[WeighingEnum.TIME.config_value] = [self.time_list[-1]] | ||
| edge_data: dict[str, Any] | ||
| avgspeed_dict: dict[str, float] | ||
|
|
||
| def __init__(self, gdf_graph: gpd.GeoDataFrame | None) -> None: | ||
| self.avgspeed_dict = { | ||
| _road_type.config_value: get_avgspeed_per_road_type(gdf_graph, _road_type) | ||
| for _road_type in RoadTypeEnum | ||
| } | ||
|
|
||
| def _calculate_time(self, dist: float) -> float: | ||
| _avgspeed = self.edge_data.get("avgspeed", None) # km/h | ||
| if not _avgspeed: | ||
| _avgspeed = self.avgspeed_dict[self.edge_data["highway"]] | ||
| return round(dist * 1e-3 / _avgspeed, 3) # h | ||
|
|
||
| def get_current_value(self) -> float: | ||
| _time = self.edge_data.get(WeighingEnum.TIME.config_value, None) # h | ||
| if _time: | ||
| return round(_time, 3) | ||
| _dist = self.edge_data.get(WeighingEnum.LENGTH.config_value, 0) # m | ||
| _time = self._calculate_time(_dist) | ||
| self.edge_data[WeighingEnum.TIME.config_value] = _time | ||
| return _time | ||
|
|
||
| def calculate_alternative_value(self, alt_dist: float) -> float: | ||
| return self._calculate_time(alt_dist) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.