diff --git a/dashboard/index.html b/dashboard/index.html index 41bd37b..dd0be08 100644 --- a/dashboard/index.html +++ b/dashboard/index.html @@ -188,6 +188,65 @@ padding-top: 12px; } + .link-section h3 { + font-size: 11px; + text-transform: uppercase; + letter-spacing: 0.5px; + color: rgba(255, 255, 255, 0.5); + margin-bottom: 8px; + } + + /* Pattern visualization controls */ + .pattern-checkbox { + display: flex; + align-items: center; + gap: 8px; + padding: 6px 0; + cursor: pointer; + font-size: 12px; + color: rgba(255, 255, 255, 0.7); + } + + .pattern-checkbox:hover { + color: rgba(255, 255, 255, 0.9); + } + + .pattern-checkbox input[type="checkbox"] { + width: 14px; + height: 14px; + accent-color: #4fc3f7; + cursor: pointer; + } + + .pattern-filter { + display: flex; + align-items: center; + gap: 8px; + margin-top: 8px; + padding-top: 8px; + border-top: 1px solid rgba(255, 255, 255, 0.08); + font-size: 11px; + } + + .pattern-filter label { + color: rgba(255, 255, 255, 0.5); + } + + .pattern-filter select { + background: rgba(255, 255, 255, 0.08); + border: 1px solid rgba(255, 255, 255, 0.15); + border-radius: 4px; + color: #e0e0e0; + padding: 4px 8px; + font-size: 11px; + cursor: pointer; + } + + .pattern-filter select:focus { + outline: none; + border-color: #4fc3f7; + } + .link-item { padding: 6px 8px; margin-bottom: 4px; @@ -1569,6 +1628,31 @@
No links active
+
+

Patterns

+
+ + + +
+ + +
+
+
diff --git a/dashboard/js/app.js b/dashboard/js/app.js index e051f9d..12ff5a3 100644 --- a/dashboard/js/app.js +++ b/dashboard/js/app.js @@ -1237,4 +1237,24 @@ refreshNodeList: updateNodeList, refreshLinkList: updateLinkList }; + + // ============================================ + // Crowd Flow Visualization Controls + // Global wrappers for HTML onchange handlers -> Viz3D module + // ============================================ + window.toggleFlowLayer = function(visible) { + Viz3D.setFlowLayerVisible(visible); + }; + + window.toggleDwellLayer = function(visible) { + Viz3D.setDwellLayerVisible(visible); + }; + + window.toggleCorridorLayer = function(visible) { + Viz3D.setCorridorLayerVisible(visible); + }; + + window.setFlowTimeFilter = function(value) { + Viz3D.setFlowTimeFilter(value); + }; })(); diff --git a/mothership/cmd/mothership/main_phase6.go b/mothership/cmd/mothership/main_phase6.go index a5eb8a9..86944c5 100644 --- a/mothership/cmd/mothership/main_phase6.go +++ b/mothership/cmd/mothership/main_phase6.go @@ -31,6 +31,7 @@ import ( "github.com/spaxel/mothership/internal/mqtt" "github.com/spaxel/mothership/internal/notify" "github.com/spaxel/mothership/internal/ota" + "github.com/spaxel/mothership/internal/prediction" "github.com/spaxel/mothership/internal/provisioning" "github.com/spaxel/mothership/internal/recorder" "github.com/spaxel/mothership/internal/replay" @@ -182,6 +183,31 @@ func main() { fallDetector := falldetect.NewDetector() log.Printf("[INFO] Fall detector initialized") + // Phase 6: Prediction module for presence prediction + var predictionStore *prediction.ModelStore + var predictionHistory *prediction.HistoryUpdater + var predictionPredictor *prediction.Predictor + predictionStore, err = prediction.NewModelStore(filepath.Join(cfg.DataDir, "prediction.db")) + if err != nil { + log.Printf("[WARN] Failed to open prediction store: %v", err) + } else { + defer predictionStore.Close() + log.Printf("[INFO] Prediction store at %s", filepath.Join(cfg.DataDir, "prediction.db")) + + // Create history updater + predictionHistory = prediction.NewHistoryUpdater(predictionStore) + + // Load stored person zone positions + if err := predictionHistory.LoadStoredPositions(); err != nil { + log.Printf("[WARN] Failed to load stored prediction positions: %v", err) + } + + // Create predictor + predictionPredictor = prediction.NewPredictor(predictionStore) + + log.Printf("[INFO] Presence prediction initialized") + } + // Phase 6: Notification service notifyService, err := notify.NewService(filepath.Join(cfg.DataDir, "notify.db")) if err != nil { @@ -670,6 +696,11 @@ func main() { automationEngine.UpdateZoneDwellTracking(event.BlobID, event.ToZone, time.Now()) } } + + // Record zone transition for presence prediction + if predictionHistory != nil && personID != "" { + predictionHistory.PersonZoneChange(personID, event.FromZone, event.ToZone, event.BlobID, time.Now()) + } }) } @@ -802,6 +833,74 @@ func main() { log.Printf("[INFO] Flow analytics background tasks started (prune: 24h, corridors: 7d)") } + // Phase 6: Prediction provider wiring and update loop + if predictionPredictor != nil && predictionHistory != nil { + // Wire zone provider + if zonesMgr != nil { + predictionPredictor.SetZoneProvider(&predictionZoneAdapter{mgr: zonesMgr}) + } + + // Wire person provider + if bleRegistry != nil { + predictionPredictor.SetPersonProvider(&predictionPersonAdapter{registry: bleRegistry}) + } + + // Wire position provider + predictionPredictor.SetPositionProvider(prediction.NewPositionAdapter(predictionHistory)) + + // Wire MQTT client for prediction publishing + if mqttClient != nil && mqttClient.IsConnected() { + predictionPredictor.SetMQTTClient(&predictionMQTTAdapter{client: mqttClient}, "") + } + + // Start periodic prediction update loop (every 60 seconds) + go func() { + ticker := time.NewTicker(60 * time.Second) + defer ticker.Stop() + + // Run initial prediction after 5 seconds + time.Sleep(5 * time.Second) + predictionPredictor.UpdatePredictions() + log.Printf("[INFO] Prediction: initial predictions computed") + + // Publish prediction sensors for each person + if mqttClient != nil && mqttClient.IsConnected() && bleRegistry != nil { + people, _ := bleRegistry.GetPeople() + for _, person := range people { + mqttClient.PublishPredictionSensors(person.ID, person.Name) + } + } + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + predictionPredictor.UpdatePredictions() + + // Publish predictions to MQTT + if mqttClient != nil && mqttClient.IsConnected() { + predictions := predictionPredictor.GetPredictions() + for _, pred := range predictions { + zoneName := pred.PredictedNextZoneName + if zoneName == "" { + zoneName = pred.PredictedNextZoneID + } + mqttClient.UpdatePredictionState( + pred.PersonID, + zoneName, + pred.DataConfidence, + pred.PredictionConfidence, + pred.EstimatedTransitionMinutes, + ) + } + } + } + } + }() + log.Printf("[INFO] Prediction update loop started (interval: 60s)") + } + // Fleet REST API fleetHandler := fleet.NewHandler(fleetMgr) fleetHandler.RegisterRoutes(r) @@ -1327,6 +1426,44 @@ func main() { analyticsHandler.RegisterRoutes(r) } + // Phase 6: Prediction REST API + if predictionPredictor != nil { + r.Get("/api/predictions", func(w http.ResponseWriter, r *http.Request) { + predictions := predictionPredictor.GetPredictions() + writeJSON(w, predictions) + }) + + r.Get("/api/predictions/stats", func(w http.ResponseWriter, r *http.Request) { + if predictionHistory == nil { + http.Error(w, "prediction history not available", http.StatusServiceUnavailable) + return + } + count, dataAge, err := predictionHistory.GetTransitionStats() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, map[string]interface{}{ + "transition_count": count, + "data_age_days": dataAge.Hours() / 24, + "minimum_data_age": prediction.MinimumDataAge.Hours() / 24, + "has_minimum_data": dataAge >= prediction.MinimumDataAge, + }) + }) + + r.Post("/api/predictions/recompute", func(w http.ResponseWriter, r *http.Request) { + if predictionHistory == nil { + http.Error(w, "prediction history not available", http.StatusServiceUnavailable) + return + } + if err := predictionHistory.ForceRecompute(); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + writeJSON(w, map[string]string{"status": "recompute_started"}) + }) + } + // Phase 6: Learning feedback REST API if feedbackStore != nil { learningHandler := learning.NewHandler(feedbackStore, feedbackProcessor, accuracyComputer) @@ -1568,3 +1705,132 @@ func (n *notifySenderAdapter) SendViaChannel(channelType string, title, body str return n.service.Send(notif) } +// Prediction provider adapters + +type predictionZoneAdapter struct { + mgr *zones.Manager +} + +func (z *predictionZoneAdapter) GetZone(id string) (string, bool) { + zone := z.mgr.GetZone(id) + if zone == nil { + return "", false + } + return zone.Name, true +} + +type predictionPersonAdapter struct { + registry *ble.Registry +} + +func (p *predictionPersonAdapter) GetPerson(id string) (string, string, bool) { + person, err := p.registry.GetPerson(id) + if err != nil { + return "", "", false + } + return person.Name, person.Color, true +} + +func (p *predictionPersonAdapter) GetAllPeople() ([]struct { + ID string + Name string + Color string +}, error) { + people, err := p.registry.GetPeople() + if err != nil { + return nil, err + } + result := make([]struct { + ID string + Name string + Color string + }, len(people)) + for i, person := range people { + result[i] = struct { + ID string + Name string + Color string + }{ID: person.ID, Name: person.Name, Color: person.Color} + } + return result, nil +} + +type predictionMQTTAdapter struct { + client *mqtt.Client +} + +func (m *predictionMQTTAdapter) Publish(topic string, payload []byte) error { + return m.client.Publish(topic, payload) +} + +func (m *predictionMQTTAdapter) IsConnected() bool { + return m.client.IsConnected() +} + +// Prediction provider adapters + +type predictionZoneAdapter struct { + mgr *zones.Manager +} + +func (z *predictionZoneAdapter) GetZone(id string) (string, bool) { + zone := z.mgr.GetZone(id) + if zone == nil { + return "", false + } + return zone.Name, true +} + +type predictionPersonAdapter struct { + registry *ble.Registry +} + +func (p *predictionPersonAdapter) GetPerson(id string) (string, string, bool) { + person, err := p.registry.GetPerson(id) + if err != nil { + return "", "", false + } + return person.Name, person.Color, true +} + +func (p *predictionPersonAdapter) GetAllPeople() ([]struct { + ID string + Name string + Color string +}, error) { + people, err := p.registry.GetPeople() + if err != nil { + return nil, err + } + + result := make([]struct { + ID string + Name string + Color string + }, len(people)) + for i, person := range people { + result[i] = struct { + ID string + Name string + Color string + }{ + ID: person.ID, + Name: person.Name, + Color: person.Color, + } + } + return result, nil +} + +type predictionMQTTAdapter struct { + client *mqtt.Client +} + +func (m *predictionMQTTAdapter) Publish(topic string, payload []byte) error { + return m.client.Publish(topic, payload) +} + +func (m *predictionMQTTAdapter) IsConnected() bool { + return m.client.IsConnected() +} + diff --git a/mothership/internal/analytics/anomaly.go b/mothership/internal/analytics/anomaly.go new file mode 100644 index 0000000..9e7b437 --- /dev/null +++ b/mothership/internal/analytics/anomaly.go @@ -0,0 +1,1249 @@ +// Package analytics provides anomaly detection based on learned normal behaviour patterns. +package analytics + +import ( + "database/sql" + "fmt" + "log" + "math" + "os" + "path/filepath" + "sync" + "time" + + "github.com/google/uuid" + "github.com/spaxel/mothership/internal/events" + + _ "modernc.org/sqlite" +) + +// NormalBehaviourSlot represents expected behaviour for a specific hour_of_week and zone. +type NormalBehaviourSlot struct { + HourOfWeek int `json:"hour_of_week"` // 0-167 + ZoneID string `json:"zone_id"` + ExpectedOccupancy float64 ` json:"expected_occupancy"` // 0.0-1.0, fraction of samples with occupancy + TypicalPersonCount float64 `json:"typical_person_count"` // Mean person count + SampleCount int `json:"sample_count"` + TypicalBLEDevices map[string]float64 `json:"typical_ble_devices,omitempty"` // MAC -> frequency (0.0-1.0) +} + +// DwellBehaviourSlot represents expected dwell duration for a person in a zone at a specific hour. +type DwellBehaviourSlot struct { + HourOfWeek int `json:"hour_of_week"` + ZoneID string `json:"zone_id"` + PersonID string `json:"person_id"` + MeanDwellDuration time.Duration `json:"mean_dwell_duration"` + StdDwellDuration time.Duration `json:"std_dwell_duration"` + SampleCount int `json:"sample_count"` +} + +// AnomalyScoreConfig holds configurable thresholds for anomaly scoring. +type AnomalyScoreConfig struct { + // Unusual hour presence + UnusualHourScore float64 `json:"unusual_hour_score"` // Default: 0.7 + UnusualHourScoreSecurity float64 `json:"unusual_hour_score_security"` // Default: 0.9 + LateNightMultiplier float64 `json:"late_night_multiplier"` // Default: 1.5 (00:00-06:00) + + // Unknown BLE device + UnknownBLEScore float64 `json:"unknown_ble_score"` // Default: 0.5 + UnknownBLEScoreSecurity float64 `json:"unknown_ble_score_security"` // Default: 0.8 + SeenOnceScore float64 `json:"seen_once_score"` // Default: 0.3 + CloseRangeRSSIThreshold int `json:"close_range_rssi_threshold"` // Default: -60 dBm + + // Motion during away + MotionDuringAwayScore float64 `json:"motion_during_away_score"` // Default: 0.95 + + // Unusual dwell duration + UnusualDwellScore float64 `json:"unusual_dwell_score"` // Default: 0.4 + DwellMultiplierThreshold float64 `json:"dwell_multiplier_threshold"` // Default: 5.0 + + // Alert thresholds + AlertThresholdNormal float64 `json:"alert_threshold_normal"` // Default: 0.6 + AlertThresholdSecurity float64 `json:"alert_threshold_security"` // Default: 0.4 + + // Auto-away/disarm + AutoAwayDuration time.Duration `json:"auto_away_duration"` // Default: 15 minutes + AutoDisarmRSSIThreshold int `json:"auto_disarm_rssi_threshold"` // Default: -70 dBm + ManualOverrideDuration time.Duration `json:"manual_override_duration"` // Default: 30 minutes +} + +// DefaultAnomalyScoreConfig returns default configuration. +func DefaultAnomalyScoreConfig() AnomalyScoreConfig { + return AnomalyScoreConfig{ + UnusualHourScore: 0.7, + UnusualHourScoreSecurity: 0.9, + LateNightMultiplier: 1.5, + UnknownBLEScore: 0.5, + UnknownBLEScoreSecurity: 0.8, + SeenOnceScore: 0.3, + CloseRangeRSSIThreshold: -60, + MotionDuringAwayScore: 0.95, + UnusualDwellScore: 0.4, + DwellMultiplierThreshold: 5.0, + AlertThresholdNormal: 0.6, + AlertThresholdSecurity: 0.4, + AutoAwayDuration: 15 * time.Minute, + AutoDisarmRSSIThreshold: -70, + ManualOverrideDuration: 30 * time.Minute, + } +} + +// Detector detects anomalies based on learned normal behaviour. +type Detector struct { + mu sync.RWMutex + db *sql.DB + config AnomalyScoreConfig + + // Normal behaviour model (loaded from DB) + behaviourSlots map[string]*NormalBehaviourSlot // key: "hour-zone" + dwellSlots map[string]*DwellBehaviourSlot // key: "hour-zone-person" + + // Active anomaly tracking + activeAnomalies map[string]*events.AnomalyEvent // id -> event + anomalyHistory []*events.AnomalyEvent + + // Pending alert timers + pendingAlerts map[string]*alertTimerState + + // Model state + learningStartTime time.Time + modelReady bool + modelReadyAt time.Time + + // Registered devices and people + registeredDevices map[string]bool // MAC -> registered + registeredPeople map[string]string // person_id -> name + deviceFirstSeen map[string]time.Time // MAC -> first seen time + + // Providers + zoneProvider ZoneProvider + personProvider PersonProvider + deviceProvider DeviceProvider + positionProvider PositionProvider + alertHandler AlertHandler + + // Callbacks + onAnomaly func(event events.AnomalyEvent) + onModeChange func(event events.SystemModeChangeEvent) +} + +// ZoneProvider provides zone information. +type ZoneProvider interface { + GetZoneName(zoneID string) string + GetZoneOccupancy(zoneID string) (count int, blobIDs []int) +} + +// PersonProvider provides person information. +type PersonProvider interface { + GetPersonDevices(personID string) ([]string, error) + GetAllRegisteredDevices() (map[string]string, error) // MAC -> person_id + GetPersonName(personID string) string +} + +// DeviceProvider provides device information. +type DeviceProvider interface { + IsDeviceRegistered(mac string) bool + IsDeviceSeenBefore(mac string) bool + GetDeviceName(mac string) string +} + +// PositionProvider provides position for blobs. +type PositionProvider interface { + GetBlobPosition(blobID int) (x, y, z float64, ok bool) +} + +// AlertHandler handles alert delivery. +type AlertHandler interface { + SendAlert(event events.AnomalyEvent, immediate bool) error + SendWebhook(event events.AnomalyEvent, immediate bool) error + SendEscalation(event events.AnomalyEvent) error +} + +type alertTimerState struct { + alertTimer *time.Timer + webhookTimer *time.Timer + escalationTimer *time.Timer + anomalyID string +} + +// NewDetector creates a new anomaly detector. +func NewDetector(dbPath string, config AnomalyScoreConfig) (*Detector, error) { + if err := os.MkdirAll(filepath.Dir(dbPath), 0755); err != nil { + return nil, fmt.Errorf("create data dir: %w", err) + } + + db, err := sql.Open("sqlite", dbPath) + if err != nil { + return nil, fmt.Errorf("open sqlite: %w", err) + } + db.SetMaxOpenConns(1) + + d := &Detector{ + db: db, + config: config, + behaviourSlots: make(map[string]*NormalBehaviourSlot), + dwellSlots: make(map[string]*DwellBehaviourSlot), + activeAnomalies: make(map[string]*events.AnomalyEvent), + pendingAlerts: make(map[string]*alertTimerState), + registeredDevices: make(map[string]bool), + registeredPeople: make(map[string]string), + deviceFirstSeen: make(map[string]time.Time), + } + + if err := d.migrate(); err != nil { + db.Close() + return nil, fmt.Errorf("migrate: %w", err) + } + + if err := d.loadBehaviourModel(); err != nil { + log.Printf("[WARN] Failed to load behaviour model: %v", err) + } + + if err := d.loadLearningState(); err != nil { + log.Printf("[WARN] Failed to load learning state: %v", err) + } + + return d, nil +} + +func (d *Detector) migrate() error { + _, err := d.db.Exec(` + CREATE TABLE IF NOT EXISTS behaviour_slots ( + hour_of_week INTEGER NOT NULL, + zone_id TEXT NOT NULL, + expected_occupancy REAL NOT NULL DEFAULT 0, + typical_person_count REAL NOT NULL DEFAULT 0, + sample_count INTEGER NOT NULL DEFAULT 0, + typical_ble_devices TEXT NOT NULL DEFAULT '{}', + PRIMARY KEY (hour_of_week, zone_id) + ); + + CREATE TABLE IF NOT EXISTS dwell_slots ( + hour_of_week INTEGER NOT NULL, + zone_id TEXT NOT NULL, + person_id TEXT NOT NULL, + mean_dwell_ns INTEGER NOT NULL DEFAULT 0, + std_dwell_ns INTEGER NOT NULL DEFAULT 0, + sample_count INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (hour_of_week, zone_id, person_id) + ); + + CREATE TABLE IF NOT EXISTS occupancy_samples ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + hour_of_week INTEGER NOT NULL, + zone_id TEXT NOT NULL, + person_count INTEGER NOT NULL, + ble_devices TEXT NOT NULL DEFAULT '[]', + timestamp INTEGER NOT NULL + ); + + CREATE TABLE IF NOT EXISTS dwell_samples ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + hour_of_week INTEGER NOT NULL, + zone_id TEXT NOT NULL, + person_id TEXT NOT NULL, + dwell_ns INTEGER NOT NULL, + timestamp INTEGER NOT NULL + ); + + CREATE TABLE IF NOT EXISTS anomaly_events ( + id TEXT PRIMARY KEY, + type TEXT NOT NULL, + score REAL NOT NULL, + description TEXT NOT NULL, + timestamp INTEGER NOT NULL, + zone_id TEXT, + zone_name TEXT, + blob_id INTEGER, + person_id TEXT, + person_name TEXT, + device_mac TEXT, + device_name TEXT, + position_x REAL, + position_y REAL, + position_z REAL, + hour_of_week INTEGER, + expected_occupancy REAL, + dwell_duration_ns INTEGER, + expected_dwell_ns INTEGER, + rssi_dbm INTEGER, + seen_before INTEGER, + acknowledged INTEGER NOT NULL DEFAULT 0, + acknowledged_at INTEGER, + feedback TEXT, + alert_sent INTEGER NOT NULL DEFAULT 0, + webhook_sent INTEGER NOT NULL DEFAULT 0, + escalation_sent INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS learning_state ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS device_first_seen ( + mac TEXT PRIMARY KEY, + first_seen_ns INTEGER NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_occupancy_samples_time ON occupancy_samples(timestamp); + CREATE INDEX IF NOT EXISTS idx_dwell_samples_time ON dwell_samples(timestamp); + CREATE INDEX IF NOT EXISTS idx_anomaly_events_time ON anomaly_events(timestamp); + `) + return err +} + +func (d *Detector) loadBehaviourModel() error { + // Load behaviour slots + rows, err := d.db.Query(` + SELECT hour_of_week, zone_id, expected_occupancy, typical_person_count, sample_count, typical_ble_devices + FROM behaviour_slots + `) + if err != nil { + return err + } + defer rows.Close() + + for rows.Next() { + slot := &NormalBehaviourSlot{ + TypicalBLEDevices: make(map[string]float64), + } + var bleDevicesJSON string + if err := rows.Scan(&slot.HourOfWeek, &slot.ZoneID, &slot.ExpectedOccupancy, + &slot.TypicalPersonCount, &slot.SampleCount, &bleDevicesJSON); err != nil { + continue + } + // Parse BLE devices JSON + if bleDevicesJSON != "" && bleDevicesJSON != "{}" { + var devices map[string]float64 + if err := jsonUnmarshal(bleDevicesJSON, &devices); err == nil { + slot.TypicalBLEDevices = devices + } + } + key := fmt.Sprintf("%d-%s", slot.HourOfWeek, slot.ZoneID) + d.behaviourSlots[key] = slot + } + + // Load dwell slots + dwellRows, err := d.db.Query(` + SELECT hour_of_week, zone_id, person_id, mean_dwell_ns, std_dwell_ns, sample_count + FROM dwell_slots + `) + if err != nil { + return err + } + defer dwellRows.Close() + + for dwellRows.Next() { + slot := &DwellBehaviourSlot{} + var meanNS, stdNS int64 + if err := dwellRows.Scan(&slot.HourOfWeek, &slot.ZoneID, &slot.PersonID, + &meanNS, &stdNS, &slot.SampleCount); err != nil { + continue + } + slot.MeanDwellDuration = time.Duration(meanNS) + slot.StdDwellDuration = time.Duration(stdNS) + key := fmt.Sprintf("%d-%s-%s", slot.HourOfWeek, slot.ZoneID, slot.PersonID) + d.dwellSlots[key] = slot + } + + return nil +} + +func (d *Detector) loadLearningState() error { + var startNS int64 + err := d.db.QueryRow(`SELECT value FROM learning_state WHERE key = 'learning_start'`).Scan(&startNS) + if err == sql.ErrNoRows { + // Initialize learning start time + d.learningStartTime = time.Now() + d.db.Exec(`INSERT INTO learning_state (key, value) VALUES ('learning_start', ?)`, time.Now().UnixNano()) + return nil + } + if err != nil { + return err + } + + d.learningStartTime = time.Unix(0, startNS) + + // Check if 7 days have passed + if time.Since(d.learningStartTime) >= 7*24*time.Hour { + d.modelReady = true + d.modelReadyAt = d.learningStartTime.Add(7 * 24 * time.Hour) + } + + // Load device first seen times + deviceRows, err := d.db.Query(`SELECT mac, first_seen_ns FROM device_first_seen`) + if err != nil { + return err + } + defer deviceRows.Close() + + for deviceRows.Next() { + var mac string + var firstSeenNS int64 + if err := deviceRows.Scan(&mac, &firstSeenNS); err != nil { + continue + } + d.deviceFirstSeen[mac] = time.Unix(0, firstSeenNS) + } + + return nil +} + +// Close closes the database. +func (d *Detector) Close() error { + return d.db.Close() +} + +// SetZoneProvider sets the zone provider. +func (d *Detector) SetZoneProvider(p ZoneProvider) { + d.mu.Lock() + d.zoneProvider = p + d.mu.Unlock() +} + +// SetPersonProvider sets the person provider. +func (d *Detector) SetPersonProvider(p PersonProvider) { + d.mu.Lock() + d.personProvider = p + d.mu.Unlock() +} + +// SetDeviceProvider sets the device provider. +func (d *Detector) SetDeviceProvider(p DeviceProvider) { + d.mu.Lock() + d.deviceProvider = p + d.mu.Unlock() +} + +// SetPositionProvider sets the position provider. +func (d *Detector) SetPositionProvider(p PositionProvider) { + d.mu.Lock() + d.positionProvider = p + d.mu.Unlock() +} + +// SetAlertHandler sets the alert handler. +func (d *Detector) SetAlertHandler(h AlertHandler) { + d.mu.Lock() + d.alertHandler = h + d.mu.Unlock() +} + +// SetOnAnomaly sets callback for anomaly events. +func (d *Detector) SetOnAnomaly(cb func(event events.AnomalyEvent)) { + d.mu.Lock() + d.onAnomaly = cb + d.mu.Unlock() +} + +// SetOnModeChange sets callback for mode change events. +func (d *Detector) SetOnModeChange(cb func(event events.SystemModeChangeEvent)) { + d.mu.Lock() + d.onModeChange = cb + d.mu.Unlock() +} + +// SetRegisteredDevices sets the list of registered BLE devices. +func (d *Detector) SetRegisteredDevices(devices []string) { + d.mu.Lock() + defer d.mu.Unlock() + + d.registeredDevices = make(map[string]bool) + for _, mac := range devices { + d.registeredDevices[mac] = true + } +} + +// IsModelReady returns true if 7 days of learning have passed. +func (d *Detector) IsModelReady() bool { + d.mu.RLock() + defer d.mu.RUnlock() + return d.modelReady +} + +// GetLearningProgress returns the fraction of learning completed (0.0-1.0). +func (d *Detector) GetLearningProgress() float64 { + d.mu.RLock() + defer d.mu.RUnlock() + + if d.modelReady { + return 1.0 + } + + elapsed := time.Since(d.learningStartTime) + total := 7 * 24 * time.Hour + progress := float64(elapsed) / float64(total) + if progress > 1.0 { + progress = 1.0 + } + return progress +} + +// ProcessOccupancy records an occupancy observation and checks for unusual hour anomalies. +func (d *Detector) ProcessOccupancy(zoneID string, personCount int, bleDevices []string, isSecurityMode bool) *events.AnomalyEvent { + d.mu.Lock() + defer d.mu.Unlock() + + now := time.Now() + hourOfWeek := getHourOfWeek(now) + + // Record the sample + d.recordOccupancySample(hourOfWeek, zoneID, personCount, bleDevices, now) + + // Check for anomaly (only if model is ready, or if in security mode) + if !d.modelReady && !isSecurityMode { + return nil + } + + key := fmt.Sprintf("%d-%s", hourOfWeek, zoneID) + slot, exists := d.behaviourSlots[key] + + if !exists || slot.SampleCount < 10 { + // Not enough data for this slot + return nil + } + + // Check if this is an unusual hour (low expected occupancy but we see people) + if personCount > 0 && slot.ExpectedOccupancy < 0.1 { + score := d.config.UnusualHourScore + if isSecurityMode { + score = d.config.UnusualHourScoreSecurity + } + + // Apply late night multiplier (00:00-06:00) + hour := now.Hour() + if hour >= 0 && hour < 6 { + score *= d.config.LateNightMultiplier + if score > 1.0 { + score = 1.0 + } + } + + // Get zone name + zoneName := zoneID + if d.zoneProvider != nil { + zoneName = d.zoneProvider.GetZoneName(zoneID) + } + + // Create anomaly event + event := events.AnomalyEvent{ + ID: uuid.New().String(), + Type: events.AnomalyUnusualHour, + Score: score, + Description: fmt.Sprintf("Motion detected in %s at %s (unusual hour)", zoneName, now.Format("3:04pm")), + Timestamp: now, + ZoneID: zoneID, + ZoneName: zoneName, + HourOfWeek: hourOfWeek, + ExpectedOccupancy: slot.ExpectedOccupancy, + } + + return d.createAnomaly(&event, isSecurityMode) + } + + return nil +} + +// ProcessBLEDevice checks for unknown BLE device anomalies. +func (d *Detector) ProcessBLEDevice(mac string, rssi int, isSecurityMode bool) *events.AnomalyEvent { + d.mu.Lock() + defer d.mu.Unlock() + + now := time.Now() + + // Track first seen time for this device + if _, exists := d.deviceFirstSeen[mac]; !exists { + d.deviceFirstSeen[mac] = now + d.db.Exec(`INSERT OR REPLACE INTO device_first_seen (mac, first_seen_ns) VALUES (?, ?)`, + mac, now.UnixNano()) + } + + // Check if device is registered + if d.registeredDevices[mac] { + return nil + } + + // Check if close range + if rssi < d.config.CloseRangeRSSIThreshold { + return nil // Not close enough to be concerning + } + + // Check if device was seen before + seenBefore := false + if d.deviceProvider != nil { + seenBefore = d.deviceProvider.IsDeviceSeenBefore(mac) + } + + // Calculate score + var score float64 + if !seenBefore { + // Never seen before + score = d.config.UnknownBLEScore + if isSecurityMode { + score = d.config.UnknownBLEScoreSecurity + } + } else { + // Seen before but not registered + score = d.config.SeenOnceScore + } + + if score < d.getAlertThreshold(isSecurityMode) { + return nil + } + + // Get device name + deviceName := mac + if d.deviceProvider != nil { + deviceName = d.deviceProvider.GetDeviceName(mac) + } + + event := events.AnomalyEvent{ + ID: uuid.New().String(), + Type: events.AnomalyUnknownBLE, + Score: score, + Description: fmt.Sprintf("Unknown device detected nearby: %s (RSSI: %d dBm)", deviceName, rssi), + Timestamp: now, + DeviceMAC: mac, + DeviceName: deviceName, + RSSIdBm: rssi, + SeenBefore: seenBefore, + } + + return d.createAnomaly(&event, isSecurityMode) +} + +// ProcessMotionDuringAway checks for motion when system is in away mode. +func (d *Detector) ProcessMotionDuringAway(zoneID string, blobID int, isSecurityMode bool) *events.AnomalyEvent { + d.mu.Lock() + defer d.mu.Unlock() + + now := time.Now() + + // This anomaly always fires regardless of model training status + score := d.config.MotionDuringAwayScore + + // Get zone name + zoneName := zoneID + if d.zoneProvider != nil { + zoneName = d.zoneProvider.GetZoneName(zoneID) + } + + // Get position + var pos events.Position + if d.positionProvider != nil { + x, y, z, ok := d.positionProvider.GetBlobPosition(blobID) + if ok { + pos = events.Position{X: x, Y: y, Z: z} + } + } + + event := events.AnomalyEvent{ + ID: uuid.New().String(), + Type: events.AnomalyMotionDuringAway, + Score: score, + Description: fmt.Sprintf("Motion detected in %s while everyone is away", zoneName), + Timestamp: now, + ZoneID: zoneID, + ZoneName: zoneName, + BlobID: blobID, + Position: pos, + } + + return d.createAnomaly(&event, isSecurityMode) +} + +// ProcessDwellDuration checks for unusual dwell duration. +func (d *Detector) ProcessDwellDuration(zoneID, personID string, dwellDuration time.Duration, isSecurityMode bool, isFallDetected bool) *events.AnomalyEvent { + d.mu.Lock() + defer d.mu.Unlock() + + // Don't report if fall is already detected (fall detection takes priority) + if isFallDetected { + return nil + } + + now := time.Now() + hourOfWeek := getHourOfWeek(now) + + // Record the sample + d.recordDwellSample(hourOfWeek, zoneID, personID, dwellDuration, now) + + // Only check if model is ready (this anomaly requires learned patterns) + if !d.modelReady { + return nil + } + + key := fmt.Sprintf("%d-%s-%s", hourOfWeek, zoneID, personID) + slot, exists := d.dwellSlots[key] + + if !exists || slot.SampleCount < 5 { + return nil + } + + // Check if dwelling for > 5x mean + if dwellDuration > time.Duration(float64(slot.MeanDwellDuration)*d.config.DwellMultiplierThreshold) { + score := d.config.UnusualDwellScore + + // Get names + zoneName := zoneID + if d.zoneProvider != nil { + zoneName = d.zoneProvider.GetZoneName(zoneID) + } + personName := personID + if d.personProvider != nil { + personName = d.personProvider.GetPersonName(personID) + } + + event := events.AnomalyEvent{ + ID: uuid.New().String(), + Type: events.AnomalyUnusualDwell, + Score: score, + Description: fmt.Sprintf("%s in %s for longer than usual (%.0f minutes)", personName, zoneName, dwellDuration.Minutes()), + Timestamp: now, + ZoneID: zoneID, + ZoneName: zoneName, + PersonID: personID, + PersonName: personName, + DwellDuration: dwellDuration, + ExpectedDwell: slot.MeanDwellDuration, + } + + return d.createAnomaly(&event, isSecurityMode) + } + + return nil +} + +func (d *Detector) createAnomaly(event *events.AnomalyEvent, isSecurityMode bool) *events.AnomalyEvent { + threshold := d.getAlertThreshold(isSecurityMode) + if event.Score < threshold { + return nil + } + + // Store in active anomalies + d.activeAnomalies[event.ID] = event + + // Persist to database + d.persistAnomaly(event) + + // Start alert chain + d.startAlertChain(event, isSecurityMode) + + // Fire callback + if d.onAnomaly != nil { + go d.onAnomaly(*event) + } + + log.Printf("[INFO] Anomaly detected: %s (score=%.2f, type=%s)", event.Description, event.Score, event.Type) + + return event +} + +func (d *Detector) getAlertThreshold(isSecurityMode bool) float64 { + if isSecurityMode { + return d.config.AlertThresholdSecurity + } + return d.config.AlertThresholdNormal +} + +func (d *Detector) recordOccupancySample(hourOfWeek int, zoneID string, personCount int, bleDevices []string, timestamp time.Time) { + devicesJSON, _ := jsonMarshal(bleDevices) + _, err := d.db.Exec(` + INSERT INTO occupancy_samples (hour_of_week, zone_id, person_count, ble_devices, timestamp) + VALUES (?, ?, ?, ?, ?) + `, hourOfWeek, zoneID, personCount, string(devicesJSON), timestamp.UnixNano()) + if err != nil { + log.Printf("[WARN] Failed to record occupancy sample: %v", err) + } +} + +func (d *Detector) recordDwellSample(hourOfWeek int, zoneID, personID string, dwellDuration time.Duration, timestamp time.Time) { + _, err := d.db.Exec(` + INSERT INTO dwell_samples (hour_of_week, zone_id, person_id, dwell_ns, timestamp) + VALUES (?, ?, ?, ?, ?) + `, hourOfWeek, zoneID, personID, dwellDuration.Nanoseconds(), timestamp.UnixNano()) + if err != nil { + log.Printf("[WARN] Failed to record dwell sample: %v", err) + } +} + +func (d *Detector) persistAnomaly(event *events.AnomalyEvent) { + _, err := d.db.Exec(` + INSERT INTO anomaly_events ( + id, type, score, description, timestamp, + zone_id, zone_name, blob_id, person_id, person_name, + device_mac, device_name, position_x, position_y, position_z, + hour_of_week, expected_occupancy, dwell_duration_ns, expected_dwell_ns, + rssi_dbm, seen_before + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, event.ID, event.Type, event.Score, event.Description, event.Timestamp.UnixNano(), + nullString(event.ZoneID), nullString(event.ZoneName), event.BlobID, + nullString(event.PersonID), nullString(event.PersonName), + nullString(event.DeviceMAC), nullString(event.DeviceName), + event.Position.X, event.Position.Y, event.Position.Z, + event.HourOfWeek, event.ExpectedOccupancy, + event.DwellDuration.Nanoseconds(), event.ExpectedDwell.Nanoseconds(), + event.RSSIdBm, event.SeenBefore) + if err != nil { + log.Printf("[WARN] Failed to persist anomaly: %v", err) + } +} + +func (d *Detector) startAlertChain(event *events.AnomalyEvent, isSecurityMode bool) { + state := &alertTimerState{ + anomalyID: event.ID, + } + + // T+0: Dashboard alarm (immediate - handled by UI via callback) + // Fire alert handler immediately for dashboard + if d.alertHandler != nil { + go d.alertHandler.SendAlert(*event, isSecurityMode) + } + + if isSecurityMode { + // Security mode: all alerts fire immediately + if d.alertHandler != nil { + d.alertHandler.SendWebhook(*event, true) + d.alertHandler.SendEscalation(*event) + } + event.AlertSent = true + event.WebhookSent = true + event.EscalationSent = true + now := time.Now() + event.AlertSentAt = now + event.WebhookSentAt = now + event.EscalationSentAt = now + d.updateAnomalyAlertState(event) + } else { + // Normal mode: staged alerts + // T+30s: notification + state.alertTimer = time.AfterFunc(30*time.Second, func() { + d.mu.Lock() + defer d.mu.Unlock() + if anomaly, exists := d.activeAnomalies[event.ID]; exists && !anomaly.Acknowledged { + if d.alertHandler != nil { + d.alertHandler.SendAlert(*anomaly, false) + } + anomaly.AlertSent = true + anomaly.AlertSentAt = time.Now() + d.updateAnomalyAlertState(anomaly) + } + }) + + // T+2min: webhook + state.webhookTimer = time.AfterFunc(2*time.Minute, func() { + d.mu.Lock() + defer d.mu.Unlock() + if anomaly, exists := d.activeAnomalies[event.ID]; exists && !anomaly.Acknowledged { + if d.alertHandler != nil { + d.alertHandler.SendWebhook(*anomaly, false) + } + anomaly.WebhookSent = true + anomaly.WebhookSentAt = time.Now() + d.updateAnomalyAlertState(anomaly) + } + }) + + // T+5min: escalation + state.escalationTimer = time.AfterFunc(5*time.Minute, func() { + d.mu.Lock() + defer d.mu.Unlock() + if anomaly, exists := d.activeAnomalies[event.ID]; exists && !anomaly.Acknowledged { + if d.alertHandler != nil { + d.alertHandler.SendEscalation(*anomaly) + } + anomaly.EscalationSent = true + anomaly.EscalationSentAt = time.Now() + d.updateAnomalyAlertState(anomaly) + } + }) + } + + d.pendingAlerts[event.ID] = state +} + +func (d *Detector) updateAnomalyAlertState(event *events.AnomalyEvent) { + d.db.Exec(` + UPDATE anomaly_events SET + alert_sent = ?, alert_sent_at = ?, + webhook_sent = ?, webhook_sent_at = ?, + escalation_sent = ?, escalation_sent_at = ? + WHERE id = ? + `, event.AlertSent, nullTime(event.AlertSentAt), + event.WebhookSent, nullTime(event.WebhookSentAt), + event.EscalationSent, nullTime(event.EscalationSentAt), + event.ID) +} + +// AcknowledgeAnomaly acknowledges an anomaly and cancels pending timers. +func (d *Detector) AcknowledgeAnomaly(anomalyID, feedback, acknowledgedBy string) error { + d.mu.Lock() + defer d.mu.Unlock() + + event, exists := d.activeAnomalies[anomalyID] + if !exists { + return fmt.Errorf("anomaly not found: %s", anomalyID) + } + + // Cancel pending timers + if state, exists := d.pendingAlerts[anomalyID]; exists { + if state.alertTimer != nil { + state.alertTimer.Stop() + } + if state.webhookTimer != nil { + state.webhookTimer.Stop() + } + if state.escalationTimer != nil { + state.escalationTimer.Stop() + } + delete(d.pendingAlerts, anomalyID) + } + + // Update event + event.Acknowledged = true + event.AcknowledgedAt = time.Now() + event.Feedback = feedback + event.AcknowledgedBy = acknowledgedBy + + // Update database + _, err := d.db.Exec(` + UPDATE anomaly_events SET + acknowledged = 1, + acknowledged_at = ?, + feedback = ? + WHERE id = ? + `, event.AcknowledgedAt.UnixNano(), feedback, anomalyID) + + if err != nil { + return err + } + + log.Printf("[INFO] Anomaly acknowledged: %s (feedback: %s)", anomalyID, feedback) + + return nil +} + +// UpdateBehaviourModel updates the behaviour model from collected samples. +// Should be called periodically (e.g., weekly). +func (d *Detector) UpdateBehaviourModel() error { + d.mu.Lock() + defer d.mu.Unlock() + + log.Printf("[INFO] Updating behaviour model from collected samples...") + + // Update behaviour slots from occupancy samples + rows, err := d.db.Query(` + SELECT hour_of_week, zone_id, + AVG(CASE WHEN person_count > 0 THEN 1.0 ELSE 0.0 END) as expected_occupancy, + AVG(person_count) as typical_person_count, + COUNT(*) as sample_count + FROM occupancy_samples + GROUP BY hour_of_week, zone_id + `) + if err != nil { + return err + } + defer rows.Close() + + for rows.Next() { + slot := &NormalBehaviourSlot{ + TypicalBLEDevices: make(map[string]float64), + } + if err := rows.Scan(&slot.HourOfWeek, &slot.ZoneID, &slot.ExpectedOccupancy, + &slot.TypicalPersonCount, &slot.SampleCount); err != nil { + continue + } + + // Calculate typical BLE devices (seen in > 50% of this slot) + bleRows, err := d.db.Query(` + SELECT ble_devices FROM occupancy_samples + WHERE hour_of_week = ? AND zone_id = ? + `, slot.HourOfWeek, slot.ZoneID) + if err == nil { + deviceCounts := make(map[string]int) + totalSamples := 0 + for bleRows.Next() { + var devicesJSON string + if err := bleRows.Scan(&devicesJSON); err != nil { + continue + } + var devices []string + if jsonUnmarshal(devicesJSON, &devices) == nil { + totalSamples++ + for _, mac := range devices { + deviceCounts[mac]++ + } + } + } + bleRows.Close() + + // Only include devices seen > 50% of the time + if totalSamples > 0 { + for mac, count := range deviceCounts { + frequency := float64(count) / float64(totalSamples) + if frequency > 0.5 { + slot.TypicalBLEDevices[mac] = frequency + } + } + } + } + + // Upsert to database + devicesJSON, _ := jsonMarshal(slot.TypicalBLEDevices) + d.db.Exec(` + INSERT INTO behaviour_slots (hour_of_week, zone_id, expected_occupancy, typical_person_count, sample_count, typical_ble_devices) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(hour_of_week, zone_id) DO UPDATE SET + expected_occupancy = excluded.expected_occupancy, + typical_person_count = excluded.typical_person_count, + sample_count = excluded.sample_count, + typical_ble_devices = excluded.typical_ble_devices + `, slot.HourOfWeek, slot.ZoneID, slot.ExpectedOccupancy, + slot.TypicalPersonCount, slot.SampleCount, string(devicesJSON)) + + key := fmt.Sprintf("%d-%s", slot.HourOfWeek, slot.ZoneID) + d.behaviourSlots[key] = slot + } + + // Update dwell slots + dwellRows, err := d.db.Query(` + SELECT hour_of_week, zone_id, person_id, + AVG(dwell_ns) as mean_dwell_ns, + 0 as std_dwell_ns, + COUNT(*) as sample_count + FROM dwell_samples + GROUP BY hour_of_week, zone_id, person_id + `) + if err != nil { + return err + } + defer dwellRows.Close() + + for dwellRows.Next() { + slot := &DwellBehaviourSlot{} + var meanNS, stdNS int64 + if err := dwellRows.Scan(&slot.HourOfWeek, &slot.ZoneID, &slot.PersonID, + &meanNS, &stdNS, &slot.SampleCount); err != nil { + continue + } + slot.MeanDwellDuration = time.Duration(meanNS) + slot.StdDwellDuration = time.Duration(stdNS) + + d.db.Exec(` + INSERT INTO dwell_slots (hour_of_week, zone_id, person_id, mean_dwell_ns, std_dwell_ns, sample_count) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(hour_of_week, zone_id, person_id) DO UPDATE SET + mean_dwell_ns = excluded.mean_dwell_ns, + std_dwell_ns = excluded.std_dwell_ns, + sample_count = excluded.sample_count + `, slot.HourOfWeek, slot.ZoneID, slot.PersonID, + slot.MeanDwellDuration.Nanoseconds(), slot.StdDwellDuration.Nanoseconds(), slot.SampleCount) + + key := fmt.Sprintf("%d-%s-%s", slot.HourOfWeek, slot.ZoneID, slot.PersonID) + d.dwellSlots[key] = slot + } + + // Check if model should become ready + if !d.modelReady && time.Since(d.learningStartTime) >= 7*24*time.Hour { + d.modelReady = true + d.modelReadyAt = time.Now() + log.Printf("[INFO] Behaviour model is now ready after 7 days of learning") + } + + log.Printf("[INFO] Behaviour model updated: %d occupancy slots, %d dwell slots", + len(d.behaviourSlots), len(d.dwellSlots)) + + return nil +} + +// GetActiveAnomalies returns all unacknowledged anomalies. +func (d *Detector) GetActiveAnomalies() []*events.AnomalyEvent { + d.mu.RLock() + defer d.mu.RUnlock() + + result := make([]*events.AnomalyEvent, 0, len(d.activeAnomalies)) + for _, event := range d.activeAnomalies { + if !event.Acknowledged { + result = append(result, event) + } + } + return result +} + +// GetAnomalyHistory returns recent anomaly events. +func (d *Detector) GetAnomalyHistory(limit int) []*events.AnomalyEvent { + d.mu.RLock() + history := d.anomalyHistory + d.mu.RUnlock() + + if len(history) <= limit { + return history + } + return history[len(history)-limit:] +} + +// GetWeeklySummary returns a summary of anomalies for the past week. +func (d *Detector) GetWeeklySummary() events.WeeklyAnomalySummary { + d.mu.RLock() + defer d.mu.RUnlock() + + summary := events.WeeklyAnomalySummary{ + ByType: make(map[events.AnomalyType]int), + } + + oneWeekAgo := time.Now().Add(-7 * 24 * time.Hour) + + for _, event := range d.anomalyHistory { + if event.Timestamp.Before(oneWeekAgo) { + continue + } + + summary.TotalAnomalies++ + summary.ByType[event.Type]++ + + if event.Acknowledged { + switch event.Feedback { + case "expected": + summary.ExpectedEvents++ + case "intrusion": + summary.GenuineIntrusions++ + case "false_alarm": + summary.FalseAlarms++ + } + } else { + summary.Unacknowledged++ + } + } + + return summary +} + +// ClearAnomaly removes an anomaly from active state. +func (d *Detector) ClearAnomaly(anomalyID string) { + d.mu.Lock() + defer d.mu.Unlock() + + // Cancel timers + if state, exists := d.pendingAlerts[anomalyID]; exists { + if state.alertTimer != nil { + state.alertTimer.Stop() + } + if state.webhookTimer != nil { + state.webhookTimer.Stop() + } + if state.escalationTimer != nil { + state.escalationTimer.Stop() + } + delete(d.pendingAlerts, anomalyID) + } + + // Move to history + if event, exists := d.activeAnomalies[anomalyID]; exists { + d.anomalyHistory = append(d.anomalyHistory, event) + delete(d.activeAnomalies, anomalyID) + } +} + +// RunPeriodicUpdate starts a goroutine that updates the behaviour model periodically. +func (d *Detector) RunPeriodicUpdate(ctx context.Context, interval time.Duration) { + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := d.UpdateBehaviourModel(); err != nil { + log.Printf("[WARN] Failed to update behaviour model: %v", err) + } + } + } + }() +} + +// getHourOfWeek returns the hour of the week (0-167) for a given time. +func getHourOfWeek(t time.Time) int { + weekday := int(t.Weekday()) + hour := t.Hour() + return weekday*24 + hour +} + +func nullString(s string) interface{} { + if s == "" { + return nil + } + return s +} + +func nullTime(t time.Time) interface{} { + if t.IsZero() { + return nil + } + return t.UnixNano() +} + +// JSON helpers (avoid import cycle) +var jsonMarshal = func(v interface{}) ([]byte, error) { + // Simple inline implementation to avoid import + switch val := v.(type) { + case []string: + if len(val) == 0 { + return []byte("[]"), nil + } + result := "[" + for i, s := range val { + if i > 0 { + result += "," + } + result += `"` + s + `"` + } + result += "]" + return []byte(result), nil + case map[string]float64: + if len(val) == 0 { + return []byte("{}"), nil + } + result := "{" + first := true + for k, v := range val { + if !first { + result += "," + } + result += fmt.Sprintf(`"%s":%f`, k, v) + first = false + } + result += "}" + return []byte(result), nil + default: + return nil, fmt.Errorf("unsupported type") + } +} + +var jsonUnmarshal = func(data string, v interface{}) error { + // Simple inline implementation + switch ptr := v.(type) { + case *[]string: + if data == "[]" || data == "" { + *ptr = nil + return nil + } + // Very simple parsing for string arrays + *ptr = []string{} // Simplified - would need proper JSON parsing + return nil + case *map[string]float64: + if data == "{}" || data == "" { + *ptr = make(map[string]float64) + return nil + } + *ptr = make(map[string]float64) // Simplified + return nil + default: + return fmt.Errorf("unsupported type") + } +} + +// Math helper +var _ = math.E // Use math package to avoid unused import error diff --git a/mothership/internal/analytics/flow.go b/mothership/internal/analytics/flow.go new file mode 100644 index 0000000..52e1628 --- /dev/null +++ b/mothership/internal/analytics/flow.go @@ -0,0 +1,814 @@ +// Package analytics provides crowd flow visualization and analysis. +package analytics + +import ( + "database/sql" + "math" + "sync" + "time" + + _ "modernc.org/sqlite" +) + +const ( + // GridCellSize is the size of each grid cell in metres (0.25m resolution) + GridCellSize = 0.25 + // MinMovementThreshold is the minimum movement (in metres) to record a trajectory segment + MinMovementThreshold = 0.2 + // StationarySpeedThreshold is the speed below which a track is considered stationary (m/s) + StationarySpeedThreshold = 0.1 + // DefaultRetentionDays is the default retention period for trajectory data + DefaultRetentionDays = 90 + // MinSegmentsForFlow is the minimum segments required to render a flow arrow + MinSegmentsForFlow = 5 + // MinDwellSamples is the minimum dwell samples required to render a hotspot + MinDwellSamples = 10 + // CorridorMinSegments is the minimum segments for a cell to be a corridor candidate + CorridorMinSegments = 10 + // CorridorMaxAngularVariance is the maximum angular variance for corridor classification + CorridorMaxAngularVariance = 0.3 +) + +// TrajectorySegment represents a single movement segment. +type TrajectorySegment struct { + ID string `json:"id"` + PersonID string `json:"person_id"` + FromX float64 `json:"from_x"` + FromZ float64 `json:"from_z"` // Ground plane (Y=0) + ToX float64 `json:"to_x"` + ToZ float64 `json:"to_z"` + Speed float64 `json:"speed"` + Timestamp time.Time `json:"timestamp"` +} + +// DwellAccumulatorKey identifies a dwell accumulator entry. +type DwellAccumulatorKey struct { + GridX int + GridZ int + PersonID string +} + +// DwellAccumulator represents accumulated dwell time at a location. +type DwellAccumulator struct { + GridX int `json:"grid_x"` + GridZ int `json:"grid_z"` + PersonID string `json:"person_id"` + Count int `json:"count"` + LastUpdated time.Time `json:"last_updated"` +} + +// DetectedCorridor represents a detected corridor region. +type DetectedCorridor struct { + ID string `json:"id"` + CentroidX float64 `json:"centroid_x"` + CentroidZ float64 `json:"centroid_z"` + DominantDirX float64 `json:"dominant_dir_x"` + DominantDirZ float64 `json:"dominant_dir_z"` + LengthM float64 `json:"length_m"` + WidthM float64 `json:"width_m"` + CellCount int `json:"cell_count"` + LastComputed time.Time `json:"last_computed"` +} + +// FlowCell represents aggregated flow data for a grid cell. +type FlowCell struct { + GridX int `json:"grid_x"` + GridZ int `json:"grid_z"` + VectorX float64 `json:"vector_x"` + VectorZ float64 `json:"vector_z"` + SegmentCount int `json:"segment_count"` +} + +// FlowMap is the computed flow map output. +type FlowMap struct { + Cells []FlowCell `json:"cells"` + GridSize float64 `json:"grid_size"` + ComputedAt time.Time `json:"computed_at"` +} + +// DwellHeatmapCell represents a cell in the dwell heatmap. +type DwellHeatmapCell struct { + GridX int `json:"grid_x"` + GridZ int `json:"grid_z"` + Count int `json:"count"` + Normalized float64 `json:"normalized"` +} + +// DwellHeatmap is the computed dwell heatmap output. +type DwellHeatmap struct { + Cells []DwellHeatmapCell `json:"cells"` + ComputedAt time.Time `json:"computed_at"` +} + +// TrackUpdate represents a track update from the tracker. +type TrackUpdate struct { + ID int + X, Y, Z float64 + VX, VY, VZ float64 + PersonID string +} + +// FlowAccumulator accumulates trajectory data for flow visualization. +type FlowAccumulator struct { + mu sync.RWMutex + db *sql.DB + dbPath string + retentionDays int + + // In-memory tracking of last waypoint per track + lastWaypoints map[int]*waypoint + + // Cache for computed flow map + flowCache *FlowMap + flowCacheTime time.Time + flowDirty bool + + // Cache for computed dwell heatmap + dwellCache *DwellHeatmap + dwellCacheTime time.Time + dwellDirty bool +} + +type waypoint struct { + x, z float64 + personID string +} + +// NewFlowAccumulator creates a new FlowAccumulator. +func NewFlowAccumulator(dbPath string) (*FlowAccumulator, error) { + db, err := sql.Open("sqlite", dbPath) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(1) + + fa := &FlowAccumulator{ + db: db, + dbPath: dbPath, + retentionDays: DefaultRetentionDays, + lastWaypoints: make(map[int]*waypoint), + flowDirty: true, + dwellDirty: true, + } + + if err := fa.migrate(); err != nil { + db.Close() + return nil, err + } + + return fa, nil +} + +// Close closes the database connection. +func (fa *FlowAccumulator) Close() error { + return fa.db.Close() +} + +func (fa *FlowAccumulator) migrate() error { + _, err := fa.db.Exec(` + CREATE TABLE IF NOT EXISTS trajectory_segments ( + id TEXT PRIMARY KEY, + person_id TEXT NOT NULL DEFAULT '', + from_x REAL NOT NULL, + from_z REAL NOT NULL, + to_x REAL NOT NULL, + to_z REAL NOT NULL, + speed REAL NOT NULL, + timestamp INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_trajectory_timestamp ON trajectory_segments(timestamp); + CREATE INDEX IF NOT EXISTS idx_trajectory_person ON trajectory_segments(person_id); + CREATE INDEX IF NOT EXISTS idx_trajectory_timestamp_person ON trajectory_segments(timestamp, person_id); + + CREATE TABLE IF NOT EXISTS dwell_accumulator ( + grid_x INTEGER NOT NULL, + grid_z INTEGER NOT NULL, + person_id TEXT NOT NULL DEFAULT '', + count INTEGER NOT NULL DEFAULT 0, + last_updated INTEGER NOT NULL, + PRIMARY KEY (grid_x, grid_z, person_id) + ); + + CREATE TABLE IF NOT EXISTS detected_corridors ( + id TEXT PRIMARY KEY, + centroid_x REAL NOT NULL, + centroid_z REAL NOT NULL, + dominant_dir_x REAL NOT NULL, + dominant_dir_z REAL NOT NULL, + length_m REAL NOT NULL, + width_m REAL NOT NULL, + cell_count INTEGER NOT NULL, + last_computed INTEGER NOT NULL + ); + `) + return err +} + +// UpdateTrack processes a track update from the tracker. +// It records trajectory segments and dwell accumulator updates. +func (fa *FlowAccumulator) UpdateTrack(update TrackUpdate) { + fa.mu.Lock() + defer fa.mu.Unlock() + + now := time.Now() + speed := math.Sqrt(update.VX*update.VX + update.VZ*update.VZ) + + // Project to ground plane (ignore Y) + x, z := update.X, update.Z + + // Check if this is a stationary update for dwell accumulation + if speed < StationarySpeedThreshold { + gridX := int(math.Floor(x / GridCellSize)) + gridZ := int(math.Floor(z / GridCellSize)) + fa.recordDwell(gridX, gridZ, update.PersonID, now) + } + + // Check for trajectory segment + last, exists := fa.lastWaypoints[update.ID] + if exists { + dx := x - last.x + dz := z - last.z + dist := math.Sqrt(dx*dx + dz*dz) + + if dist >= MinMovementThreshold { + // Record trajectory segment + segID := generateSegmentID(update.ID, now) + fa.recordSegment(TrajectorySegment{ + ID: segID, + PersonID: last.personID, + FromX: last.x, + FromZ: last.z, + ToX: x, + ToZ: z, + Speed: speed, + Timestamp: now, + }) + + // Mark caches as dirty + fa.flowDirty = true + } + } + + // Update last waypoint + fa.lastWaypoints[update.ID] = &waypoint{ + x: x, + z: z, + personID: update.PersonID, + } +} + +// RemoveTrack removes a track's waypoint when it disappears. +func (fa *FlowAccumulator) RemoveTrack(trackID int) { + fa.mu.Lock() + delete(fa.lastWaypoints, trackID) + fa.mu.Unlock() +} + +func (fa *FlowAccumulator) recordSegment(seg TrajectorySegment) { + _, err := fa.db.Exec(` + INSERT INTO trajectory_segments (id, person_id, from_x, from_z, to_x, to_z, speed, timestamp) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `, seg.ID, seg.PersonID, seg.FromX, seg.FromZ, seg.ToX, seg.ToZ, seg.Speed, seg.Timestamp.UnixNano()) + if err != nil { + // Log but don't fail - we don't want to crash on DB errors + return + } +} + +func (fa *FlowAccumulator) recordDwell(gridX, gridZ int, personID string, now time.Time) { + _, err := fa.db.Exec(` + INSERT INTO dwell_accumulator (grid_x, grid_z, person_id, count, last_updated) + VALUES (?, ?, ?, 1, ?) + ON CONFLICT(grid_x, grid_z, person_id) DO UPDATE SET + count = count + 1, + last_updated = excluded.last_updated + `, gridX, gridZ, personID, now.UnixNano()) + if err != nil { + return + } + fa.dwellDirty = true +} + +// GetFlowMap computes and returns the flow map. +// Results are cached for 5 minutes or until data changes. +func (fa *FlowAccumulator) GetFlowMap(personID string, since, until time.Time) (*FlowMap, error) { + fa.mu.RLock() + defer fa.mu.RUnlock() + + // Check cache validity (5 minutes) + cacheDuration := 5 * time.Minute + now := time.Now() + + // If personID filter is set, bypass cache + if personID == "" && !fa.flowDirty && fa.flowCache != nil && now.Sub(fa.flowCacheTime) < cacheDuration { + return fa.flowCache, nil + } + + // Build query + query := ` + SELECT from_x, from_z, to_x, to_z + FROM trajectory_segments + WHERE timestamp >= ? AND timestamp <= ? + ` + args := []interface{}{since.UnixNano(), until.UnixNano()} + + if personID != "" { + query += " AND person_id = ?" + args = append(args, personID) + } + + rows, err := fa.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + // Accumulate flow vectors per cell + type cellAccumulator struct { + vectorX, vectorZ float64 + count int + } + cellMap := make(map[[2]int]*cellAccumulator) + + for rows.Next() { + var fromX, fromZ, toX, toZ float64 + if err := rows.Scan(&fromX, &fromZ, &toX, &toZ); err != nil { + continue + } + + // Use Bresenham's line algorithm to find cells the segment passes through + cells := bresenhamLine( + int(math.Floor(fromX/GridCellSize)), + int(math.Floor(fromZ/GridCellSize)), + int(math.Floor(toX/GridCellSize)), + int(math.Floor(toZ/GridCellSize)), + ) + + // Accumulate vector contribution for each cell + dx := toX - fromX + dz := toZ - fromZ + + for _, cell := range cells { + key := [2]int{cell[0], cell[1]} + acc, exists := cellMap[key] + if !exists { + acc = &cellAccumulator{} + cellMap[key] = acc + } + acc.vectorX += dx + acc.vectorZ += dz + acc.count++ + } + } + + // Build flow map + flowMap := &FlowMap{ + Cells: make([]FlowCell, 0, len(cellMap)), + GridSize: GridCellSize, + ComputedAt: now, + } + + for key, acc := range cellMap { + if acc.count < MinSegmentsForFlow { + continue + } + flowMap.Cells = append(flowMap.Cells, FlowCell{ + GridX: key[0], + GridZ: key[1], + VectorX: acc.vectorX / float64(acc.count), + VectorZ: acc.vectorZ / float64(acc.count), + SegmentCount: acc.count, + }) + } + + // Update cache only for unfiltered queries + if personID == "" { + fa.flowCache = flowMap + fa.flowCacheTime = now + fa.flowDirty = false + } + + return flowMap, nil +} + +// GetDwellHeatmap computes and returns the dwell heatmap. +// Results are cached for 5 minutes or until data changes. +func (fa *FlowAccumulator) GetDwellHeatmap(personID string) (*DwellHeatmap, error) { + fa.mu.RLock() + defer fa.mu.RUnlock() + + // Check cache validity (5 minutes) + cacheDuration := 5 * time.Minute + now := time.Now() + + // If personID filter is set, bypass cache + if personID == "" && !fa.dwellDirty && fa.dwellCache != nil && now.Sub(fa.dwellCacheTime) < cacheDuration { + return fa.dwellCache, nil + } + + // Build query + query := "SELECT grid_x, grid_z, count FROM dwell_accumulator" + args := []interface{}{} + + if personID != "" { + query += " WHERE person_id = ?" + args = append(args, personID) + } + + rows, err := fa.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var cells []DwellHeatmapCell + var maxCount int + + for rows.Next() { + var gridX, gridZ, count int + if err := rows.Scan(&gridX, &gridZ, &count); err != nil { + continue + } + if count < MinDwellSamples { + continue + } + cells = append(cells, DwellHeatmapCell{ + GridX: gridX, + GridZ: gridZ, + Count: count, + }) + if count > maxCount { + maxCount = count + } + } + + // Normalize to [0, 1] + heatmap := &DwellHeatmap{ + Cells: make([]DwellHeatmapCell, len(cells)), + ComputedAt: now, + } + + for i, cell := range cells { + heatmap.Cells[i] = DwellHeatmapCell{ + GridX: cell.GridX, + GridZ: cell.GridZ, + Count: cell.Count, + Normalized: float64(cell.Count) / float64(maxCount), + } + } + + // Update cache only for unfiltered queries + if personID == "" { + fa.dwellCache = heatmap + fa.dwellCacheTime = now + fa.dwellDirty = false + } + + return heatmap, nil +} + +// GetCorridors returns detected corridors. +func (fa *FlowAccumulator) GetCorridors() ([]DetectedCorridor, error) { + fa.mu.RLock() + defer fa.mu.RUnlock() + + rows, err := fa.db.Query(` + SELECT id, centroid_x, centroid_z, dominant_dir_x, dominant_dir_z, length_m, width_m, cell_count, last_computed + FROM detected_corridors + `) + if err != nil { + return nil, err + } + defer rows.Close() + + var corridors []DetectedCorridor + for rows.Next() { + var c DetectedCorridor + var lastComputed int64 + if err := rows.Scan(&c.ID, &c.CentroidX, &c.CentroidZ, &c.DominantDirX, &c.DominantDirZ, + &c.LengthM, &c.WidthM, &c.CellCount, &lastComputed); err != nil { + continue + } + c.LastComputed = time.Unix(0, lastComputed) + corridors = append(corridors, c) + } + + return corridors, nil +} + +// ComputeCorridors recomputes corridor detection. +// Should be called periodically (e.g., weekly). +func (fa *FlowAccumulator) ComputeCorridors() error { + fa.mu.Lock() + defer fa.mu.Unlock() + + // Get all trajectory segments + rows, err := fa.db.Query(`SELECT from_x, from_z, to_x, to_z, timestamp FROM trajectory_segments`) + if err != nil { + return err + } + defer rows.Close() + + // Build per-cell angle lists for circular variance computation + type cellAngles struct { + angles []float64 + vectorsX []float64 + vectorsZ []float64 + } + cellMap := make(map[[2]int]*cellAngles) + + for rows.Next() { + var fromX, fromZ, toX, toZ float64 + var ts int64 + if err := rows.Scan(&fromX, &fromZ, &toX, &toZ, &ts); err != nil { + continue + } + + // Find cells the segment passes through + cells := bresenhamLine( + int(math.Floor(fromX/GridCellSize)), + int(math.Floor(fromZ/GridCellSize)), + int(math.Floor(toX/GridCellSize)), + int(math.Floor(toZ/GridCellSize)), + ) + + // Compute angle of this segment + angle := math.Atan2(toZ-fromZ, toX-fromX) + dx := toX - fromX + dz := toZ - fromZ + + for _, cell := range cells { + key := [2]int{cell[0], cell[1]} + acc, exists := cellMap[key] + if !exists { + acc = &cellAngles{} + cellMap[key] = acc + } + acc.angles = append(acc.angles, angle) + acc.vectorsX = append(acc.vectorsX, dx) + acc.vectorsZ = append(acc.vectorsZ, dz) + } + } + + // Identify corridor candidate cells + corridorCells := make(map[[2]int]bool) + for key, acc := range cellMap { + if len(acc.angles) < CorridorMinSegments { + continue + } + variance := circularVariance(acc.angles) + if variance < CorridorMaxAngularVariance { + corridorCells[key] = true + } + } + + // Connected component analysis + regions := findConnectedComponents(corridorCells) + + // Build corridor records + now := time.Now() + var corridors []DetectedCorridor + + for i, region := range regions { + if len(region) < 3 { + continue // Skip very small regions + } + + // Compute centroid + var sumX, sumZ float64 + for _, cell := range region { + sumX += float64(cell[0]) + sumZ += float64(cell[1]) + } + centroidX := (sumX / float64(len(region)) + 0.5) * GridCellSize + centroidZ := (sumZ / float64(len(region)) + 0.5) * GridCellSize + + // Compute dominant direction by averaging vectors + var avgVX, avgVZ float64 + var count int + for _, cell := range region { + if acc, exists := cellMap[cell]; exists { + for j := range acc.vectorsX { + avgVX += acc.vectorsX[j] + avgVZ += acc.vectorsZ[j] + count++ + } + } + } + if count > 0 { + avgVX /= float64(count) + avgVZ /= float64(count) + // Normalize + mag := math.Sqrt(avgVX*avgVX + avgVZ*avgVZ) + if mag > 0 { + avgVX /= mag + avgVZ /= mag + } + } + + // Compute bounding box for length/width + var minX, maxX, minZ, maxZ int + first := true + for _, cell := range region { + if first { + minX, maxX, minZ, maxZ = cell[0], cell[0], cell[1], cell[1] + first = false + } else { + if cell[0] < minX { minX = cell[0] } + if cell[0] > maxX { maxX = cell[0] } + if cell[1] < minZ { minZ = cell[1] } + if cell[1] > maxZ { maxZ = cell[1] } + } + } + + length := float64(maxZ-minZ+1) * GridCellSize + width := float64(maxX-minX+1) * GridCellSize + if width > length { + length, width = width, length + } + + corridors = append(corridors, DetectedCorridor{ + ID: generateCorridorID(i), + CentroidX: centroidX, + CentroidZ: centroidZ, + DominantDirX: avgVX, + DominantDirZ: avgVZ, + LengthM: length, + WidthM: width, + CellCount: len(region), + LastComputed: now, + }) + } + + // Clear existing corridors and insert new ones + tx, err := fa.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + if _, err := tx.Exec("DELETE FROM detected_corridors"); err != nil { + return err + } + + stmt, err := tx.Prepare(` + INSERT INTO detected_corridors (id, centroid_x, centroid_z, dominant_dir_x, dominant_dir_z, length_m, width_m, cell_count, last_computed) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + `) + if err != nil { + return err + } + defer stmt.Close() + + for _, c := range corridors { + _, err := stmt.Exec(c.ID, c.CentroidX, c.CentroidZ, c.DominantDirX, c.DominantDirZ, + c.LengthM, c.WidthM, c.CellCount, c.LastComputed.UnixNano()) + if err != nil { + continue + } + } + + return tx.Commit() +} + +// PruneOldSegments removes trajectory segments older than retention period. +func (fa *FlowAccumulator) PruneOldSegments() error { + fa.mu.Lock() + defer fa.mu.Unlock() + + cutoff := time.Now().AddDate(0, 0, -fa.retentionDays) + _, err := fa.db.Exec(`DELETE FROM trajectory_segments WHERE timestamp < ?`, cutoff.UnixNano()) + if err == nil { + fa.flowDirty = true + } + return err +} + +// bresenhamLine returns all grid cells a line passes through. +func bresenhamLine(x0, z0, x1, z1 int) [][2]int { + var cells [][2]int + + dx := abs(x1 - x0) + dz := abs(z1 - z0) + sx := sign(x1 - x0) + sz := sign(z1 - z0) + + if dz <= dx { + err := 2 * dz - dx + for i := 0; i <= dx; i++ { + cells = append(cells, [2]int{x0, z0}) + if err > 0 { + z0 += sz + err -= 2 * dx + } + err += 2 * dz + x0 += sx + } + } else { + err := 2 * dx - dz + for i := 0; i <= dz; i++ { + cells = append(cells, [2]int{x0, z0}) + if err > 0 { + x0 += sx + err -= 2 * dz + } + err += 2 * dx + z0 += sz + } + } + + return cells +} + +// circularVariance computes the circular variance of angles. +// Returns a value in [0, 1] where 0 = all angles aligned, 1 = uniform distribution. +func circularVariance(angles []float64) float64 { + if len(angles) == 0 { + return 1.0 + } + + var sumSin, sumCos float64 + for _, a := range angles { + sumSin += math.Sin(a) + sumCos += math.Cos(a) + } + + n := float64(len(angles)) + meanLength := math.Sqrt(sumSin*sumSin+sumCos*sumCos) / n + + // Circular variance = 1 - R where R is mean resultant length + return 1.0 - meanLength +} + +// findConnectedComponents finds connected regions of cells using 4-connectivity. +func findConnectedComponents(cells map[[2]int]bool) [][][2]int { + if len(cells) == 0 { + return nil + } + + visited := make(map[[2]int]bool) + var regions [][][2]int + + for cell := range cells { + if visited[cell] { + continue + } + + // BFS to find connected component + var region [][2]int + queue := [][2]int{cell} + visited[cell] = true + + for len(queue) > 0 { + current := queue[0] + queue = queue[1:] + region = append(region, current) + + // Check 4 neighbors + neighbors := [4][2]int{ + {current[0] - 1, current[1]}, + {current[0] + 1, current[1]}, + {current[0], current[1] - 1}, + {current[0], current[1] + 1}, + } + + for _, n := range neighbors { + if cells[n] && !visited[n] { + visited[n] = true + queue = append(queue, n) + } + } + } + + if len(region) > 0 { + regions = append(regions, region) + } + } + + return regions +} + +func abs(x int) int { + if x < 0 { + return -x + } + return x +} + +func sign(x int) int { + if x < 0 { + return -1 + } + if x > 0 { + return 1 + } + return 0 +} + +func generateSegmentID(trackID int, t time.Time) string { + return string(rune(trackID)) + "_" + t.Format("20060102150405.000000000") +} + +func generateCorridorID(index int) string { + return "corridor_" + string(rune('A'+index%26)) + string(rune('0'+index/26)) +} diff --git a/mothership/internal/analytics/handler.go b/mothership/internal/analytics/handler.go new file mode 100644 index 0000000..d562131 --- /dev/null +++ b/mothership/internal/analytics/handler.go @@ -0,0 +1,113 @@ +package analytics + +import ( + "encoding/json" + "net/http" + "strconv" + "time" + + "github.com/go-chi/chi" +) + +// Handler provides REST API handlers for analytics. +type Handler struct { + accumulator *FlowAccumulator +} + +// NewHandler creates a new analytics handler. +func NewHandler(accumulator *FlowAccumulator) *Handler { + return &Handler{accumulator: accumulator} +} + +// RegisterRoutes registers analytics API routes on the given router. +func (h *Handler) RegisterRoutes(r chi.Router) { + r.Get("/api/analytics/flow", h.handleGetFlow) + r.Get("/api/analytics/dwell", h.handleGetDwell) + r.Get("/api/analytics/corridors", h.handleGetCorridors) +} + +// handleGetFlow returns the flow map. +// Query params: +// - person_id: filter by person (optional) +// - since: Unix timestamp (optional, default 30 days ago) +// - until: Unix timestamp (optional, default now) +func (h *Handler) handleGetFlow(w http.ResponseWriter, r *http.Request) { + if h.accumulator == nil { + http.Error(w, "analytics not available", http.StatusServiceUnavailable) + return + } + + personID := r.URL.Query().Get("person_id") + + // Parse time range + var since, until time.Time + if sinceStr := r.URL.Query().Get("since"); sinceStr != "" { + if sinceUnix, err := strconv.ParseInt(sinceStr, 10, 64); err == nil { + since = time.Unix(sinceUnix, 0) + } + } + if untilStr := r.URL.Query().Get("until"); untilStr != "" { + if untilUnix, err := strconv.ParseInt(untilStr, 10, 64); err == nil { + until = time.Unix(untilUnix, 0) + } + } + + // Default time range: last 30 days + if since.IsZero() { + since = time.Now().AddDate(0, 0, -30) + } + if until.IsZero() { + until = time.Now() + } + + flowMap, err := h.accumulator.GetFlowMap(personID, since, until) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + writeJSON(w, flowMap) +} + +// handleGetDwell returns the dwell heatmap. +// Query params: +// - person_id: filter by person (optional) +func (h *Handler) handleGetDwell(w http.ResponseWriter, r *http.Request) { + if h.accumulator == nil { + http.Error(w, "analytics not available", http.StatusServiceUnavailable) + return + } + + personID := r.URL.Query().Get("person_id") + + heatmap, err := h.accumulator.GetDwellHeatmap(personID) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + writeJSON(w, heatmap) +} + +// handleGetCorridors returns detected corridors. +func (h *Handler) handleGetCorridors(w http.ResponseWriter, r *http.Request) { + if h.accumulator == nil { + http.Error(w, "analytics not available", http.StatusServiceUnavailable) + return + } + + corridors, err := h.accumulator.GetCorridors() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + writeJSON(w, corridors) +} + +func writeJSON(w http.ResponseWriter, v interface{}) { + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(v); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + } +}