summaryrefslogtreecommitdiff
path: root/backend/src/loader.py
blob: 7e536c13d4c577c44afef3b23dd778183c159e80 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
import csv
import logging
from pathlib import Path
from typing import Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

BASE_DIR = Path(__file__).resolve().parent.parent / 'data'
MAP_DENORMALIZED_FILE = BASE_DIR / 'mapDenormalize.csv'
MAP_JUMPS_FILE = BASE_DIR / 'mapJumps.csv'
INV_TYPES_FILE = BASE_DIR / 'invTypes.csv'
REGION_IDS_FILE = BASE_DIR / 'mapRegions.csv'

# Define constants
TYPE_ID_SOLAR_SYSTEM = 5
GROUP_ID_STATION = 15
GROUP_ID_STARGATE = 10

REGION_IDS = [
    10000002,  # The Forge
    10000016,  # Lonetrek
    10000030,  # Heimatar
    10000033,  # The Citadel
    10000032,  # Sinq Laison
    10000042,  # Metropolis
    10000064,  # Essence
    10000037,  # Everyshore
    10000043,  # Domain
    10000036,  # Devoid
    10000052,  # Kador
    10000067,  # Genesis
    10000068,  # Verge Vendor
    10000001,  # Derelik
    10000020,  # Tash-Murkon
]

def load_map_data() -> tuple[dict[int, Any], dict[int, Any]]:
    """
    Loads map data including stations and solar systems.

    Returns:
        Tuple containing dictionaries of stations and solar systems.
    """
    regions = read_regions()
    stations: dict[int, Any] = {}
    solar_systems: dict[int, Any] = {}

    with MAP_DENORMALIZED_FILE.open(newline='', encoding='utf-8') as csvfile:
        reader = csv.reader(csvfile)
        next(reader)  # Skip header
        for row in reader:
            process_row(row, solar_systems, stations, regions)

    # Update station data with additional solar system and region info
    for station_id, station_data in stations.items():
        ss_id = station_data['solar_system_id']
        solar_system = solar_systems.get(ss_id)
        if solar_system:
            station_data['solar_system_name'] = solar_system['item_name']
            station_data['solar_system_position'] = solar_system['position']
            station_data['region_id'] = solar_system['region_id']
            station_data['region_name'] = regions.get(station_data['region_id'], 'Unknown')

    return stations, solar_systems

def process_row(
    row: list[str],
    solar_systems: dict[int, Any],
    stations: dict[int, Any],
    regions: dict[int, str]
) -> None:
    """
    Processes a single row from the map denormalized CSV file.

    Args:
        row: A list of CSV fields.
        solar_systems: Dictionary to store solar system data.
        stations: Dictionary to store station data.
        regions: Dictionary of region IDs to region names.
    """
    (
        item_id, type_id, group_id, solar_system_id,
        _, region_id, _, x, y, z, _,
        item_name, security, _, _
    ) = row

    try:
        region_id_int = int(region_id)
    except (ValueError, TypeError):
        return

    if region_id_int not in REGION_IDS:
        return

    item_id_int = int(item_id)
    type_id_int = int(type_id)
    group_id_int = int(group_id)
    solar_system_id_int = int(solar_system_id)

    if type_id_int == TYPE_ID_SOLAR_SYSTEM:
        solar_systems[item_id_int] = {
            'item_name': item_name,
            'position': (float(x), float(y), float(z)),
            'region_id': region_id_int,
            'region_name': regions.get(region_id_int, 'Unknown')
        }
    elif group_id_int in (GROUP_ID_STATION, GROUP_ID_STARGATE):
        stations[item_id_int] = {
            'is_station': group_id_int == GROUP_ID_STATION,
            'item_name': item_name,
            'security': float(security),
            'position': (float(x), float(y), float(z)),
            'solar_system_id': solar_system_id_int,
            'region_id': region_id_int,
            'jumps': []
        }

def load_items() -> dict[int, Any]:
    """
    Loads item data from the inventory types CSV file.

    Returns:
        Dictionary of items with type IDs as keys.
    """
    items: dict[int, Any] = {}
    with INV_TYPES_FILE.open(newline='', encoding='utf-8') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            try:
                type_id = int(row['typeID'])
                items[type_id] = {
                    'type_id': type_id,
                    'group_id': int(row['groupID']),
                    'type_name': row['typeName'],
                    'description': row['description'],
                    'mass': float(row['mass']) if row['mass'] else None,
                    'volume': float(row['volume']) if row['volume'] else None,
                    'capacity': float(row['capacity']) if row['capacity'] else None,
                }
            except ValueError as e:
                logger.warning(f"Could not parse item with typeID {row.get('typeID')}: {e}")
            except KeyError as e:
                logger.error(f"Missing expected field in inventory types CSV: {e}")
    return items

def read_regions() -> dict[int, str]:
    """
    Reads region data from the regions CSV file.

    Returns:
        Dictionary mapping region IDs to region names.
    """
    regions: dict[int, str] = {}
    with REGION_IDS_FILE.open(newline='', encoding='utf-8') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            try:
                region_id = int(row['regionID'])
                regions[region_id] = row['regionName'].strip()
            except ValueError as e:
                logger.warning(f"Could not parse region ID: {e}")
            except KeyError as e:
                logger.error(f"Missing expected field in regions CSV: {e}")
    return regions

def load_star_gate_connections() -> list[tuple[int, int]]:
    """
    Loads stargate connections from the map jumps CSV file.

    Returns:
        List of tuples representing stargate connections.
    """
    star_gate_connections: list[tuple[int, int]] = []
    with MAP_JUMPS_FILE.open(newline='', encoding='utf-8') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            try:
                from_stargate = int(row['stargateID'])
                to_stargate = int(row['destinationID'])
                star_gate_connections.append((from_stargate, to_stargate))
            except ValueError as e:
                logger.warning(f"Could not parse stargate connection: {e}")
            except KeyError as e:
                logger.error(f"Missing expected field in stargate jumps CSV: {e}")
    return star_gate_connections