""" BIM Viewer Server with APS API proxy endpoints. Serves static files + /api/token, /api/metadata, /api/properties, /api/tree, /api/thumbnail """ import json import os import re from http.server import HTTPServer, SimpleHTTPRequestHandler from urllib.parse import urlparse, parse_qs import requests from dotenv import load_dotenv load_dotenv() APS_CLIENT_ID = os.getenv("APS_CLIENT_ID") APS_CLIENT_SECRET = os.getenv("APS_CLIENT_SECRET") BASE_URL = "https://developer.api.autodesk.com" # Cache token server-side _token_cache = {"token": None, "expires_at": 0} def get_server_token(): import time if _token_cache["token"] and _token_cache["expires_at"] > time.time(): return _token_cache["token"] resp = requests.post( f"{BASE_URL}/authentication/v2/token", data={"grant_type": "client_credentials", "scope": "data:read viewables:read"}, auth=(APS_CLIENT_ID, APS_CLIENT_SECRET), ) resp.raise_for_status() data = resp.json() _token_cache["token"] = data["access_token"] _token_cache["expires_at"] = time.time() + data["expires_in"] - 60 return _token_cache["token"] class BIMHandler(SimpleHTTPRequestHandler): def do_GET(self): parsed = urlparse(self.path) path = parsed.path params = parse_qs(parsed.query) routes = { "/api/token": self.handle_token, "/api/metadata": self.handle_metadata, "/api/properties": self.handle_properties, "/api/tree": self.handle_tree, "/api/thumbnail": self.handle_thumbnail, "/api/manifest": self.handle_manifest, "/api/phases": self.handle_phases, } handler = routes.get(path) if handler: handler(params) else: super().do_GET() def send_json(self, data, status=200): self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode()) def send_error_json(self, msg, status=500): self.send_json({"error": msg}, status) def handle_token(self, params): try: token = get_server_token() self.send_json({"access_token": token, "expires_in": 3600}) except Exception as e: self.send_error_json(str(e)) def handle_metadata(self, params): """GET /api/metadata?urn=xxx - List views/GUIDs in model""" urn = params.get("urn", [None])[0] if not urn: return self.send_error_json("Missing urn parameter", 400) try: token = get_server_token() resp = requests.get( f"{BASE_URL}/modelderivative/v2/designdata/{urn}/metadata", headers={"Authorization": f"Bearer {token}"}, ) resp.raise_for_status() self.send_json(resp.json()) except Exception as e: self.send_error_json(str(e)) def handle_properties(self, params): """GET /api/properties?urn=xxx&guid=yyy - Get all element properties""" urn = params.get("urn", [None])[0] guid = params.get("guid", [None])[0] if not urn or not guid: return self.send_error_json("Missing urn or guid", 400) try: token = get_server_token() resp = requests.get( f"{BASE_URL}/modelderivative/v2/designdata/{urn}/metadata/{guid}/properties", headers={"Authorization": f"Bearer {token}"}, params={"forceget": "true"}, ) resp.raise_for_status() self.send_json(resp.json()) except Exception as e: self.send_error_json(str(e)) def handle_tree(self, params): """GET /api/tree?urn=xxx&guid=yyy - Get object tree hierarchy""" urn = params.get("urn", [None])[0] guid = params.get("guid", [None])[0] if not urn or not guid: return self.send_error_json("Missing urn or guid", 400) try: token = get_server_token() resp = requests.get( f"{BASE_URL}/modelderivative/v2/designdata/{urn}/metadata/{guid}", headers={"Authorization": f"Bearer {token}"}, params={"forceget": "true"}, ) resp.raise_for_status() self.send_json(resp.json()) except Exception as e: self.send_error_json(str(e)) def handle_thumbnail(self, params): """GET /api/thumbnail?urn=xxx - Get model thumbnail image""" urn = params.get("urn", [None])[0] if not urn: return self.send_error_json("Missing urn", 400) try: token = get_server_token() resp = requests.get( f"{BASE_URL}/modelderivative/v2/designdata/{urn}/thumbnail", headers={"Authorization": f"Bearer {token}"}, params={"width": 400, "height": 400}, ) resp.raise_for_status() self.send_response(200) self.send_header("Content-Type", "image/png") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(resp.content) except Exception as e: self.send_error_json(str(e)) def handle_manifest(self, params): """GET /api/manifest?urn=xxx - Get translation manifest""" urn = params.get("urn", [None])[0] if not urn: return self.send_error_json("Missing urn", 400) try: token = get_server_token() resp = requests.get( f"{BASE_URL}/modelderivative/v2/designdata/{urn}/manifest", headers={"Authorization": f"Bearer {token}"}, ) resp.raise_for_status() self.send_json(resp.json()) except Exception as e: self.send_error_json(str(e)) def handle_phases(self, params): """GET /api/phases?urn=xxx&guid=yyy - Extract phase data per element""" urn = params.get("urn", [None])[0] guid = params.get("guid", [None])[0] if not urn or not guid: return self.send_error_json("Missing urn or guid", 400) try: token = get_server_token() resp = requests.get( f"{BASE_URL}/modelderivative/v2/designdata/{urn}/metadata/{guid}/properties", headers={"Authorization": f"Bearer {token}"}, params={"forceget": "true"}, ) resp.raise_for_status() data = resp.json() collection = data.get("data", {}).get("collection", []) phases = {} # phase_name -> list of dbIds elements = {} # dbId -> {name, phaseCreated, phaseDemolished, category} phase_names = set() for elem in collection: db_id = elem.get("objectid") elem_name = elem.get("name", "") props = elem.get("properties", {}) phase_created = None phase_demolished = None category = "" # Extract Phasing info phasing = props.get("Phasing", {}) if isinstance(phasing, dict): phase_created = phasing.get("Phase Created") phase_demolished = phasing.get("Phase Demolished") # Extract category from __category__ cat_data = props.get("__category__", {}) if isinstance(cat_data, dict): category = cat_data.get("__category__", "") if phase_created: phase_names.add(phase_created) if phase_created not in phases: phases[phase_created] = [] phases[phase_created].append(db_id) elements[str(db_id)] = { "name": elem_name, "phaseCreated": phase_created, "phaseDemolished": phase_demolished, "category": category, } sorted_phases = sorted(phase_names) self.send_json({ "phases": sorted_phases, "phaseElements": phases, "elements": elements, "totalElements": len(collection), "phasedElements": len(elements), }) except Exception as e: self.send_error_json(str(e)) if __name__ == "__main__": port = 9090 server = HTTPServer(("0.0.0.0", port), BIMHandler) print(f"BIM Viewer server running at http://localhost:{port}") print("API endpoints: /api/token, /api/metadata, /api/properties, /api/tree, /api/thumbnail, /api/manifest") print("Press Ctrl+C to stop") server.serve_forever()