"""Opt-in Runtime bridge: one existing CameraCommands owner and existing Manager lifecycle. This module intentionally does not import the Sidecar, GPU or a database. The Manager's command lock also fences legacy CLI/API writers during projection. """ import asyncio from contextlib import asynccontextmanager from copy import deepcopy from dataclasses import dataclass, field import hmac import json import logging from contextra_trace import emit as trace_emit from contextra_trace.http import REQUEST_ID import math from urllib.parse import urlsplit from uuid import uuid5, NAMESPACE_URL from uuid import uuid4 from app.multi_roi.errors import RoiError from fastapi import APIRouter, Request from fastapi.responses import JSONResponse CODES = frozenset({'FIRE','FALL','INTRUSION','LOITERING','FIGHT','CROWD'}) MAX_BODY = 65536 LOG = logging.getLogger(__name__) def command_error(error): import re value = str(error) if isinstance(error, RoiError): return error.code if isinstance(error, ControlError) and re.fullmatch(r'[A-Z][A-Z0-9_]{0,63}', value): return value return 'RUNTIME_COMMAND_REJECTED' class ControlError(ValueError): pass def decode(raw): def pairs(items): d = {} for k, v in items: if k in d: raise ControlError('DUPLICATE_KEY') d[k] = v return d def bad(_): raise ControlError('INVALID_NUMBER') if len(raw) > MAX_BODY: raise ControlError('BODY_TOO_LARGE') try: d = json.loads(raw, object_pairs_hook=pairs, parse_constant=bad) if not isinstance(d, dict): raise ControlError('BODY_NOT_OBJECT') return d except (ValueError, UnicodeError, RecursionError): raise ControlError('INVALID_BODY') from None def camera_id(d): v = d.get('cameraId') if type(v) is not int or not 1 <= v <= 2**53-1: raise ControlError('CAMERA_ID_INVALID') return v def validate_start(d): camera_id(d) required = {'cameraId','rtspUrl','startWallTimeMs','rtpClockRate','scenarios'} if not required <= set(d) or set(d)-required-{'recServerId','recCameraId'}: raise ControlError('START_FIELDS_INVALID') u = d['rtspUrl'] try: p = urlsplit(u) if (not isinstance(u, str) or len(u)>4096 or p.scheme!='rtsp' or not p.hostname or p.port == 0 or p.fragment or any(c.isspace() or ord(c)<32 for c in u)): raise ValueError() except (TypeError, ValueError): raise ControlError('RTSP_INVALID') from None if type(d['startWallTimeMs']) is not int or not 0 < d['startWallTimeMs'] <= 2**53-1: raise ControlError('ANCHOR_INVALID') if type(d['rtpClockRate']) is not int or d['rtpClockRate'] != 90000: raise ControlError('CLOCK_RATE_UNSUPPORTED') scenarios = d['scenarios'] if not isinstance(scenarios, list) or len(scenarios) != 6: raise ControlError('SCENARIO_SNAPSHOT_REQUIRED') seen = set() for s in scenarios: if (not isinstance(s, dict) or not {'scenarioCode','enabled'} <= set(s) or set(s)-{'scenarioCode','enabled','rois','parameters'}): raise ControlError('SCENARIO_FIELDS_INVALID') name = s['scenarioCode'] if not isinstance(name, str) or name not in CODES or name in seen or type(s['enabled']) is not bool: raise ControlError('SCENARIO_CODE_INVALID') seen.add(name) # parameters is intentionally fail-soft; strict Start/ROI/identity validation # stays here while scenario_parameters sanitizes individual tuning values. rois = s.get('rois') # Normalize before command canonicalization so equivalent empty ROI requests deduplicate. if rois is None: s['rois'] = rois = [] if not isinstance(rois, list) or len(rois) > 32: raise ControlError('ROI_LIST_INVALID') for roi in rois: if not isinstance(roi, dict) or set(roi) != {'label','points'}: raise ControlError('ROI_FIELDS_INVALID') label, points = roi['label'], roi['points'] if (not isinstance(label,str) or not label.strip() or len(label)>64 or any(ord(c)<32 for c in label) or not isinstance(points,list) or not 3<=len(points)<=32): raise ControlError('ROI_INVALID') for pair in points: if not isinstance(pair, list) or len(pair)!=2 or any(type(v) not in (float,int) or not math.isfinite(v) or not 0<=v<=1 for v in pair): raise ControlError('ROI_COORDINATES_INVALID') for k in ('recServerId','recCameraId'): if d.get(k) is not None and (not isinstance(d[k], str) or not d[k] or len(d[k])>256): raise ControlError('REC_ID_INVALID') # Multi-provider runtime owns coexistence. Scenario tuning values are # sanitized later against the saved/default candidate spec. @dataclass(frozen=True) class Start: # Canonical text makes duplicate in-flight requests compare by value, not object identity. text: str = field(repr=False) class ControlService: def __init__(self, manager, server_id, template=None, *, roi_policy_path=None): self.manager, self.server_id, self.template = manager, server_id, deepcopy(template) from app.multi_roi.policy import PolicyStore self.roi_policy = PolicyStore(roi_policy_path, command_lock=manager._commands, snapshots=self.roi_snapshots) def roi_snapshots(self): values = [] for group in getattr(self.manager, '_groups', {}).values(): for spec in (group.spec, getattr(group, '_pending_restart', None)): if spec and spec.get('multi_roi'): values.append(spec['multi_roi']) if hasattr(self.manager, 'effective_specs'): values.extend(s['multi_roi'] for s in self.manager.effective_specs() if s.get('multi_roi')) return values def validate_request(self, value): policy = self.roi_policy.value if policy['mode'] == 'multi_roi': from app.multi_roi.adapter import normalize normalize(value, policy) value['rois'].sort(key=lambda r: r['roiId']) value['rules'].sort(key=lambda r: r['ruleId']) else: if 'contract' in value: raise RoiError('CONTRACT_MODE_MISMATCH') validate_start(value) def start_camera(self, request): d = decode(request.text) self.validate_request(d) m = self.manager with m._commands: if self.roi_policy.value['mode'] == 'multi_roi': from app.multi_roi.projection import start return start(self, d) if 'contract' in d: raise RoiError('CONTRACT_MODE_MISMATCH') try: sid = m.camera_stream_id(str(d['cameraId'])) g = m.get(sid) except KeyError: g = None sid = 'camera-' + uuid5(NAMESPACE_URL, str(d['cameraId'])).hex if g is None: if self.template is None: raise ControlError('CAMERA_NOT_PROVISIONED') spec = deepcopy(self.template) spec.update(stream_id=sid, source_id=str(d['cameraId']), server_id=self.server_id, camera_id=d['cameraId'], source_type='rtsp', autostart=False) else: if g.spec.get('server_id') != self.server_id: raise ControlError('SERVER_ID_CONFLICT') # Never revive a Stop still cleaning up. Restart already owned by Manager is different. if g.observed_state == 'stopping' and g._pending_restart is None: raise ControlError('STOP_CLEANUP_PENDING') spec = deepcopy(m._saved_spec(g)) if spec.get('source_type', 'rtsp') != 'rtsp': raise ControlError('SOURCE_TYPE_CONFLICT') if spec.get('camera_id', d['cameraId']) != d['cameraId']: raise ControlError('CAMERA_ID_TYPE_CONFLICT') spec.pop('multi_roi', None) spec['uri'] = d['rtspUrl'] spec['camera_id'] = d['cameraId'] old_anchor = spec.get('frame_time_anchor') or {} # Replaying the same test Anchor must not promote it to a real clock source. spec['frame_time_anchor'] = {'startWallTimeMs': d['startWallTimeMs'], 'synthetic': bool(old_anchor.get('synthetic', False)) if old_anchor.get('startWallTimeMs') == d['startWallTimeMs'] else False} spec['tracking'] = dict(spec.get('tracking') or {}, enabled=True) from app.scenario_parameters import apply as apply_scenario_parameters, legacy_projection_view for public_request in d['scenarios']: apply_scenario_parameters(spec, public_request) projected_scenarios = legacy_projection_view(spec, d['scenarios']) intrusion = next(s for s in projected_scenarios if s['scenarioCode']=='INTRUSION') cfg = dict(spec.get('intrusion') or {}) cfg.update(enabled=intrusion['enabled'], confidence=intrusion['confidence']) roi = intrusion['rois'][0] if intrusion['rois'] else None cfg['roi'] = roi['points'] if roi else [[0.,0.],[1.,0.],[1.,1.],[0.,1.]] if roi: cfg['roi_name'] = roi['label'] else: cfg.pop('roi_name', None) spec['intrusion'] = cfg from app.region_scenario import project_loitering, validate as validate_region from app.fire_config import project_fire from app.fall_runtime import project_fall from app.fight_runtime import project_fight from app.crowd_runtime import project_crowd_candidate spec['scenarios'] = 'off' try: project_loitering(spec, projected_scenarios) project_fall(spec, projected_scenarios) project_fight(spec, projected_scenarios) project_fire(spec, projected_scenarios) spec = project_crowd_candidate(spec, projected_scenarios) validate_region(spec) except ValueError as exc: if str(exc) == 'NEEDS_MULTI_PROVIDER_RUNTIME_CONFIRMATION' or str(exc).startswith('CROWD_'): raise ControlError(str(exc)) from None raise ControlError('LOITERING_CONFIGURATION_UNSUPPORTED') from exc # Non-secret provenance: unsupported requests are explicit skips, not silent success. spec['secure_control_projection'] = { 'profile': 'REST_STATELESS', 'skippedScenarios': [s['scenarioCode'] for s in d['scenarios'] if s['enabled'] and s['scenarioCode'] not in ('INTRUSION', 'LOITERING', 'FIRE', 'FALL', 'FIGHT', 'CROWD')], 'ignoredIntrusionRois': max(0, len(intrusion['rois'])-1), 'requestedRtpClockRate': d['rtpClockRate'], } if spec.get('crowd_start_rejection'): spec['secure_control_projection']['skippedScenarios'].append('CROWD') spec['secure_control_projection']['crowdSkipReason'] = spec['crowd_start_rejection'] # Inference/model/source options and local storage knobs remain the saved values. spec['scenarios'] = 'off' for key in ('recServerId','recCameraId'): if key in d: spec['rec_server_id' if key=='recServerId' else 'rec_camera_id'] = d[key] return m.apply_and_start(spec) def stop_camera(self, camera): with self.manager._commands: try: sid = self.manager.camera_stream_id(str(camera)) except KeyError: return None group = self.manager.get(sid) if group.spec.get('server_id') != self.server_id: raise ControlError('SERVER_ID_CONFLICT') return self.manager.stop(sid) def authorized(request, token): values = request.headers.getlist('authorization') return (len(values) == 1 and hmac.compare_digest(values[0].encode('utf-8'), ('Bearer '+token).encode('utf-8'))) def build_router(manager, token, server_id, template=None, *, roi_policy_path=None): from app.application.camera_commands import CameraCommands, CommandUnavailable service = ControlService(manager, server_id, template, roi_policy_path=roi_policy_path) owner = CameraCommands(service, capacity=16, history=64) def completed(job, future): error = future.exception() LOG.info('secure_command operation=%s camera=%s action=%s outcome=%s error=%s', job.operation_id, job.camera_id, job.action, 'FAILED' if error else 'COMMAND_APPLIED', command_error(error) if error else None) try: group = None if error else future.result() lifecycle = getattr(group, 'lifecycle', None) snapshot = lifecycle.snapshot() if lifecycle else {} trace_emit(direction='WEB_TO_AI', component='runtime', phase='COMMAND_APPLIED', requestId=getattr(job, '_trace_request', None), operationId=job.operation_id, cameraId=str(job.camera_id), action=job.action, runId=snapshot.get('run_id'), observedState=snapshot.get('observed_state', 'STOPPED' if group is None and not error else 'UNKNOWN'), error=command_error(error) if error else None) except Exception: pass @asynccontextmanager async def lifespan(app): try: yield finally: if not await asyncio.to_thread(owner.close, 2): raise RuntimeError('SECURE_COMMAND_CLEANUP_PENDING') router = APIRouter(lifespan=lifespan) from app.multi_roi.routes import policy_router router.include_router(policy_router(service.roi_policy, token)) @router.get('/internal/secure-control/training/settings') def training_settings(request: Request): if not authorized(request, token): return JSONResponse({'error':'CONTROL_UNAUTHORIZED'},status_code=401) training = getattr(request.app.state, 'training', None) if training is None: return JSONResponse({'error':'TRAINING_NOT_CONFIGURED'},status_code=503) return dict(**training.store.settings(), status=training.status()) @router.patch('/internal/secure-control/training/settings') async def patch_training_settings(request: Request): if not authorized(request, token): return JSONResponse({'error':'CONTROL_UNAUTHORIZED'},status_code=401) training = getattr(request.app.state, 'training', None) if training is None: return JSONResponse({'error':'TRAINING_NOT_CONFIGURED'},status_code=503) try: raw = bytearray() async with asyncio.timeout(1): async for chunk in request.stream(): raw.extend(chunk) if len(raw)>4096: return JSONResponse({'error':'BODY_TOO_LARGE'},status_code=413) value = decode(bytes(raw)) if set(value) != {'expected_revision','changes'}: raise ValueError('Invalid training patch.') return await asyncio.to_thread(training.store.patch,value['expected_revision'],value['changes']) except RuntimeError: return JSONResponse({'error':'REVISION_CONFLICT'},status_code=409) except (ValueError, TypeError, TimeoutError): return JSONResponse({'error':'INVALID_TRAINING_SETTINGS'},status_code=422) except OSError: return JSONResponse({'error':'TRAINING_PERSISTENCE_FAILED'},status_code=503) @router.get('/internal/secure-control/runtime-status') async def runtime_status(request: Request): if not authorized(request, token): return JSONResponse({'error': 'CONTROL_UNAUTHORIZED'}, status_code=401) # Parent-only lifecycle snapshots; never query a child collector over IPC. with manager._lock: groups = list(manager._groups.values()) streams = [] counts = dict(running=0, starting=0, stopped=0, error=0, total=len(groups)) for group in groups: s = group.lifecycle.snapshot() observed = s['observed_state'].upper() error = None if observed not in ('RUNNING', 'STARTING', 'STOPPED', 'ERROR'): error, observed = 'TRANSITION_' + observed, 'ERROR' elif s.get('last_error'): error = 'RUNTIME_REPORTED_ERROR' counts[observed.lower()] += 1 streams.append({'cameraId': str(group.spec.get('source_id') or group.stream_id), 'streamId': group.stream_id, 'runId': s['run_id'], 'runGeneration': s['run_generation'], 'desiredState': s['desired_state'].upper(), 'observedState': observed, 'lastError': error}) return {'serverId': server_id, 'status': 'READY', 'modelName': None, 'dependencies': {'runtime': {'ok': True, 'error': None}}, 'streamCounts': counts, 'streams': streams} async def command(request, action): request_id = REQUEST_ID.get() or uuid4().hex if not authorized(request, token): return JSONResponse({'error':'CONTROL_UNAUTHORIZED'},status_code=401) raw=bytearray() try: async with asyncio.timeout(.25): async for chunk in request.stream(): if len(raw)+len(chunk)>MAX_BODY: return JSONResponse({'error':'BODY_TOO_LARGE'},status_code=413) raw.extend(chunk) except TimeoutError: return JSONResponse({'error':'BODY_TIMEOUT'},status_code=408) try: d = decode(raw) cam = camera_id(d) if action == 'start': service.validate_request(d) value = Start(json.dumps(d,sort_keys=True,separators=(',',':'),allow_nan=False)) else: if set(d) != {'cameraId'}: raise ControlError('STOP_FIELDS_INVALID') value = None job = owner.submit(cam, value, action) # Register once even when an in-flight duplicate reuses the same operation. if not getattr(job, '_secure_observed', False): job._secure_observed = True job._trace_request = request_id LOG.info('secure_command request=%s operation=%s camera=%s action=%s outcome=ACCEPTED', request_id, job.operation_id, cam, action) job.future.add_done_callback(lambda done: completed(job, done)) trace_emit(direction='WEB_TO_AI', component='runtime', phase='COMMAND_ACCEPTED', requestId=request_id, operationId=job.operation_id, cameraId=str(cam), action=action) # Wait only for admission work, never for GPU/first Frame. Timeout does NOT cancel the job. try: future = asyncio.wrap_future(job.future) # The HTTP budget may end before queued work. Consume later errors without # cancelling work or losing its status in CameraCommands' bounded history. future.add_done_callback(lambda done: None if done.cancelled() else done.exception()) group = await asyncio.wait_for(asyncio.shield(future), .35) if group is None: state = 'STOPPED' else: state = group.observed_state.upper() except asyncio.TimeoutError: state = 'COMMAND_PENDING' except RoiError as exc: return JSONResponse(exc.payload(request_id), status_code=409) except ControlError as exc: return JSONResponse({'error':command_error(exc)},status_code=409) except Exception: if d.get('contract'): return JSONResponse(dict(RoiError('START_FAILED', camera=cam, retained=None).payload(request_id), observedState='UNKNOWN', phase='START'), status_code=409) return JSONResponse({'error':'RUNTIME_COMMAND_REJECTED'},status_code=409) skipped = [] response = {'result':'COMMAND_ACCEPTED','cameraId':cam,'operationId':job.operation_id, 'observedState':state,'skippedScenarios':skipped} if d.get('contract'): if state == 'ERROR': return JSONResponse(dict(RoiError('START_FAILED', camera=cam, retained=False).payload(request_id), observedState=state, phase='START'), status_code=409) response.update(requestId=request_id, message='명령을 접수했습니다. 실제 실행 상태를 확인하세요.') if job.future.done() and not job.future.exception(): response['change'] = job.future.result().summary return JSONResponse(response,status_code=202) except RoiError as exc: LOG.info('roi_command request=%s outcome=REJECTED code=%s details=%s', request_id, exc.code, exc.details) return JSONResponse(exc.payload(request_id), status_code=422) except ControlError as exc: if service.roi_policy.value['mode'] == 'multi_roi': return JSONResponse(dict(RoiError('INVALID_SNAPSHOT').payload(request_id), reason=command_error(exc)),status_code=422) return JSONResponse({'error':str(exc)},status_code=422) except CommandUnavailable: if service.roi_policy.value['mode'] == 'multi_roi': return JSONResponse(RoiError('CAMERA_COMMAND_BUSY', retained=None).payload(request_id),status_code=503) return JSONResponse({'error':'CAMERA_COMMAND_BUSY'},status_code=503) @router.post('/internal/secure-control/start') async def start(request:Request): return await command(request,'start') @router.post('/internal/secure-control/stop') async def stop(request:Request): return await command(request,'stop') @router.get('/internal/secure-control/operations/{identifier}') async def status(identifier:str,request:Request): if not authorized(request, token): return JSONResponse({'error':'CONTROL_UNAUTHORIZED'},status_code=401) job=owner.get(identifier) if job is None: return JSONResponse({'error':'NOT_FOUND'},status_code=404) state='PENDING' error=None if job.future.done(): state='FAILED' if job.future.exception() else 'COMMAND_APPLIED' if job.future.exception(): error=command_error(job.future.exception()) result = {'operationId':identifier,'cameraId':job.camera_id,'status':state,'error':error} if job.future.done(): exc = job.future.exception() if isinstance(exc, RoiError): result['diagnostic'] = exc.payload(identifier) elif exc is None and hasattr(job.future.result(), 'summary'): result['change'] = job.future.result().summary return result return router