import logging import functools from dataclasses import dataclass import networkx as nx from math import sqrt, log 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__) LOCAL_SYSTEM_HUB_POSITION = (0.0, 0.0, 0.0) @dataclass(frozen=True) class LegMetrics: weight: float time: float risk: float class NavigationGraph: """ Represents the navigation graph for stations and solar systems, allowing computation of shortest paths based on time and risk. """ def __init__(self, ship: dict[str, Any]): """ Initializes the NavigationGraph. Args: ship: A dictionary containing ship parameters like max warp speed, max subwarp speed, etc. """ self.graph = nx.Graph() self.stations, self.solar_systems = load_map_data() self._add_intra_system_connections() self._load_star_gate_connections() self._add_time_and_risk(ship) def shortest_path(self, source: int, target: int) -> Optional[list[int]]: """ Returns the shortest path between two stations. Args: source: The starting station ID. target: The target station ID. Returns: A list of station IDs representing the shortest path, or None if no path exists. """ try: path = nx.shortest_path(self.graph, source, target, weight="weight") return path except nx.NetworkXNoPath: 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. Args: source: The starting station ID. target: The target station ID. Returns: The length of the shortest path, or None if no path exists. """ try: 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. Args: path: A list of station IDs representing the path. Returns: A list of dictionaries containing location information. """ 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 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, formatted_offsets def _get_location_info(self, location_id: int, index: int) -> Optional[dict[str, Any]]: """ Retrieves location information for a given ID. Args: location_id: The ID of the location (station or system). index: The position in the route. Returns: A dictionary with location details, or None if not found. """ if location_id in self.stations: station_data = self.stations[location_id] solar_system = self.solar_systems.get(station_data['solar_system_id'], {}) return { "id": index, "location_id": location_id, "station_id": location_id, "solar_system_id": station_data['solar_system_id'], "location_type": "station", "location": f"{station_data['region_name']} - {solar_system.get('item_name', '')} - {station_data['item_name']}", "actions": [], } elif location_id in self.solar_systems: solar_system_data = self.solar_systems[location_id] return { "id": index, "location_id": location_id, "solar_system_id": location_id, "location_type": "system", "location": f"{solar_system_data['region_name']} - {solar_system_data['item_name']}", "actions": [], } else: logger.warning(f"Location ID {location_id} not found in stations or solar systems") return None 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. Args: path: A list of station IDs. Returns: A list of location IDs with combined warp gates. """ combined_path = [] raw_offsets = [] 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, raw_offsets def _add_intra_system_connections(self) -> None: """ Adds connections between stations within the same solar system. """ stations_by_system: dict[int, list[int]] = {} for station_id, station_data in self.stations.items(): system_id = station_data['solar_system_id'] stations_by_system.setdefault(system_id, []).append(station_id) 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: """ Loads stargate connections into the graph. """ connections = load_star_gate_connections() for from_station, to_station in connections: if from_station in self.stations and to_station in self.stations: self.graph.add_edge(from_station, to_station) def _add_time_and_risk(self, ship: dict[str, Any]) -> None: """ Calculates and adds time and risk attributes to graph edges. Args: ship: A dictionary containing ship parameters. """ for u, v in self.graph.edges(): time = self._calculate_time(u, v, ship) risk = self._calculate_risk(v, ship.get("gankers_areas", {})) weight = time * ship.get("player_cost", 1) + ship.get("risk_cost", 1) * risk self.graph[u][v]['time'] = time self.graph[u][v]['risk'] = risk self.graph[u][v]['weight'] = weight def _calculate_time(self, from_node: int, to_node: int, ship: dict[str, Any]) -> float: """ Calculates the time to travel between two nodes. Args: 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 nodes. """ max_warp_speed = ship.get("max_warp_speed", 1) max_subwarp_speed = ship.get("max_subwarp_speed", 1) # 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 (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) time_in_warp += 2.0 # Add 2 seconds for initial delay return time_in_warp def _calculate_time_in_warp(self, warp_dist: float, max_warp_speed: float, max_subwarp_speed: float) -> float: """ Calculates the time spent in warp between two points. Args: warp_dist: The distance to warp in meters. max_warp_speed: The maximum warp speed of the ship in AU/s. max_subwarp_speed: The maximum sub-warp speed of the ship in m/s. Returns: The time in seconds spent in warp. """ AU_IN_M = 149_597_870_700 # Astronomical Unit in meters k_accel = max_warp_speed k_decel = min(max_warp_speed / 3, 2) warp_dropout_speed = min(max_subwarp_speed / 2, 100) max_ms_warp_speed = max_warp_speed * AU_IN_M accel_dist = AU_IN_M decel_dist = max_ms_warp_speed / k_decel minimum_dist = accel_dist + decel_dist cruise_time = 0.0 if minimum_dist > warp_dist: max_ms_warp_speed = warp_dist * k_accel * k_decel / (k_accel + k_decel) else: cruise_time = (warp_dist - minimum_dist) / max_ms_warp_speed accel_time = log(max_ms_warp_speed / k_accel) / k_accel decel_time = log(max_ms_warp_speed / warp_dropout_speed) / k_decel total_time = cruise_time + accel_time + decel_time return total_time def _calculate_risk(self, to_node: int, gankers_areas: dict[str, float]) -> float: """ Calculates the risk of traveling to a node. Args: 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. """ # 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 if security < 0.5: risk += 0.005 if security < 0.1: risk += 0.5 risk += gankers_areas.get(solar_system_name, 0) return risk