path: root/backend/src/graph.py

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