import logging import os import pickle import networkx as nx from math import sqrt, log from pathlib import Path from typing import Any, Optional 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' 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) self.paths = self._load_paths() 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 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 self.paths[source][target] except KeyError: logger.warning(f"No path length found between {source} and {target}") return None 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. """ combined_path = self._combine_warp_gates(path) formatted_route = [] for index, location_id in enumerate(combined_path): location_info = self._get_location_info(location_id, index) if location_info is not None: formatted_route.append(location_info) return formatted_route 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 _combine_warp_gates(self, path: 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 = [] for i, (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) 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']) last_station = self.stations.get(path[-1]) if last_station and last_station['is_station']: combined_path.append(path[-1]) return combined_path 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_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]) 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_station: int, to_station: int, ship: dict[str, Any]) -> float: """ Calculates the time to travel between two stations. Args: from_station: The starting station ID. to_station: The destination station ID. ship: A dictionary containing ship parameters. Returns: The time in seconds to travel between the two stations. """ 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 # Intra-system travel position_from = self.stations[from_station]['position'] position_to = self.stations[to_station]['position'] 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_station: int, gankers_areas: dict[str, float]) -> float: """ Calculates the risk of traveling to a station. Args: to_station: The destination station ID. 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', '') # 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 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