from __future__ import annotations import json from pathlib import Path from typing import Any from ._ui_server import serve_dashboard from .platform import ( build_models_api, build_opencode_provider_config, build_registry_analytics, catalog_models, compare_models, route_or_chat_payload, route_scenario_matrix, scenario_presets, ) from .registry import load_model_registry PRODUCT_NAME = "LumynaX MaramaRoute" PACKAGE_ROOT = Path(__file__).resolve().parent PACKAGE_PARENT = PACKAGE_ROOT.parent def default_registry_path() -> Path: candidates = [ Path.cwd() / "products" / "lumynax-marama-route" / "configs" / "lumynax_model_registry.json", Path.cwd() / "configs" / "lumynax_model_registry.json", PACKAGE_ROOT / "configs" / "lumynax_model_registry.json", PACKAGE_PARENT / "configs" / "lumynax_model_registry.json", ] for path in candidates: if path.exists(): return path return candidates[0] def default_chat_request_path() -> Path: candidates = [ Path.cwd() / "products" / "lumynax-marama-route" / "examples" / "request.chat-code.json", Path.cwd() / "examples" / "request.chat-code.json", PACKAGE_ROOT / "examples" / "request.chat-code.json", PACKAGE_PARENT / "examples" / "request.chat-code.json", ] for path in candidates: if path.exists(): return path return candidates[0] def load_json_mapping(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8-sig")) if not isinstance(payload, dict): raise ValueError(f"Expected JSON object in {path}") return payload def route_dashboard_payload(payload: dict[str, Any], registry_path: Path) -> dict[str, Any]: return route_or_chat_payload(payload, load_model_registry(registry_path)) def build_dashboard_state(registry_path: Path) -> dict[str, Any]: models = load_model_registry(registry_path) example = ( load_json_mapping(default_chat_request_path()) if default_chat_request_path().exists() else _fallback_chat_example() ) analytics = build_registry_analytics(models) return { "registry_path": str(registry_path), "model_count": len(models), "resident_nz": analytics["resident_nz"], "runtimes": sorted(analytics["runtimes"]), "example": example, "models": build_models_api(models), "analytics": analytics, "catalog": catalog_models(models, {"limit": 14})["models"], "scenarios": scenario_presets(), "matrix": route_scenario_matrix(models), "opencode_config": build_opencode_provider_config(models), } def handle_api_request( method: str, path: str, payload: dict[str, Any] | None, registry_path: Path, ) -> tuple[int, dict[str, Any]]: if method == "GET" and path == "/api/health": state = build_dashboard_state(registry_path) return 200, {"ok": True, "product": PRODUCT_NAME, "model_count": state["model_count"]} if method == "GET" and path == "/api/models": return 200, build_models_api(load_model_registry(registry_path)) if method == "GET" and path == "/api/analytics": return 200, {"ok": True, **build_registry_analytics(load_model_registry(registry_path))} if method == "GET" and path == "/api/state": return 200, {"ok": True, **build_dashboard_state(registry_path)} if method == "POST" and path == "/api/route" and payload is not None: status = 200 result = route_dashboard_payload(payload, registry_path) if not result["ok"]: status = 422 return status, result if method == "POST" and path == "/api/catalog" and payload is not None: return 200, catalog_models(load_model_registry(registry_path), payload) if method == "POST" and path == "/api/compare" and payload is not None: model_ids = payload.get("model_ids") or payload.get("models") or [] if isinstance(model_ids, str): model_ids = [item.strip() for item in model_ids.split(",") if item.strip()] request_payload = payload.get("request") if isinstance(payload.get("request"), dict) else None result = compare_models(load_model_registry(registry_path), list(model_ids), request_payload) return (200 if result["ok"] else 422), result if method == "POST" and path == "/api/matrix" and payload is not None: scenarios = payload.get("scenarios") if isinstance(payload.get("scenarios"), list) else None return 200, route_scenario_matrix(load_model_registry(registry_path), scenarios) if method == "POST" and path == "/api/opencode-config" and payload is not None: return 200, { "ok": True, "config": build_opencode_provider_config( load_model_registry(registry_path), base_url=str(payload.get("base_url") or "http://127.0.0.1:8787/v1"), ), } return 404, {"ok": False, "error": "not_found"} def smoke_ui(registry_path: Path | None = None) -> dict[str, Any]: resolved_registry = registry_path or default_registry_path() state = build_dashboard_state(resolved_registry) routed = route_dashboard_payload(state["example"], resolved_registry) if not routed["ok"]: raise RuntimeError("MaramaRoute UI smoke route did not select a model") catalog = catalog_models(load_model_registry(resolved_registry), {"task_type": "code", "limit": 3}) matrix = route_scenario_matrix(load_model_registry(resolved_registry)) if not catalog["models"] or not matrix["ok"]: raise RuntimeError("MaramaRoute expanded UI smoke checks failed") return { "ok": True, "product": PRODUCT_NAME, "model_count": state["model_count"], "selected_model": routed["route_decision"]["selected_model"]["model_id"], "catalog_count": catalog["count"], "scenario_count": len(matrix["scenarios"]), } def run_ui( *, registry_path: Path | None = None, host: str = "127.0.0.1", port: int = 8787, open_browser: bool = False, smoke: bool = False, ) -> int: resolved_registry = registry_path or default_registry_path() if smoke: print(json.dumps(smoke_ui(resolved_registry), indent=2, sort_keys=True)) return 0 html = build_expanded_dashboard_html(build_dashboard_state(resolved_registry)) return serve_dashboard( product_name=PRODUCT_NAME, html=html, api_handler=lambda method, path, payload: handle_api_request( method, path, payload, resolved_registry, ), host=host, port=port, open_browser=open_browser, ) def build_expanded_dashboard_html(state: dict[str, Any]) -> str: initial = json.dumps(state, sort_keys=True).replace(" LumynaX MaramaRoute Console

LumynaX
MaramaRoute

Sovereign router control surface for LumynaX model selection.
0models
0NZ resident
0local runtime
0JSON ready
0max ctx
SELECTEDpending
MODEready
FALLBACKS0
RECEIPTnone

Route Result


        

        

          
Scenario Matrix

          

            
ScenarioModelRejected

            
          

        

      

      

        
Model Catalog

        

          
ModelRuntimeFitCaps

          
        

      

      

        
Provider Config

        

      

    

  

  
  

"""
    return html.replace("__INITIAL__", initial)


def build_dashboard_html(state: dict[str, Any]) -> str:
    initial = json.dumps(state, sort_keys=True).replace("


  
  
  LumynaX MaramaRoute
  


  

    

      

        
LumynaX
MaramaRoute

        
Sovereign model router for AbteeXAILab Hugging Face releases

      

      

        
0models

        
0NZ resident

        
0runtimes

      

    

    

      

        

          
          
          
        

        
        
      

      

        

          
SELECTEDpending

          
MODEpending

          
REJECTED0

        

        

      

    

  

  
  

"""


def _fallback_chat_example() -> dict[str, Any]:
    return {
        "model": "lumynax/code",
        "messages": [
            {
                "role": "user",
                "content": "Refactor this private Python service and return a JSON diff plan.",
            },
        ],
        "response_format": {"type": "json_object"},
        "route": {
            "jurisdiction": "NZ",
            "data_sensitivity": "restricted",
            "task_type": "code",
            "requires_local": True,
            "requires_json": True,
            "max_fallbacks": 3,
        },
    }