""" Extracts Traffic Lanes, Pedestrian Splines, Interiors, and Traffic Signals from readable_world_dump_4229938_final.zip without unpacking the full 5.1 GB archive. Produces high-performance, compact JSON caches in Ananta.Server/ClientData/4229938/World/. """ import sys import os import io import json import math import csv import zipfile import time ZIP_PATH = r"c:\Users\Phitchayut\Desktop\Drop asset\readable_world_dump_4229938_final.zip" OUT_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "Ananta.Server", "ClientData", "4229938", "World") os.makedirs(OUT_DIR, exist_ok=True) print(f"[EXTRACT] Reading from: {ZIP_PATH}") print(f"[EXTRACT] Target output directory: {OUT_DIR}") t0 = time.time() with zipfile.ZipFile(ZIP_PATH, "r") as z: # ------------------------------------------------------------- # 1. Extract Road Networks (Vehicles) # ------------------------------------------------------------- road_sources = [ ("city", "readable_world_dump_4229938_final/03_ZONEGRAPHS/full_json/vfc_1286_e108_h2b81bc528d8b6797.json"), ("airport", "readable_world_dump_4229938_final/03_ZONEGRAPHS/full_json/vfc_999_e28_hcec9b1c776a677f4.json"), ("longqi", "readable_world_dump_4229938_final/03_ZONEGRAPHS/full_json/vfc_997_e3_h859e50d46cdca6f9.json"), ] all_road_lanes = [] spatial_grid = {} # "cellX_cellZ" -> [globalLaneIndex] grid_size = 100.0 # 100 meter spatial buckets global_lane_counter = 0 for region_name, zip_entry in road_sources: print(f"[ROAD] Processing {region_name} from {zip_entry}...") with z.open(zip_entry) as f: data = json.load(f) lanes = data.get("lanes", []) pts = data.get("lanePoints", []) tangents = data.get("laneTangentVectors", []) links = data.get("laneLinks", []) # Offset to map local destLaneIndex to globalLaneIndex base_lane_idx = global_lane_counter for local_id, lane in enumerate(lanes): tag_names = lane.get("tagNames", []) # Filter for vehicle road lanes is_vehicle = "Vehicle" in tag_names or "CapillaryRoad" in tag_names or "Freeway" in tag_names p_begin = lane.get("pointsBegin", 0) p_end = lane.get("pointsEnd", 0) lane_pts = pts[p_begin:p_end] if not lane_pts or len(lane_pts) < 2: continue lane_tangents = tangents[p_begin:p_end] if tangents else [] # Gather outgoing and adjacent links l_begin = lane.get("linksBegin", 0) l_end = lane.get("linksEnd", 0) lane_links = links[l_begin:l_end] if links else [] outgoing = [] adjacent = [] for link in lane_links: dest = link.get("destLaneIndex", -1) l_type = link.get("type", 0) if dest >= 0 and dest < len(lanes): if l_type == 1: # Outgoing outgoing.append(base_lane_idx + dest) elif l_type == 4: # Adjacent (lane change) adjacent.append(base_lane_idx + dest) # Determine speed limit: Freeway = 16.67 m/s (60km/h), Normal = 11.11 m/s (40km/h), Alley = 8.33 m/s (30km/h) if "Freeway" in tag_names: speed_limit = 16.67 elif "VehicleAlley" in tag_names: speed_limit = 8.33 else: speed_limit = 11.11 # Compute bounding box xs = [p[0] for p in lane_pts] ys = [p[1] for p in lane_pts] zs = [p[2] for p in lane_pts] min_x, max_x = min(xs), max(xs) min_y, max_y = min(ys), max(ys) min_z, max_z = min(zs), max(zs) # Compact points: [[x, y, z], ...] rounded to 2 decimals compact_pts = [[round(p[0], 2), round(p[1], 2), round(p[2], 2)] for p in lane_pts] compact_tangents = [[round(t[0], 3), round(t[1], 3), round(t[2], 3)] for t in lane_tangents] if lane_tangents else [] g_idx = global_lane_counter global_lane_counter += 1 lane_entry = { "id": g_idx, "region": region_name, "width": round(lane.get("width", 4.0), 1), "speed": speed_limit, "turn": lane.get("turnDirectionName", "Straight"), "tags": tag_names, "pts": compact_pts, "tangents": compact_tangents, "next": outgoing, "adj": adjacent, "bbox": [round(min_x, 1), round(min_y, 1), round(min_z, 1), round(max_x, 1), round(max_y, 1), round(max_z, 1)] } all_road_lanes.append(lane_entry) # Register in spatial grid cells min_cx = int(math.floor(min_x / grid_size)) max_cx = int(math.floor(max_x / grid_size)) min_cz = int(math.floor(min_z / grid_size)) max_cz = int(math.floor(max_z / grid_size)) for cx in range(min_cx, max_cx + 1): for cz in range(min_cz, max_cz + 1): ckey = f"{cx}_{cz}" if ckey not in spatial_grid: spatial_grid[ckey] = [] spatial_grid[ckey].append(g_idx) road_cache = { "metadata": { "totalLanes": len(all_road_lanes), "gridSize": grid_size, "totalGridCells": len(spatial_grid) }, "spatialGrid": spatial_grid, "lanes": all_road_lanes } road_out_path = os.path.join(OUT_DIR, "RoadNetwork.json") print(f"[ROAD] Writing {len(all_road_lanes)} road lanes to {road_out_path}...") with open(road_out_path, "w", encoding="utf-8") as out_f: json.dump(road_cache, out_f, separators=(",", ":")) print(f"[ROAD] Done! File size: {os.path.getsize(road_out_path) / 1024 / 1024:.2f} MB") # ------------------------------------------------------------- # 2. Extract Pedestrian Sidewalks & Crosswalks # ------------------------------------------------------------- ped_source = "readable_world_dump_4229938_final/03_ZONEGRAPHS/full_json/vfc_1000_e8_hfef331e26e35309d.json" print(f"[PED] Processing pedestrian graph from {ped_source}...") with z.open(ped_source) as f: data = json.load(f) ped_lanes = data.get("lanes", []) ped_pts = data.get("lanePoints", []) ped_tangents = data.get("laneTangentVectors", []) ped_links = data.get("laneLinks", []) compact_ped_lanes = [] ped_spatial_grid = {} for local_id, lane in enumerate(ped_lanes): p_begin = lane.get("pointsBegin", 0) p_end = lane.get("pointsEnd", 0) lane_pts = ped_pts[p_begin:p_end] if not lane_pts or len(lane_pts) < 2: continue tag_names = lane.get("tagNames", []) compact_pts = [[round(p[0], 2), round(p[1], 2), round(p[2], 2)] for p in lane_pts] xs = [p[0] for p in lane_pts] zs = [p[2] for p in lane_pts] min_x, max_x = min(xs), max(xs) min_z, max_z = min(zs), max(zs) ped_entry = { "id": local_id, "tags": tag_names, "pts": compact_pts, "isCrosswalk": "Crosswalk" in tag_names } compact_ped_lanes.append(ped_entry) min_cx = int(math.floor(min_x / grid_size)) max_cx = int(math.floor(max_x / grid_size)) min_cz = int(math.floor(min_z / grid_size)) max_cz = int(math.floor(max_z / grid_size)) for cx in range(min_cx, max_cx + 1): for cz in range(min_cz, max_cz + 1): ckey = f"{cx}_{cz}" if ckey not in ped_spatial_grid: ped_spatial_grid[ckey] = [] ped_spatial_grid[ckey].append(local_id) ped_cache = { "metadata": { "totalLanes": len(compact_ped_lanes), "gridSize": grid_size, "totalGridCells": len(ped_spatial_grid) }, "spatialGrid": ped_spatial_grid, "lanes": compact_ped_lanes } ped_out_path = os.path.join(OUT_DIR, "PedestrianNetwork.json") print(f"[PED] Writing {len(compact_ped_lanes)} pedestrian lanes to {ped_out_path}...") with open(ped_out_path, "w", encoding="utf-8") as out_f: json.dump(ped_cache, out_f, separators=(",", ":")) print(f"[PED] Done! File size: {os.path.getsize(ped_out_path) / 1024 / 1024:.2f} MB") # ------------------------------------------------------------- # 3. Extract Interiors & Shops # ------------------------------------------------------------- print("[INTERIORS] Processing interiors from interiors_full.csv...") interiors_list = [] interior_spatial_grid = {} with z.open("readable_world_dump_4229938_final/00_INDEX/interiors_full.csv") as f: reader = csv.reader(io.TextIOWrapper(f, encoding="utf-8-sig")) header = next(reader) for row in reader: indoor_id = row[2] name = row[3] sector = row[4] wx_s, wy_s, wz_s = row[7], row[8], row[9] if not wx_s or not wy_s or not wz_s: continue try: wx = round(float(wx_s), 2) wy = round(float(wy_s), 2) wz = round(float(wz_s), 2) except ValueError: continue idx = len(interiors_list) entry = { "id": int(indoor_id) if indoor_id.isdigit() else 0, "name": name, "sector": sector, "pos": [wx, wy, wz] } interiors_list.append(entry) cx = int(math.floor(wx / grid_size)) cz = int(math.floor(wz / grid_size)) ckey = f"{cx}_{cz}" if ckey not in interior_spatial_grid: interior_spatial_grid[ckey] = [] interior_spatial_grid[ckey].append(idx) interiors_cache = { "metadata": { "totalInteriors": len(interiors_list), "gridSize": grid_size }, "spatialGrid": interior_spatial_grid, "interiors": interiors_list } interiors_out_path = os.path.join(OUT_DIR, "Interiors.json") print(f"[INTERIORS] Writing {len(interiors_list)} interiors to {interiors_out_path}...") with open(interiors_out_path, "w", encoding="utf-8") as out_f: json.dump(interiors_cache, out_f, indent=2, ensure_ascii=False) print(f"[INTERIORS] Done! File size: {os.path.getsize(interiors_out_path) / 1024:.2f} KB") # ------------------------------------------------------------- # 4. Extract Traffic Signals & Intersections # ------------------------------------------------------------- print("[SIGNALS] Processing traffic lights and intersections...") with z.open("readable_world_dump_4229938_final/04_WORLD_JSON/full_json/vfc_47_e11_h7140c25172fd7615.json") as f: lights_raw = json.load(f).get("TrafficLights", []) with z.open("readable_world_dump_4229938_final/04_WORLD_JSON/full_json/vfc_47_e13_h64c11d788d8b4dd9.json") as f: inter_raw = json.load(f).get("interInfos", []) compact_lights = [] for l in lights_raw: pos = l.get("position", {}) fwd = l.get("forward", {}) compact_lights.append({ "handle": l.get("trafficLightHandle", 0), "inter": l.get("inter", -1), "zbr": l.get("zbr", -1), "pos": [round(pos.get("x", 0), 2), round(pos.get("y", 0), 2), round(pos.get("z", 0), 2)], "fwd": [round(fwd.get("x", 0), 3), round(fwd.get("y", 0), 3), round(fwd.get("z", 0), 3)], }) compact_inter = [] for it in inter_raw: compact_inter.append({ "id": it.get("intersection", 0), "zebras": [z.get("zebra") for z in it.get("zebraConnector", [])] }) signals_cache = { "lights": compact_lights, "intersections": compact_inter } signals_out_path = os.path.join(OUT_DIR, "TrafficSignals.json") print(f"[SIGNALS] Writing {len(compact_lights)} lights and {len(compact_inter)} intersections to {signals_out_path}...") with open(signals_out_path, "w", encoding="utf-8") as out_f: json.dump(signals_cache, out_f, indent=2, ensure_ascii=False) print(f"[SIGNALS] Done! File size: {os.path.getsize(signals_out_path) / 1024:.2f} KB") t_elapsed = time.time() - t0 print(f"[ALL COMPLETE] World and Traffic data extraction completed in {t_elapsed:.2f} seconds!")