"""NEF teaching adapter: fixed AF, external IDs, verified UE path, durable callbacks. This uses the lab observer, not Open5GS Namf/Nudm event-exposure APIs. """ import asyncio import json import os import time import uuid from contextlib import asynccontextmanager import httpx from fastapi import FastAPI, Header, HTTPException from pydantic import BaseModel, ConfigDict, Field, model_validator from backend.teaching import Store, Events, authorize, service_error ALIASES = {f"meter-ue{n}":f"ue{n}" for n in range(1,5)} class Subscription(BaseModel): model_config = ConfigDict(extra="forbid") external_id: str sender: str command_id: str = Field(pattern=r"^[0-9a-f]{32}$") @model_validator(mode="after") def known_target(self): if self.external_id not in ALIASES or self.sender not in ALIASES.values() or self.sender == ALIASES[self.external_id]: raise ValueError("Choose distinct working phones and a known external device ID") return self class Adapter: def __init__(self, store, client, backend, af): self.store,self.client,self.backend,self.af = store,client,backend,af self.events = Events(store,client,backend) async def subscribe(self, request): active = self.store.all("subscriptions") if len(active)>=16: raise HTTPException(409,"Cancel an existing subscription before adding another") sub = {"id":uuid.uuid4().hex, **request.model_dump(), "recipient":ALIASES[request.external_id], "created_at":time.time(), "available":None, "last_error":None} self.store.put("subscriptions",sub) await self.events.emit("af","nef","Availability subscription",sub, "The Metriczone application subscribes to a known device through the NEF teaching API.") await self.events.emit("nef","af","Subscription accepted",sub, "The teaching adapter maps the external device ID to its lab UE. The callback destination is fixed to the Metriczone AF.") return sub async def poll_once(self): for sub in self.store.all("subscriptions"): try: response = await self.client.get(self.backend+"/api/nef/observe",params={"sender":sub["sender"],"recipient":sub["recipient"]}, headers={"X-Teaching-Token":os.getenv("TEACHING_TOKEN","local-nef-teaching")},timeout=15) response.raise_for_status() observation = response.json() if not isinstance(observation,dict) or not isinstance(observation.get("available"),bool): raise ValueError("Observer returned invalid availability evidence") except (httpx.HTTPError,ValueError) as error: # Observer failure is unknown, never a fabricated offline/online transition. current = self.store.get("subscriptions",sub["id"]) if current: current["last_error"] = service_error(error,"Lab observer") self.store.put("subscriptions",current) continue current = self.store.get("subscriptions",sub["id"]) if not current: continue available = observation["available"] changed = current["available"] is None or current["available"] != available current.update(available=available,observation=observation,last_error=None) notification = None # Commit the transition and its outbox entry in one transaction. with self.store.connection() as db: db.execute("UPDATE items SET value=? WHERE kind='subscriptions' AND id=?",(json.dumps(current),current["id"])) if changed and available: notification = {"id":uuid.uuid4().hex,"subscription_id":sub["id"],"command_id":sub["command_id"], "recipient":sub["recipient"],"external_id":sub["external_id"],"available":True, "observed_at":time.time(),"evidence":observation,"attempts":0} db.execute("INSERT INTO items VALUES ('notifications',?,?)",(notification["id"],json.dumps(notification))) if changed: await self.events.emit("observer","nef", "Device path ready" if available else "Device unavailable", {**sub,"observation":observation}, "The lab observer checks the real UE containers and a tunnel-bound probe; this is not a native AMF/UDM exposure notification.") for notification in self.store.all("notifications"): if not self.store.get("subscriptions",notification["subscription_id"]): self.store.delete("notifications",notification["id"]) continue if notification["attempts"] == 0: await self.events.emit("nef","af","Availability notification",notification, "The NEF teaching adapter calls the Metriczone application's fixed callback after a successful bidirectional UE probe.") notification["attempts"] += 1 self.store.put("notifications",notification) try: response = await self.client.post(self.af+"/teaching/v1/notifications",json=notification, headers={"X-Teaching-Token":os.getenv("TEACHING_TOKEN","local-nef-teaching")},timeout=10) response.raise_for_status() except httpx.HTTPError: continue self.store.delete("notifications",notification["id"]) async def run(self): while True: await self.poll_once() await asyncio.sleep(1) def create_app(path=None, transport=None): store = Store(path or os.getenv("TEACHING_DB","/data/nef.sqlite3")) @asynccontextmanager async def lifespan(app): async with httpx.AsyncClient(transport=transport) as client: app.state.adapter = Adapter(store,client,os.getenv("BACKEND_URL","http://172.24.0.30:8000"),os.getenv("AF_URL","http://172.24.0.10:8000")) tasks=[asyncio.create_task(app.state.adapter.run()),asyncio.create_task(app.state.adapter.events.run())] try: yield finally: for task in tasks:task.cancel() await asyncio.gather(*tasks,return_exceptions=True) app = FastAPI(title="Metriczone NEF teaching adapter",lifespan=lifespan) @app.get("/health") async def health():return {"status":"ok","teaching":True} @app.post("/teaching/v1/subscriptions",status_code=201) async def subscribe(request:Subscription,x_af_key:str|None=Header(default=None)): authorize(x_af_key,"NEF_AF_TOKEN","local-metriczone-af") return await app.state.adapter.subscribe(request) @app.delete("/teaching/v1/subscriptions/{subscription_id}") async def cancel(subscription_id:str,x_af_key:str|None=Header(default=None)): authorize(x_af_key,"NEF_AF_TOKEN","local-metriczone-af") store.delete("subscriptions",subscription_id) return {"cancelled":True} @app.get("/teaching/v1/subscriptions") async def subscriptions(x_af_key:str|None=Header(default=None)): authorize(x_af_key,"NEF_AF_TOKEN","local-metriczone-af") return store.all("subscriptions") return app