diff options
Diffstat (limited to 'backend/src/graph.py')
| -rw-r--r-- | backend/src/graph.py | 229 |
1 files changed, 175 insertions, 54 deletions
diff --git a/backend/src/graph.py b/backend/src/graph.py index 9e4534a..951bd45 100644 --- a/backend/src/graph.py +++ b/backend/src/graph.py @@ -1,17 +1,24 @@ import logging -import os -import pickle +import functools +from dataclasses import dataclass import networkx as nx from math import sqrt, log -from pathlib import Path from typing import Any, Optional +from tqdm import tqdm from loader import load_map_data, load_star_gate_connections logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) -PATHS_FILE = Path(__file__).resolve().parent.parent / 'cache' / 'path_cache.pkl' +LOCAL_SYSTEM_HUB_POSITION = (0.0, 0.0, 0.0) + + +@dataclass(frozen=True) +class LegMetrics: + weight: float + time: float + risk: float class NavigationGraph: @@ -33,7 +40,6 @@ class NavigationGraph: self._add_intra_system_connections() self._load_star_gate_connections() self._add_time_and_risk(ship) - self.paths = self._load_paths() def shortest_path(self, source: int, target: int) -> Optional[list[int]]: """ @@ -54,6 +60,7 @@ class NavigationGraph: logger.warning(f"No path found between {source} and {target}") return None + @functools.lru_cache(maxsize=100000) def shortest_path_length(self, source: int, target: int) -> Optional[float]: """ Returns the length of the shortest path between two stations. @@ -66,11 +73,75 @@ class NavigationGraph: The length of the shortest path, or None if no path exists. """ try: - return self.paths[source][target] - except KeyError: + return nx.shortest_path_length(self.graph, source=source, target=target, weight="weight") + except nx.NetworkXNoPath: logger.warning(f"No path length found between {source} and {target}") return None + def build_trade_leg_metrics(self, ship_location: Optional[int], trades: list[Any]) -> dict[tuple[int, int], Optional[LegMetrics]]: + """ + Builds weighted travel metrics for the legs used by a list of trades. + + Args: + ship_location: The current ship location. + trades: The trade opportunities that need pricing. + + Returns: + A mapping of (source, target) pairs to weighted travel metrics. + """ + if ship_location is None: + return {} + + source_targets: dict[int, set[int]] = {ship_location: set()} + for trade in trades: + source_targets.setdefault(ship_location, set()).add(trade.from_station) + source_targets.setdefault(trade.from_station, set()).add(trade.to_station) + + return self.build_leg_metrics(source_targets) + + def build_leg_metrics(self, source_targets: dict[int, set[int]]) -> dict[tuple[int, int], Optional[LegMetrics]]: + """ + Computes weighted leg metrics for a batch of sources and targets. + + Args: + source_targets: Mapping of a source node to the targets needed from it. + + Returns: + A mapping of (source, target) pairs to weighted travel metrics. + """ + leg_metrics: dict[tuple[int, int], Optional[LegMetrics]] = {} + + for source, targets in tqdm( + source_targets.items(), + total=len(source_targets), + desc="Building leg metrics", + ): + if source not in self.graph: + for target in targets: + leg_metrics[(source, target)] = None if source != target else LegMetrics(0.0, 0.0, 0.0) + continue + + try: + weights, paths = nx.single_source_dijkstra(self.graph, source=source, weight="weight") + except nx.NodeNotFound: + for target in targets: + leg_metrics[(source, target)] = None if source != target else LegMetrics(0.0, 0.0, 0.0) + continue + + for target in targets: + if source == target: + leg_metrics[(source, target)] = LegMetrics(0.0, 0.0, 0.0) + continue + + path = paths.get(target) + if path is None: + leg_metrics[(source, target)] = None + continue + + leg_metrics[(source, target)] = self._summarize_path(path, weights[target]) + + return leg_metrics + def formatted_route(self, path: list[int]) -> list[dict[str, Any]]: """ Formats the route for the ESI API and frontend. @@ -81,15 +152,30 @@ class NavigationGraph: Returns: A list of dictionaries containing location information. """ - combined_path = self._combine_warp_gates(path) + formatted_route, _ = self.formatted_route_with_offsets(path) + return formatted_route + + def formatted_route_with_offsets(self, path: list[int]) -> tuple[list[dict[str, Any]], list[int]]: + """ + Formats the route and returns the raw-path offset for each display step. + + Args: + path: A list of station IDs representing the raw path. + + Returns: + A tuple of formatted route steps and their starting offsets in the raw path. + """ + combined_path, raw_offsets = self._combine_warp_gates(path) formatted_route = [] + formatted_offsets = [] - for index, location_id in enumerate(combined_path): - location_info = self._get_location_info(location_id, index) + for location_id, raw_offset in zip(combined_path, raw_offsets): + location_info = self._get_location_info(location_id, len(formatted_route)) if location_info is not None: formatted_route.append(location_info) + formatted_offsets.append(raw_offset) - return formatted_route + return formatted_route, formatted_offsets def _get_location_info(self, location_id: int, index: int) -> Optional[dict[str, Any]]: """ @@ -128,7 +214,28 @@ class NavigationGraph: logger.warning(f"Location ID {location_id} not found in stations or solar systems") return None - def _combine_warp_gates(self, path: list[int]) -> list[int]: + def _summarize_path(self, path: list[int], weight: float) -> LegMetrics: + """ + Summarizes the time and risk of a chosen weighted path. + + Args: + path: The chosen weighted path. + weight: The total weighted cost of the path. + + Returns: + The weighted, time, and risk totals for that path. + """ + transport_time = 0.0 + risk = 0.0 + + for u, v in zip(path, path[1:]): + edge = self.graph[u][v] + transport_time += edge["time"] + risk += edge["risk"] + + return LegMetrics(weight=weight, time=transport_time, risk=risk) + + def _combine_warp_gates(self, path: list[int]) -> tuple[list[int], list[int]]: """ Combines warp gates in the path to reduce redundant entries. @@ -139,24 +246,28 @@ class NavigationGraph: A list of location IDs with combined warp gates. """ combined_path = [] + raw_offsets = [] - for i, (current_id, next_id) in enumerate(zip(path, path[1:])): + for index, (current_id, next_id) in enumerate(zip(path, path[1:])): current_station = self.stations.get(current_id) next_station = self.stations.get(next_id) if current_station and current_station['is_station']: combined_path.append(current_id) + raw_offsets.append(index) elif (current_station and next_station and not current_station['is_station'] and not next_station['is_station'] and current_station['solar_system_id'] == next_station['solar_system_id']): combined_path.append(current_station['solar_system_id']) + raw_offsets.append(index) last_station = self.stations.get(path[-1]) if last_station and last_station['is_station']: combined_path.append(path[-1]) + raw_offsets.append(len(path) - 1) - return combined_path + return combined_path, raw_offsets def _add_intra_system_connections(self) -> None: """ @@ -167,10 +278,16 @@ class NavigationGraph: system_id = station_data['solar_system_id'] stations_by_system.setdefault(system_id, []).append(station_id) - for system_stations in stations_by_system.values(): - for i in range(len(system_stations)): - for j in range(i + 1, len(system_stations)): - self.graph.add_edge(system_stations[i], system_stations[j]) + for system_id, system_stations in stations_by_system.items(): + self.graph.add_node(system_id) + for station_id in system_stations: + self.graph.add_edge(system_id, station_id) + + # Keep direct same-system edges so weighted pricing uses the real + # station-to-station distance instead of an artificial hub detour. + for index, from_station in enumerate(system_stations): + for to_station in system_stations[index + 1:]: + self.graph.add_edge(from_station, to_station) def _load_star_gate_connections(self) -> None: """ @@ -196,28 +313,40 @@ class NavigationGraph: self.graph[u][v]['risk'] = risk self.graph[u][v]['weight'] = weight - def _calculate_time(self, from_station: int, to_station: int, ship: dict[str, Any]) -> float: + def _calculate_time(self, from_node: int, to_node: int, ship: dict[str, Any]) -> float: """ - Calculates the time to travel between two stations. + Calculates the time to travel between two nodes. Args: - from_station: The starting station ID. - to_station: The destination station ID. + from_node: The starting node ID (station or system hub). + to_node: The destination node ID (station or system hub). ship: A dictionary containing ship parameters. Returns: - The time in seconds to travel between the two stations. + The time in seconds to travel between the two nodes. """ max_warp_speed = ship.get("max_warp_speed", 1) max_subwarp_speed = ship.get("max_subwarp_speed", 1) - # Inter-system travel (using stargates) - if self.stations[from_station]['solar_system_id'] != self.stations[to_station]['solar_system_id']: - return 9.0 # Fixed time for stargate jumps + # Helper to get node position + def get_node_position(node_id): + if node_id in self.stations: + return self.stations[node_id]['position'] + elif node_id in self.solar_systems: + # Solar-system rows use universe coordinates while stations and + # stargates use local system coordinates. Synthetic hub nodes + # must stay in the local frame to avoid impossible warp lengths. + return LOCAL_SYSTEM_HUB_POSITION + return (0.0, 0.0, 0.0) + + # Both are stations: check if it's an inter-system jump + if from_node in self.stations and to_node in self.stations: + if self.stations[from_node]['solar_system_id'] != self.stations[to_node]['solar_system_id']: + return 9.0 # Fixed time for stargate jumps - # Intra-system travel - position_from = self.stations[from_station]['position'] - position_to = self.stations[to_station]['position'] + # Intra-system travel (station <-> hub or station <-> system) + position_from = get_node_position(from_node) + position_to = get_node_position(to_node) warp_dist = sqrt(sum((position_from[i] - position_to[i]) ** 2 for i in range(3))) time_in_warp = self._calculate_time_in_warp(warp_dist, max_warp_speed, max_subwarp_speed) @@ -260,20 +389,31 @@ class NavigationGraph: total_time = cruise_time + accel_time + decel_time return total_time - def _calculate_risk(self, to_station: int, gankers_areas: dict[str, float]) -> float: + def _calculate_risk(self, to_node: int, gankers_areas: dict[str, float]) -> float: """ - Calculates the risk of traveling to a station. + Calculates the risk of traveling to a node. Args: - to_station: The destination station ID. + to_node: The destination node ID (station or system hub). gankers_areas: A dictionary mapping solar system names to risk factors. Returns: A risk score as a float. """ - station_data = self.stations[to_station] - security = station_data.get('security', 1.0) - solar_system_name = station_data.get('solar_system_name', '') + # Determine security and solar_system_name based on node type + if to_node in self.stations: + station_data = self.stations[to_node] + security = station_data.get('security', 1.0) + solar_system_name = station_data.get('solar_system_name', '') + elif to_node in self.solar_systems: + system_data = self.solar_systems[to_node] + # Can default to high-sec if system info doesn't have security, + # but usually it's known, actually we'll default to 1.0 + security = system_data.get('security', 1.0) + solar_system_name = system_data.get('item_name', '') + else: + security = 1.0 + solar_system_name = '' # Base risk calculation based on security status risk = 0.00004 + (1 - security) * 0.0004 @@ -285,22 +425,3 @@ class NavigationGraph: risk += gankers_areas.get(solar_system_name, 0) return risk - - def _load_paths(self) -> dict[int, dict[int, float]]: - """ - Loads or generates the shortest path lengths between all pairs. - - Returns: - A dictionary of dictionaries containing path lengths. - """ - if PATHS_FILE.exists(): - with PATHS_FILE.open("rb") as f: - logger.info("Loading cached paths from file") - return pickle.load(f) - else: - logger.info("Generating path lengths and caching to file") - dist = dict(nx.all_pairs_dijkstra_path_length(self.graph)) - PATHS_FILE.parent.mkdir(parents=True, exist_ok=True) - with PATHS_FILE.open("wb") as f: - pickle.dump(dist, f) - return dist |