From caca00bfdf79895dfeb983761a1de66deb9e396f Mon Sep 17 00:00:00 2001 From: Yiren Zhang Date: Thu, 2 Jul 2026 10:17:45 +0200 Subject: [PATCH 1/2] feat: add Semantic Adaptive Surveillance System --- .gitignore | 5 + SASS/config/base.pkl | 37 +++ SASS/machines/cloudViewer.pkl | 54 ++++ SASS/machines/edgeCamera.pkl | 347 ++++++++++++++++++++++ SASS/machines/incidentDetector.pkl | 163 ++++++++++ SASS/machines/incidentManager.pkl | 209 +++++++++++++ SASS/machines/mediaCapacityArbitrator.pkl | 126 ++++++++ SASS/main.pkl | 168 +++++++++++ SASS/policies/correlationRules.pkl | 7 + SASS/policies/escalationPolicy.pkl | 8 + SASS/policies/priorityPolicy.pkl | 6 + SASS/services/caption_service.py | 259 ++++++++++++++++ SASS/services/requirements.txt | 7 + SASS/services/run_all.sh | 23 ++ SASS/services/support_services.py | 201 +++++++++++++ SASS/topology/deployment.pkl | 9 + 16 files changed, 1629 insertions(+) create mode 100644 .gitignore create mode 100644 SASS/config/base.pkl create mode 100644 SASS/machines/cloudViewer.pkl create mode 100644 SASS/machines/edgeCamera.pkl create mode 100644 SASS/machines/incidentDetector.pkl create mode 100644 SASS/machines/incidentManager.pkl create mode 100644 SASS/machines/mediaCapacityArbitrator.pkl create mode 100644 SASS/main.pkl create mode 100644 SASS/policies/correlationRules.pkl create mode 100644 SASS/policies/escalationPolicy.pkl create mode 100644 SASS/policies/priorityPolicy.pkl create mode 100644 SASS/services/caption_service.py create mode 100644 SASS/services/requirements.txt create mode 100644 SASS/services/run_all.sh create mode 100644 SASS/services/support_services.py create mode 100644 SASS/topology/deployment.pkl diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..114793c --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.DS_Store +**/.DS_Store +.idea +.vscode +__pycache__ \ No newline at end of file diff --git a/SASS/config/base.pkl b/SASS/config/base.pkl new file mode 100644 index 0000000..96029b0 --- /dev/null +++ b/SASS/config/base.pkl @@ -0,0 +1,37 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +// Tunable input parameters of the SaSS. +params = new Mapping { + // --- Detection (Incident Detector) --- + ["anomalyGateThreshold"] = "0.001" // per-camera anomaly score gate (0..1) + ["detectionThreshold"] = "0.001" // incident confidence threshold (crossing => detected) + ["correlationWindowMs"] = "5000" // within-zone correlation window + ["correlationK"] = "1" // k in k-of-n (min cameras of the same zone) + + // --- Incident lifecycle (Incident Manager) --- + ["reassessIntervalMs"] = "5000" // X: re-assess a detected incident every X ms + + // --- Capacity (Media Capacity Arbitrator) --- + ["maxConcurrentVideo"] = "4" // global cap on concurrent video escalations + + // --- Semantic heartbeat (continuous, low-bandwidth observation cadence) --- + // Captions are emitted periodically in EVERY mode (this is the heartbeat). + ["captionIntervalMs"] = "1200" // one observation per camera per ~1.2s + + // --- Escalation as a FRAME BURST (NOT a continuous stream) --- + // On entering a video mode the camera requests ONE burst of frames and then + // waits; it does not stream. A burst is defined by (count, step): + // count = how many frames to fetch + // step = stride between fetched frames (in source frames) + // The media gateway extracts exactly these frames from CHAD, time-aligned to + // the camera's current cursor (shared via the caption service). + ["snapshotBurstCount"] = "1" + ["snapshotBurstStep"] = "1" + ["videoLowBurstCount"] = "5" // sparse evidence + ["videoLowBurstStep"] = "8" + ["videoHighBurstCount"] = "20" // dense evidence + ["videoHighBurstStep"] = "2" + + // Cooldown after a burst is released/preempted, before returning to caption. + ["cooldownMs"] = "2000" +} diff --git a/SASS/machines/cloudViewer.pkl b/SASS/machines/cloudViewer.pkl new file mode 100644 index 0000000..3db920b --- /dev/null +++ b/SASS/machines/cloudViewer.pkl @@ -0,0 +1,54 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +// --------------------------------------------------------------------------- +// cloudViewer +// Single shared, PASSIVE instance. It subscribes to two kinds of events and +// persists them via the external incident store. It never participates in +// real-time control: it only receives information and stores it for auditing, +// monitoring, and offline review. Subscription delivers information to clients; +// it does not give them control over the system. +// --------------------------------------------------------------------------- + +sm = new csml.StateMachine { + states { + ["listening"] = new csml.Initial { + on { + // ---- incident lifecycle updates ---- + ["incidentNotification"] = new csml.Transition { + yields { + new csml.Invoke { + type = "incidentStoreService" + input { + ["incidentId"] = "$incidentId" + ["siteId"] = "$siteId" + ["zoneId"] = "$zoneId" + ["cameraIds"] = "$cameraIds" + ["tags"] = "$tags" + ["confidence"] = "$confidence" + ["status"] = "$status" + } + } + new csml.Ctr { counter = "operatorViewer.incidentStored" } + } + } + // ---- frame-burst evidence captured on escalation ---- + ["mediaReady"] = new csml.Transition { + yields { + new csml.Invoke { + type = "incidentStoreService" + input { + ["incidentId"] = "$incidentId" + ["cameraId"] = "$cameraId" + ["zoneId"] = "$zoneId" + ["mode"] = "$mode" + ["mediaRef"] = "$mediaRef" + ["status"] = "'media'" + } + } + new csml.Ctr { counter = "operatorViewer.mediaStored" } + } + } + } + } + } +} diff --git a/SASS/machines/edgeCamera.pkl b/SASS/machines/edgeCamera.pkl new file mode 100644 index 0000000..5174d94 --- /dev/null +++ b/SASS/machines/edgeCamera.pkl @@ -0,0 +1,347 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +// --------------------------------------------------------------------------- +// edgeCamera (with Log statements for runtime correctness checking) +// --------------------------------------------------------------------------- +// One instance per camera. Two architectural roles in one state machine: +// * Camera Mode Controller -> the STATES are the camera modes +// (captionOnly, snapshot, videoLowFps, videoHighFps, cooldown). +// * Caption Generator -> a periodic caption tick (in EVERY mode) invokes +// the captioning service and emits one semantic observation (heartbeat). +// --------------------------------------------------------------------------- + +sm = new csml.StateMachine { + transient { + ["currentMode"] = "'caption'" + ["caption"] = "''" + ["tags"] = "''" + ["anomalyScore"] = "0.0" + ["mediaRef"] = "''" + ["incidentId"] = "''" + } + states { + + // ----- CaptionOnly (default) ----- + ["captionOnly"] = new csml.Initial { + entry { + new csml.Eval { expression = "currentMode = 'caption'" } + new csml.Log { message = "'[' + cameraId + '] ENTER captionOnly (mode=caption)'" } + } + after { + ["captionTick"] = new csml.Timeout { + delay = "captionIntervalMs" + triggers = new csml.Emit { event = new csml.Internal { topic = "captionTick" } } + } + } + on { + ["captionTick"] = new csml.Transition { + yields { + new csml.Log { message = "'[' + cameraId + '] captionTick fired (heartbeat)'" } + new csml.Invoke { + type = "captionService" + input { ["cameraId"] = "cameraId"; ["zoneId"] = "zoneId"; ["mode"] = "currentMode" } + output { "caption"; "tags"; "anomalyScore" } + } + new csml.Log { message = "'[' + cameraId + '] observation: tags=' + tags + ' score=' + anomalyScore + ' caption=' + caption" } + new csml.Emit { + event { + topic = "observation" + data { + ["siteId"] = "siteId" + ["zoneId"] = "zoneId" + ["cameraId"] = "cameraId" + ["caption"] = "caption" + ["tags"] = "tags" + ["anomalyScore"] = "anomalyScore" + ["mode"] = "currentMode" + } + } + } + new csml.Ctr { counter = "traffic.observation.events" } + } + } + ["setSnapshot"] = new csml.Transition { + to = "snapshotMode" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setSnapshot received -> snapshotMode'" } + } + } + ["setVideoLow"] = new csml.Transition { + to = "videoLowFps" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setVideoLow received -> videoLowFps'" } + } + } + ["setVideoHigh"] = new csml.Transition { + to = "videoHighFps" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setVideoHigh received -> videoHighFps'" } + } + } + } + } + + // ----- Snapshot: fetch a 1-frame burst on entry, keep captioning ----- + ["snapshotMode"] = new csml.State { + entry { + new csml.Eval { expression = "currentMode = 'snapshot'" } + new csml.Log { message = "'[' + cameraId + '] ENTER snapshotMode: requesting burst count=' + snapshotBurstCount + ' step=' + snapshotBurstStep" } + new csml.Invoke { + type = "mediaBurstService" + input { + ["cameraId"] = "cameraId" + ["zoneId"] = "zoneId" + ["incidentId"] = "incidentId" + ["mode"] = "'snapshot'" + ["count"] = "snapshotBurstCount" + ["step"] = "snapshotBurstStep" + } + output { "mediaRef" } + emits { + new csml.ConditionalEvent { + event { topic = "mediaReady" } + } + } + } + new csml.Ctr { counter = "traffic.snapshot.bursts" } + } + after { + ["captionTick"] = new csml.Timeout { + delay = "captionIntervalMs" + triggers = new csml.Emit { event = new csml.Internal { topic = "captionTick" } } + } + } + on { + ["captionTick"] = new csml.Transition { + yields { + new csml.Log { message = "'[' + cameraId + '] captionTick fired (mode=snapshot)'" } + new csml.Invoke { + type = "captionService" + input { ["cameraId"] = "cameraId"; ["zoneId"] = "zoneId"; ["mode"] = "currentMode" } + output { "caption"; "tags"; "anomalyScore" } + } + new csml.Emit { + event { + topic = "observation" + data { + ["siteId"] = "siteId" + ["zoneId"] = "zoneId" + ["cameraId"] = "cameraId" + ["caption"] = "caption" + ["tags"] = "tags" + ["anomalyScore"] = "anomalyScore" + ["mode"] = "currentMode" + } + } + } + } + } + ["setVideoLow"] = new csml.Transition { + to = "videoLowFps" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setVideoLow -> videoLowFps'" } + } + } + ["setVideoHigh"] = new csml.Transition { + to = "videoHighFps" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setVideoHigh -> videoHighFps'" } + } + } + ["releaseToCaption"] = new csml.Transition { + to = "cooldown" + yields { new csml.Log { message = "'[' + cameraId + '] releaseToCaption -> cooldown'" } } + } + } + } + + // ----- VideoLowFps: fetch a 5-frame sparse burst on entry ----- + ["videoLowFps"] = new csml.State { + entry { + new csml.Eval { expression = "currentMode = 'videoLow'" } + new csml.Log { message = "'[' + cameraId + '] ENTER videoLowFps: requesting burst count=' + videoLowBurstCount + ' step=' + videoLowBurstStep" } + new csml.Invoke { + type = "mediaBurstService" + input { + ["cameraId"] = "cameraId" + ["zoneId"] = "zoneId" + ["incidentId"] = "incidentId" + ["mode"] = "'videoLow'" + ["count"] = "videoLowBurstCount" + ["step"] = "videoLowBurstStep" + } + output { "mediaRef" } + emits { + new csml.ConditionalEvent { + event { topic = "mediaReady" } + } + } + } + new csml.Ctr { counter = "traffic.videoLow.bursts" } + } + after { + ["captionTick"] = new csml.Timeout { + delay = "captionIntervalMs" + triggers = new csml.Emit { event = new csml.Internal { topic = "captionTick" } } + } + } + on { + ["captionTick"] = new csml.Transition { + yields { + new csml.Log { message = "'[' + cameraId + '] captionTick fired (mode=videoLow)'" } + new csml.Invoke { + type = "captionService" + input { ["cameraId"] = "cameraId"; ["zoneId"] = "zoneId"; ["mode"] = "currentMode" } + output { "caption"; "tags"; "anomalyScore" } + } + new csml.Emit { + event { + topic = "observation" + data { + ["siteId"] = "siteId" + ["zoneId"] = "zoneId" + ["cameraId"] = "cameraId" + ["caption"] = "caption" + ["tags"] = "tags" + ["anomalyScore"] = "anomalyScore" + ["mode"] = "currentMode" + } + } + } + } + } + ["setVideoHigh"] = new csml.Transition { + to = "videoHighFps" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setVideoHigh -> videoHighFps'" } + } + } + ["setSnapshot"] = new csml.Transition { + to = "snapshotMode" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setSnapshot -> snapshotMode'" } + } + } + ["releaseToCaption"] = new csml.Transition { + to = "cooldown" + yields { new csml.Log { message = "'[' + cameraId + '] releaseToCaption -> cooldown'" } } + } + ["preemptToCooldown"] = new csml.Transition { + to = "cooldown" + yields { new csml.Log { message = "'[' + cameraId + '] PREEMPTED -> cooldown'" } } + } + } + } + + // ----- VideoHighFps: fetch a 20-frame dense burst on entry ----- + ["videoHighFps"] = new csml.State { + entry { + new csml.Eval { expression = "currentMode = 'videoHigh'" } + new csml.Log { message = "'[' + cameraId + '] ENTER videoHighFps: requesting burst count=' + videoHighBurstCount + ' step=' + videoHighBurstStep" } + new csml.Invoke { + type = "mediaBurstService" + input { + ["cameraId"] = "cameraId" + ["zoneId"] = "zoneId" + ["incidentId"] = "incidentId" + ["mode"] = "'videoHigh'" + ["count"] = "videoHighBurstCount" + ["step"] = "videoHighBurstStep" + } + output { "mediaRef" } + emits { + new csml.ConditionalEvent { + event { topic = "mediaReady" } + } + } + } + new csml.Ctr { counter = "traffic.videoHigh.bursts" } + } + after { + ["captionTick"] = new csml.Timeout { + delay = "captionIntervalMs" + triggers = new csml.Emit { event = new csml.Internal { topic = "captionTick" } } + } + } + on { + ["captionTick"] = new csml.Transition { + yields { + new csml.Log { message = "'[' + cameraId + '] captionTick fired (mode=videoHigh)'" } + new csml.Invoke { + type = "captionService" + input { ["cameraId"] = "cameraId"; ["zoneId"] = "zoneId"; ["mode"] = "currentMode" } + output { "caption"; "tags"; "anomalyScore" } + } + new csml.Emit { + event { + topic = "observation" + data { + ["siteId"] = "siteId" + ["zoneId"] = "zoneId" + ["cameraId"] = "cameraId" + ["caption"] = "caption" + ["tags"] = "tags" + ["anomalyScore"] = "anomalyScore" + ["mode"] = "currentMode" + } + } + } + } + } + ["setVideoLow"] = new csml.Transition { + to = "videoLowFps" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setVideoLow -> videoLowFps'" } + } + } + ["setSnapshot"] = new csml.Transition { + to = "snapshotMode" + yields { + new csml.Eval { expression = "incidentId = $incidentId.toString()" } + new csml.Log { message = "'[' + cameraId + '] setSnapshot -> snapshotMode'" } + } + } + ["releaseToCaption"] = new csml.Transition { + to = "cooldown" + yields { new csml.Log { message = "'[' + cameraId + '] releaseToCaption -> cooldown'" } } + } + ["preemptToCooldown"] = new csml.Transition { + to = "cooldown" + yields { new csml.Log { message = "'[' + cameraId + '] PREEMPTED -> cooldown'" } } + } + } + } + + // ----- Cooldown ----- + ["cooldown"] = new csml.State { + entry { + new csml.Log { message = "'[' + cameraId + '] ENTER cooldown: releasing capacity'" } + new csml.Emit { + event { topic = "releaseCapacity"; data { ["cameraId"] = "cameraId"; ["zoneId"] = "zoneId" } } + target = "'arbitrator'" + } + new csml.Eval { expression = "currentMode = 'caption'" } + } + after { + ["cooldownEnd"] = new csml.Timeout { + delay = "cooldownMs" + triggers = new csml.Emit { event = new csml.Internal { topic = "cooldownEnd" } } + } + } + on { + ["cooldownEnd"] = new csml.Transition { + to = "captionOnly" + yields { new csml.Log { message = "'[' + cameraId + '] cooldownEnd -> captionOnly'" } } + } + } + } + } +} diff --git a/SASS/machines/incidentDetector.pkl b/SASS/machines/incidentDetector.pkl new file mode 100644 index 0000000..120b7ff --- /dev/null +++ b/SASS/machines/incidentDetector.pkl @@ -0,0 +1,163 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" +import "incidentManager.pkl" + +// --------------------------------------------------------------------------- +// incidentDetector +// --------------------------------------------------------------------------- +// One instance per zone. Turns the periodic observation events from that zone's +// cameras into DETECTED incidents. +// +// Three logical stages (Reviewer feedback: detection is a confidence-threshold +// crossing, not raw k-of-n): +// 1. anomaly-score gate -> keep only candidate anomalies +// 2. within-zone correlation -> group candidates in a short time window +// 3. confidence-threshold test -> incident is DETECTED the moment the +// incident confidence value crosses detectionThreshold (an input param). +// --------------------------------------------------------------------------- + +sm = new csml.StateMachine { + transient { + ["confidence"] = "0.0" // incident confidence value (count + score + agreement) + ["candidateCount"] = "0" // how many cameras contributed candidate anomalies + ["activeTags"] = "''" // semantic group being correlated (for agreement) + ["contributing"] = "''" // comma-joined contributing cameraIds + ["incidentSeq"] = "0" // local incident sequence number + } + states { + + // ----- idle: waiting for the first candidate anomaly ----- + ["idle"] = new csml.Initial { + on { + ["observation"] = new csml.Transition { + provided = "$anomalyScore >= anomalyGateThreshold" + to = "collecting" + yields { + new csml.Eval { expression = "candidateCount = 1" } + new csml.Eval { expression = "activeTags = $tags.toString()" } + new csml.Eval { expression = "contributing = $cameraId.toString()" } + // confidence seeded with this camera's anomaly score + new csml.Eval { expression = "confidence = $anomalyScore" } + new csml.Log { message = "'[detector-' + zoneId + '] GATE PASS camera=' + $cameraId.toString() + ' tags=' + activeTags + ' score=' + confidence.toString() + ' threshold=' + anomalyGateThreshold.toString()" } + new csml.Ctr { counter = "detector.candidate.gatePass" } + } + } + // observations below the gate are simply dropped (no transition) + } + } + + // ----- collecting: accumulate within the correlation window ----- + ["collecting"] = new csml.State { + after { + ["windowElapsed"] = new csml.Timeout { + delay = "correlationWindowMs" + triggers = new csml.Emit { event = new csml.Internal { topic = "windowElapsed" } } + } + } + on { + ["observation"] = new csml.Transition { + provided = "$anomalyScore >= anomalyGateThreshold" + yields { + new csml.Match { + cases { + // Semantically related candidate from a (possibly) different camera + new csml.Case { + of = "$tags.toString() == activeTags" + yields { + new csml.Eval { expression = "candidateCount = candidateCount + 1" } + new csml.Eval { expression = "contributing = contributing + ',' + $cameraId.toString()" } + // confidence = sum of anomaly scores (count*strength), scaled by + // semantic agreement (here, exact tag match contributes fully). + new csml.Eval { expression = "confidence = confidence + $anomalyScore" } + new csml.Log { message = "'[detector-' + zoneId + '] CORRELATED camera=' + $cameraId.toString() + ' tags=' + $tags.toString() + ' score=' + $anomalyScore.toString() + ' candidateCount=' + candidateCount.toString() + ' confidence=' + confidence.toString() + ' contributing=' + contributing" } + new csml.Ctr { counter = "detector.candidate.correlated" } + } + } + } + } + } + } + ["windowElapsed"] { to = "evaluate" } + } + } + + // ----- evaluate: detection decision = confidence crossing ----- + ["evaluate"] = new csml.State { + entry { + new csml.Log { message = "'[detector-' + zoneId + '] EVALUATE candidateCount=' + candidateCount.toString() + ' requiredK=' + correlationK.toString() + ' confidence=' + confidence.toString() + ' detectionThreshold=' + detectionThreshold.toString() + ' tags=' + activeTags + ' contributing=' + contributing" } + } + always { + new csml.Transition { + provided = "confidence >= detectionThreshold && candidateCount >= correlationK" + to = "detectedOutcome" + } + new csml.Transition { + provided = "confidence < detectionThreshold || candidateCount < correlationK" + to = "discardedOutcome" + } + } + } + + // ----- detectedOutcome: instantiate manager before resetting detector state ----- + ["detectedOutcome"] = new csml.State { + entry { + new csml.Eval { expression = "incidentSeq = incidentSeq + 1" } + new csml.Log { message = "'[detector-' + zoneId + '] DETECTED incidentId=incident-' + zoneId + '-' + incidentSeq.toString() + ' candidateCount=' + candidateCount.toString() + ' confidence=' + confidence.toString() + ' cameras=' + contributing + ' tags=' + activeTags" } + // Dynamically instantiate ONE Incident Manager for this incident. + // The instance name is computed at runtime: incident--. + new csml.Instantiate { + instances { + [Pair("incident-", "zoneId + '-' + incidentSeq.toString()")] = new csml.Instance { + stateMachineName = "incidentManager" + // The manager only needs to hear events addressed to it. + data { + ["incidentId"] = "'incident-' + zoneId + '-' + incidentSeq.toString()" + ["siteId"] = "siteId" + ["zoneId"] = "zoneId" + ["cameraIds"] = "contributing" + ["tags"] = "activeTags" + ["confidence"] = "confidence" + } + } + } + } + new csml.Ctr { counter = "incident.detected.events" } + // reset for the next correlation window + new csml.Eval { expression = "confidence = 0.0" } + new csml.Eval { expression = "candidateCount = 0" } + new csml.Eval { expression = "activeTags = ''" } + new csml.Eval { expression = "contributing = ''" } + } + after { + ["toIdle"] = new csml.Timeout { + delay = "1" + triggers = new csml.Emit { event = new csml.Internal { topic = "toIdle" } } + } + } + on { + ["toIdle"] { to = "idle" } + } + } + + // ----- discardedOutcome: discard before resetting detector state ----- + ["discardedOutcome"] = new csml.State { + entry { + new csml.Log { message = "'[detector-' + zoneId + '] DISCARDED candidateCount=' + candidateCount.toString() + ' requiredK=' + correlationK.toString() + ' confidence=' + confidence.toString() + ' detectionThreshold=' + detectionThreshold.toString()" } + new csml.Ctr { counter = "incident.discarded.events" } + // reset for the next correlation window + new csml.Eval { expression = "confidence = 0.0" } + new csml.Eval { expression = "candidateCount = 0" } + new csml.Eval { expression = "activeTags = ''" } + new csml.Eval { expression = "contributing = ''" } + } + after { + ["toIdle"] = new csml.Timeout { + delay = "1" + triggers = new csml.Emit { event = new csml.Internal { topic = "toIdle" } } + } + } + on { + ["toIdle"] { to = "idle" } + } + } + } +} diff --git a/SASS/machines/incidentManager.pkl b/SASS/machines/incidentManager.pkl new file mode 100644 index 0000000..cb3d5cd --- /dev/null +++ b/SASS/machines/incidentManager.pkl @@ -0,0 +1,209 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +// --------------------------------------------------------------------------- +// incidentManager +// --------------------------------------------------------------------------- +// One instance per ACTIVE incident, dynamically instantiated by the Incident +// Detector when an incident is detected. It owns the lifecycle of exactly one +// incident. +// +// Lifecycle (Reviewer feedback): candidate -> detected -> resolved. +// +// Instance data (from the Detector's Instantiate): +// incidentId, siteId, zoneId, cameraIds, tags, confidence +// --------------------------------------------------------------------------- + +sm = new csml.StateMachine { + transient { + ["reassessMisses"] = "0" // consecutive re-assessments without confirmation + ["stillHolds"] = "false" // set true whenever a confirming observation arrives + ["escalationRequested"] = "false" // request video only once per incident manager + } + states { + + // ----- detected ----- + ["detected"] = new csml.Initial { + entry { + new csml.Log { message = "'[' + incidentId + '] ENTER detected zone=' + zoneId + ' cameras=' + cameraIds + ' tags=' + tags + ' confidence=' + confidence.toString()" } + new csml.Match { + cases { + new csml.Case { + of = "escalationRequested == false" + yields { + new csml.Eval { expression = "escalationRequested = true" } + // Escalation is an automatic consequence of detection. Severity is + // derived from confidence; here we ask the arbitrator for the level. + new csml.Log { message = "'[' + incidentId + '] requestVideo -> arbitrator camera=' + cameraIds.split(',')[0] + ' confidence=' + confidence.toString()" } + new csml.Emit { + event { + topic = "requestVideo" + data { + ["incidentId"] = "incidentId" + ["zoneId"] = "zoneId" + // pick first contributing camera as the escalation target + ["cameraId"] = "cameraIds.split(',')[0]" + ["confidence"] = "confidence" + } + } + target = "'arbitrator'" + } + // Notify any subscribed clients via the Cloud Viewer with full info. + new csml.Log { message = "'[' + incidentId + '] incidentNotification status=detected -> operatorViewer'" } + new csml.Emit { + event { + topic = "incidentNotification" + data { + ["incidentId"] = "incidentId" + ["siteId"] = "siteId" + ["zoneId"] = "zoneId" + ["cameraIds"] = "cameraIds" + ["tags"] = "tags" + ["confidence"] = "confidence" + ["status"] = "'detected'" + } + } + target = "'operatorViewer'" + } + new csml.Ctr { counter = "incident.manager.detected" } + } + } + } + } + } + after { + ["reassess"] = new csml.Timeout { + delay = "reassessIntervalMs" + triggers = new csml.Emit { event = new csml.Internal { topic = "reassess" } } + } + } + on { + // A confirming observation for this incident's zone keeps it alive. + ["observation"] = new csml.Transition { + provided = "$zoneId.toString() == zoneId && $anomalyScore >= anomalyGateThreshold" + yields { + new csml.Eval { expression = "stillHolds = true" } + new csml.Log { message = "'[' + incidentId + '] confirming observation camera=' + $cameraId.toString() + ' score=' + $anomalyScore.toString() + ' threshold=' + anomalyGateThreshold.toString()" } + } + } + // The arbitrator's decision is translated into a camera-mode command. + ["grantVideoLow"] = new csml.Transition { + yields { + new csml.Log { message = "'[' + incidentId + '] grantVideoLow received -> setVideoLow camera=' + $cameraId.toString()" } + new csml.Emit { + event { topic = "setVideoLow"; data { ["cameraId"] = "$cameraId"; ["incidentId"] = "incidentId" } } + target = "$cameraId" + } + } + } + ["grantVideoHigh"] = new csml.Transition { + yields { + new csml.Log { message = "'[' + incidentId + '] grantVideoHigh received -> setVideoHigh camera=' + $cameraId.toString()" } + new csml.Emit { + event { topic = "setVideoHigh"; data { ["cameraId"] = "$cameraId"; ["incidentId"] = "incidentId" } } + target = "$cameraId" + } + } + } + ["grantSnapshot"] = new csml.Transition { + yields { + new csml.Log { message = "'[' + incidentId + '] grantSnapshot received -> setSnapshot camera=' + $cameraId.toString()" } + new csml.Emit { + event { topic = "setSnapshot"; data { ["cameraId"] = "$cameraId"; ["incidentId"] = "incidentId" } } + target = "$cameraId" + } + } + } + // Periodic re-assessment. + ["reassess"] = new csml.Transition { + to = "reassessing" + yields { + new csml.Log { message = "'[' + incidentId + '] reassess timer fired stillHolds=' + stillHolds.toString() + ' misses=' + reassessMisses.toString()" } + } + } + } + } + + // ----- reassessing: decide whether the incident still holds ----- + ["reassessing"] = new csml.State { + entry { + new csml.Log { message = "'[' + incidentId + '] ENTER reassessing stillHolds=' + stillHolds.toString() + ' misses=' + reassessMisses.toString()" } + } + always { + new csml.Transition { + provided = "stillHolds == true" + to = "reassessHold" + } + new csml.Transition { + provided = "stillHolds == false" + to = "reassessMiss" + } + } + } + + ["reassessHold"] = new csml.State { + entry { + new csml.Eval { expression = "reassessMisses = 0" } + new csml.Eval { expression = "stillHolds = false" } + new csml.Log { message = "'[' + incidentId + '] reassess HOLD -> reset misses and continue detected'" } + new csml.Ctr { counter = "incident.manager.reassess.hold" } + } + always { + new csml.Transition { to = "detected" } + } + } + + ["reassessMiss"] = new csml.State { + entry { + new csml.Eval { expression = "reassessMisses = reassessMisses + 1" } + new csml.Log { message = "'[' + incidentId + '] reassess MISS misses=' + reassessMisses.toString()" } + new csml.Ctr { counter = "incident.manager.reassess.miss" } + } + always { + new csml.Transition { + provided = "reassessMisses >= 2" + to = "resolved" + yields { + new csml.Log { message = "'[' + incidentId + '] resolve condition met misses=' + reassessMisses.toString()" } + } + } + new csml.Transition { + provided = "reassessMisses < 2" + to = "detected" + yields { + new csml.Log { message = "'[' + incidentId + '] stay detected misses=' + reassessMisses.toString()" } + } + } + } + } + + // ----- resolved (terminal) ----- + ["resolved"] = new csml.Terminal { + entry { + new csml.Log { message = "'[' + incidentId + '] ENTER resolved -> release camera=' + cameraIds.split(',')[0]" } + // Release every contributing camera back to CaptionOnly (via cooldown). + new csml.Emit { + event { topic = "releaseToCaption"; data { ["cameraId"] = "cameraIds.split(',')[0]" } } + target = "cameraIds.split(',')[0]" + } + // Archive the incident + media to the (passive) Cloud Viewer. + new csml.Log { message = "'[' + incidentId + '] incidentNotification status=resolved -> operatorViewer'" } + new csml.Emit { + event { + topic = "incidentNotification" + data { + ["incidentId"] = "incidentId" + ["siteId"] = "siteId" + ["zoneId"] = "zoneId" + ["cameraIds"] = "cameraIds" + ["tags"] = "tags" + ["confidence"] = "confidence" + ["status"] = "'resolved'" + } + } + target = "'operatorViewer'" + } + new csml.Ctr { counter = "incident.manager.resolved" } + } + } + } +} diff --git a/SASS/machines/mediaCapacityArbitrator.pkl b/SASS/machines/mediaCapacityArbitrator.pkl new file mode 100644 index 0000000..d4e3490 --- /dev/null +++ b/SASS/machines/mediaCapacityArbitrator.pkl @@ -0,0 +1,126 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +// --------------------------------------------------------------------------- +// mediaCapacityArbitrator +// --------------------------------------------------------------------------- +// Single shared instance. Enforces the global concurrent-video budget and +// arbitrates competing requestVideo events with Grant / Defer / Preempt / Deny. +// State-dependent decisions against a global allocation view => a stateful +// arbiter (the reactive heart of the system), NOT a stream operator. +// +// Persistent params used: maxConcurrentVideo +// --------------------------------------------------------------------------- + +sm = new csml.StateMachine { + transient { + ["activeVideoCount"] = "0" + ["lastVictim"] = "''" + } + states { + ["serving"] = new csml.Initial { + on { + ["requestVideo"] = new csml.Transition { + yields { + new csml.Log { message = "'[arbitrator] requestVideo incident=' + $incidentId.toString() + ' camera=' + $cameraId.toString() + ' confidence=' + $confidence.toString() + ' activeVideoCount=' + activeVideoCount.toString() + ' max=' + maxConcurrentVideo.toString()" } + new csml.Match { + cases { + // GRANT: capacity available. Choose level from confidence. + new csml.Case { + of = "activeVideoCount < maxConcurrentVideo" + yields { + new csml.Eval { expression = "activeVideoCount = activeVideoCount + 1" } + new csml.Log { message = "'[arbitrator] capacity GRANTED incident=' + $incidentId.toString() + ' activeVideoCount=' + activeVideoCount.toString()" } + new csml.Match { + cases { + new csml.Case { + of = "$confidence >= 2.5" + yields { + new csml.Log { message = "'[arbitrator] grantVideoHigh -> ' + $incidentId.toString() + ' camera=' + $cameraId.toString()" } + new csml.Emit { + event { topic = "grantVideoHigh"; data { ["cameraId"] = "$cameraId"; ["incidentId"] = "$incidentId" } } + target = "$incidentId" + } + } + } + new csml.Case { + of = "$confidence >= 1.8 && $confidence < 2.5" + yields { + new csml.Log { message = "'[arbitrator] grantVideoLow -> ' + $incidentId.toString() + ' camera=' + $cameraId.toString()" } + new csml.Emit { + event { topic = "grantVideoLow"; data { ["cameraId"] = "$cameraId"; ["incidentId"] = "$incidentId" } } + target = "$incidentId" + } + } + } + new csml.Case { + of = "$confidence < 1.8" + yields { + new csml.Log { message = "'[arbitrator] grantSnapshot -> ' + $incidentId.toString() + ' camera=' + $cameraId.toString()" } + new csml.Emit { + event { topic = "grantSnapshot"; data { ["cameraId"] = "$cameraId"; ["incidentId"] = "$incidentId" } } + target = "$incidentId" + } + } + } + } + } + new csml.Ctr { counter = "arbitrator.grant" } + } + } + // PREEMPT: full, but this request is high-confidence/critical. + new csml.Case { + of = "activeVideoCount >= maxConcurrentVideo && $confidence > 2.5" + yields { + new csml.Log { message = "'[arbitrator] PREEMPT for incident=' + $incidentId.toString() + ' victim=' + lastVictim + ' camera=' + $cameraId.toString()" } + // Preempt one existing stream (modeled abstractly): no net + // change in count, victim camera is sent to cooldown. + new csml.Emit { + event { topic = "preemptToCooldown"; data { ["cameraId"] = "lastVictim" } } + target = "lastVictim" + } + new csml.Emit { + event { + topic = "grantVideoHigh" + data { ["cameraId"] = "$cameraId"; ["incidentId"] = "$incidentId" } + } + target = "$incidentId" + } + new csml.Eval { expression = "lastVictim = $cameraId.toString()" } + new csml.Ctr { counter = "arbitrator.preempt" } + } + } + // DEFER/DENY: full and not critical. + new csml.Case { + of = "activeVideoCount >= maxConcurrentVideo && $confidence <= 2.5" + yields { + new csml.Log { message = "'[arbitrator] DEFER incident=' + $incidentId.toString() + ' confidence=' + $confidence.toString() + ' activeVideoCount=' + activeVideoCount.toString()" } + new csml.Ctr { counter = "arbitrator.defer" } + } + } + } + } + new csml.Eval { expression = "lastVictim = $cameraId.toString()" } + } + } + // Capacity is returned when a camera enters cooldown. + ["releaseCapacity"] = new csml.Transition { + yields { + new csml.Log { message = "'[arbitrator] releaseCapacity activeVideoCount=' + activeVideoCount.toString()" } + new csml.Match { + cases { + new csml.Case { + of = "activeVideoCount > 0" + yields { + new csml.Eval { expression = "activeVideoCount = activeVideoCount - 1" } + new csml.Log { message = "'[arbitrator] capacity RELEASED activeVideoCount=' + activeVideoCount.toString()" } + new csml.Ctr { counter = "arbitrator.release" } + } + } + } + } + } + } + } + } + } +} diff --git a/SASS/main.pkl b/SASS/main.pkl new file mode 100644 index 0000000..768c887 --- /dev/null +++ b/SASS/main.pkl @@ -0,0 +1,168 @@ +amends + "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +import "config/base.pkl" +import "machines/edgeCamera.pkl" +import "machines/incidentDetector.pkl" +import "machines/incidentManager.pkl" +import "machines/mediaCapacityArbitrator.pkl" +import "machines/cloudViewer.pkl" +import "policies/correlationRules.pkl" +import "policies/escalationPolicy.pkl" +import "policies/priorityPolicy.pkl" +import "topology/deployment.pkl" + +// =========================================================================== +// Semantic Adaptive Surveillance System (SaSS) - CSM realization +// +// edgeCamera : 1 state machine per camera (modes + captioning) +// incidentDetector : 1 state machine per zone (within-zone detect) +// incidentManager : 1 state machine per incident (dynamic, spawned) +// mediaCapacityArbitrator : 1 shared state machine (global capacity) +// cloudViewer : 1 shared passive state machine (storage sink) +// +// Scenario: person-related events, single site (site-1), three zones A/B/C, +// four cameras per zone. +// =========================================================================== + +collaborativeStateMachine { + stateMachines { + ["edgeCamera"] = edgeCamera.sm + ["incidentDetector"] = incidentDetector.sm + ["incidentManager"] = incidentManager.sm + ["mediaCapacityArbitrator"] = mediaCapacityArbitrator.sm + ["cloudViewer"] = cloudViewer.sm + } + persistent { + // detection / lifecycle / capacity parameters (input parameters) + ["anomalyGateThreshold"] = base.params["anomalyGateThreshold"] + ["detectionThreshold"] = base.params["detectionThreshold"] + ["correlationWindowMs"] = base.params["correlationWindowMs"] + ["correlationK"] = base.params["correlationK"] + ["reassessIntervalMs"] = base.params["reassessIntervalMs"] + ["maxConcurrentVideo"] = base.params["maxConcurrentVideo"] + // heartbeat cadence (continuous observation, every mode) + ["captionIntervalMs"] = base.params["captionIntervalMs"] + // escalation frame-burst parameters (count/step per mode) + ["snapshotBurstCount"] = base.params["snapshotBurstCount"] + ["snapshotBurstStep"] = base.params["snapshotBurstStep"] + ["videoLowBurstCount"] = base.params["videoLowBurstCount"] + ["videoLowBurstStep"] = base.params["videoLowBurstStep"] + ["videoHighBurstCount"] = base.params["videoHighBurstCount"] + ["videoHighBurstStep"] = base.params["videoHighBurstStep"] + ["cooldownMs"] = base.params["cooldownMs"] + // policies / topology (informational) + ["policy.correlation"] = correlationRules.rules["scope"] + ["policy.escalation"] = escalationPolicy.rules["lowBelow"] + ["policy.priority"] = priorityPolicy.rules["ordering"] + ["topology.site"] = deployment.topology["site"] + } +} + +instances { + // ----- cameras: 4 per zone ----- + for (i in IntSeq(0, 3)) { + ["zoneA-cam-\(i)"] { + stateMachineName = "edgeCamera" + data { + ["cameraId"] = "'zoneA-cam-\(i)'" + ["zoneId"] = "'A'" + ["siteId"] = "'site-1'" + } + } + } + for (i in IntSeq(0, 3)) { + ["zoneB-cam-\(i)"] { + stateMachineName = "edgeCamera" + data { + ["cameraId"] = "'zoneB-cam-\(i)'" + ["zoneId"] = "'B'" + ["siteId"] = "'site-1'" + } + } + } + for (i in IntSeq(0, 3)) { + ["zoneC-cam-\(i)"] { + stateMachineName = "edgeCamera" + data { + ["cameraId"] = "'zoneC-cam-\(i)'" + ["zoneId"] = "'C'" + ["siteId"] = "'site-1'" + } + } + } + + // ----- one Incident Detector per zone (named detector-) ----- + ["detector-A"] { + stateMachineName = "incidentDetector" + subscription = Regex("zoneA-cam-.*") + data { ["zoneId"] = "'A'"; ["siteId"] = "'site-1'" } + } + ["detector-B"] { + stateMachineName = "incidentDetector" + subscription = Regex("zoneB-cam-.*") + data { ["zoneId"] = "'B'"; ["siteId"] = "'site-1'" } + } + ["detector-C"] { + stateMachineName = "incidentDetector" + subscription = Regex("zoneC-cam-.*") + data { ["zoneId"] = "'C'"; ["siteId"] = "'site-1'" } + } + + // ----- shared cloud components ----- + ["arbitrator"] { + stateMachineName = "mediaCapacityArbitrator" + data { ["siteId"] = "'cloud-central'" } + } + ["operatorViewer"] { + stateMachineName = "cloudViewer" + subscription = Regex("incident-.*|zone[A-C]-cam-.*") + data { ["siteId"] = "'cloud-central'" } + } + + // NOTE: incidentManager instances are NOT declared here. They are created + // dynamically at runtime by the Incident Detector via the Instantiate action + // (one per detected incident, named incident--). +} + +// =========================================================================== +// External service bindings (outside compute / IO; NOT part of CSM logic). +// +// captionService -> BLIP_CAM (https://github.com/zawawiAI/BLIP_CAM) +// takes cameraId + mode, returns +// { caption, tags, anomalyScore }. Owns the per-camera +// CHAD cursor and exposes it for time alignment. +// mediaBurstService -> on escalation, extracts a finite FRAME BURST from +// CHAD (count/step), time-aligned to the camera's +// current cursor, and returns a mediaRef. No streaming. +// incidentStoreService -> persists incidents + media refs for the Operator Viewer. +// =========================================================================== +bindings { + new HttpServiceImplementationBinding { + name = "captionService" + `local` = false + scheme = "http" + host = "localhost" + port = 9100 + endPoint = "/caption" + method = "POST" + } + new HttpServiceImplementationBinding { + name = "mediaBurstService" + `local` = false + scheme = "http" + host = "localhost" + port = 9200 + endPoint = "/burst" + method = "POST" + } + new HttpServiceImplementationBinding { + name = "incidentStoreService" + `local` = false + scheme = "http" + host = "localhost" + port = 9300 + endPoint = "/store" + method = "POST" + } +} diff --git a/SASS/policies/correlationRules.pkl b/SASS/policies/correlationRules.pkl new file mode 100644 index 0000000..6ef023b --- /dev/null +++ b/SASS/policies/correlationRules.pkl @@ -0,0 +1,7 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +// Within-zone only. No cross-zone correlation. +rules = new Mapping { + ["scope"] = "'within-zone'" + ["agreement"] = "'exact-tag-match'" +} diff --git a/SASS/policies/escalationPolicy.pkl b/SASS/policies/escalationPolicy.pkl new file mode 100644 index 0000000..ab2d2e0 --- /dev/null +++ b/SASS/policies/escalationPolicy.pkl @@ -0,0 +1,8 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +// Maps incident confidence to the requested visual evidence level. +rules = new Mapping { + ["snapshotBelow"] = "1.8" + ["lowBelow"] = "2.5" + ["highAtOrAbove"] = "2.5" +} diff --git a/SASS/policies/priorityPolicy.pkl b/SASS/policies/priorityPolicy.pkl new file mode 100644 index 0000000..927d39e --- /dev/null +++ b/SASS/policies/priorityPolicy.pkl @@ -0,0 +1,6 @@ +import "https://raw.githubusercontent.com/CollaborativeStateMachines/Cirrina/refs/heads/develop/src/main/resources/pkl/csm/csml.pkl" + +rules = new Mapping { + ["ordering"] = "'confidence-desc'" + ["preemptWhen"] = "'confidence>2.5'" +} diff --git a/SASS/services/caption_service.py b/SASS/services/caption_service.py new file mode 100644 index 0000000..cc43f40 --- /dev/null +++ b/SASS/services/caption_service.py @@ -0,0 +1,259 @@ +#!/usr/bin/env python3 +""" +caption_service.py +================== +HTTP captioning service for the SaSS edge camera state machines, and the OWNER +of the per-camera CHAD cursor (the single source of truth for "where is this +camera now", shared with the media burst service for time alignment). + +Endpoints +--------- + POST /caption { cameraId, zoneId, mode } + -> { caption, tags, anomalyScore } + Advances the camera's cursor by one heartbeat window, captions the current + CHAD frame with BLIP, and derives anomalyScore from CHAD's binary label. + + GET /cursor?cameraId=... + -> { cameraId, stem, frameIdx, fps } + Reports the camera's CURRENT position so the media burst service can fetch + escalation frames that are time-aligned with the latest observation. + +Requires CHAD (videos + anomaly labels) and BLIP. +""" + +import argparse +import os +import threading +from collections import defaultdict +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import urlparse, parse_qs +import json + +import numpy as np +import cv2 +import torch +from PIL import Image +# Match BLIP_CAM (zawawiAI/BLIP_CAM): same processor/model classes & weights. +from transformers import AutoProcessor, AutoModelForImageTextToText + + +class BlipCaptioner: + """BLIP captioner mirroring zawawiAI/BLIP_CAM (load_models + _generate_caption).""" + + def __init__(self, model_name="Salesforce/blip-image-captioning-large"): + self._lock = threading.Lock() + self.processor = AutoProcessor.from_pretrained(model_name) + self.model = AutoModelForImageTextToText.from_pretrained(model_name) + self.device = "cuda" if torch.cuda.is_available() else "cpu" + if self.device == "cuda": + torch.cuda.set_per_process_memory_fraction(0.9) + self.model = self.model.to("cuda") + self.model.eval() + print(f"[caption_service] BLIP on {self.device}") + + def caption_image(self, image: "Image.Image") -> str: + with self._lock: + with torch.no_grad(): + inputs = self.processor(images=image, return_tensors="pt") + inputs = {k: v.to(self.device) for k, v in inputs.items()} + outputs = self.model.generate( + **inputs, max_length=30, num_beams=5, num_return_sequences=1 + ) + return self.processor.batch_decode(outputs, skip_special_tokens=True)[0].strip() + + +# --------------------------------------------------------------------------- +# CHAD dataset source + shared per-camera cursor +# --------------------------------------------------------------------------- +class ChadSource: + """ + Owns the per-camera cursor and reads CHAD videos + binary anomaly labels. + + CHAD layout: + CHAD_Videos/_