"""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 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) @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 = 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 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)) # 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