diff --git a/src/officesense_pi/src/config/config.ts b/src/officesense_pi/src/config/config.ts index b3d7b68..c1e2566 100644 --- a/src/officesense_pi/src/config/config.ts +++ b/src/officesense_pi/src/config/config.ts @@ -18,6 +18,12 @@ interface Config { port: number; password: string; }; + influx: { + url: string; + token: string; + org: string; + bucket: string; + }; core: { hysteresis: number; candidateHysteresis: number; @@ -49,13 +55,21 @@ const config: Config = { port: parseInt(process.env.REDIS_PORT ?? "6379"), password: process.env.REDIS_PASSWORD ?? "redispassword", }, + influx: { + url: process.env.INFLUX_URL ?? "http://10.24.4.13:8086", + token: process.env.INFLUX_TOKEN ?? "", + org: process.env.INFLUX_ORG ?? "officesense", + bucket: process.env.INFLUX_BUCKET ?? "officesense", + }, core: { hysteresis: parseInt(process.env.CORE_HYSTERESIS ?? "6"), candidateHysteresis: parseInt(process.env.CORE_CANDIDATE_HYSTERESIS ?? "5"), debounceMS: parseInt(process.env.CORE_DEBOUNCE_MS ?? String(3 * 1000)), minSamples: parseInt(process.env.CORE_MIN_SAMPLES ?? "4"), transitionTTL: parseInt(process.env.CORE_TRANSITION_TTL ?? String(5 * 60 * 1000)), - transitionCleanupInterval: parseInt(process.env.CORE_TRANSITION_CLEANUP_INTERVAL ?? String(60 * 1000)), + transitionCleanupInterval: parseInt( + process.env.CORE_TRANSITION_CLEANUP_INTERVAL ?? String(60 * 1000) + ), lossThreshold: parseInt(process.env.CORE_LOSS_THRESHOLD ?? String(8 * 1000)), userTTL: parseInt(process.env.CORE_USER_TTL ?? String(3 * 60 * 1000)), verifyTimeout: parseInt(process.env.CORE_VERIFY_TIMEOUT ?? String(5 * 60 * 1000)), @@ -63,4 +77,4 @@ const config: Config = { }, }; -export default config; \ No newline at end of file +export default config; diff --git a/src/officesense_pi/src/core/analyze.ts b/src/officesense_pi/src/core/analyze.ts index 3d1ca7a..7b069c7 100644 --- a/src/officesense_pi/src/core/analyze.ts +++ b/src/officesense_pi/src/core/analyze.ts @@ -3,15 +3,45 @@ import { prisma } from "../lib/prisma.js"; import { RoomTransition, transitions } from "./transition.js"; import config from "../config/config.js"; import { type RedisUserData } from "../api/livedata/livedata.repository.js"; +import { getInflux } from "../influx/influx.js"; +import { Point } from "@influxdata/influxdb-client"; const adjectives = ["Crazy", "Silent", "Dark", "Fast", "Lucky", "Wild", "Epic"]; - const nouns = ["Tiger", "Wolf", "Falcon", "Shadow", "Ninja", "Dragon", "Phoenix"]; const running = new Set(); const userRooms = new Map(); const userPseudo = new Map(); +async function writeRoomEvent( + event: "enter" | "leave", + pseudoID: string, + roomID: string, + rssi: number +) { + try { + const room = await prisma.room.findUnique({ + where: { id: roomID }, + select: { name: true }, + }); + + const roomName = room?.name ?? roomID; + + const point = new Point("room_events") + .tag("pseudo_id", pseudoID) + .tag("room_id", roomID) + .tag("room_name", roomName) + .tag("event", event) + .floatField("rssi", rssi); + + getInflux().writePoint(point); + + console.log(`[InfluxDB] Wrote '${event}' event for pseudo ${pseudoID} in room ${roomName}`); + } catch (err) { + console.error("[InfluxDB] Failed to write event:", err); + } +} + export async function updateRoomOccupancyListener(message: string, channel: string) { console.log(`[Redis] Key ${message} expired.`); @@ -21,10 +51,11 @@ export async function updateRoomOccupancyListener(message: string, channel: stri const redis = getRedis(); - if (!lastRoomID) { - console.log("[!] Key not found in local map"); - } else { + if (!lastRoomID) console.log("[!] Key not found in local map"); + else { await redis.decr(`room:${lastRoomID}`); + + if (userPseudoID) await writeRoomEvent("leave", userPseudoID, lastRoomID, 0); } await redis.del(`_user:${userPseudoID}`); @@ -35,11 +66,8 @@ export async function updateRoomOccupancyListener(message: string, channel: stri function generateNickname() { const adjective = adjectives[Math.floor(Math.random() * adjectives.length)]; - const noun = nouns[Math.floor(Math.random() * nouns.length)]; - const number = Math.floor(Math.random() * 1000); - return `${adjective}${noun}${number}`; } @@ -103,14 +131,14 @@ export async function analyzeData( ); await redis.set(`_user:${user.pseudoID}`, userID); - await redis.incr(`room:${roomID}`); userRooms.set(userID, roomID); userPseudo.set(userID, user.pseudoID); - console.log("[Redis] Stored new user in redis."); + await writeRoomEvent("enter", user.pseudoID, roomID, metrics.rssi); + console.log("[Redis] Stored new user in redis."); return; } @@ -123,7 +151,6 @@ export async function analyzeData( resObj["timestamp"] = Date.now(); await redis.set(`user:${userID}`, JSON.stringify(resObj), { PX: config.core.userTTL }); - return; } @@ -140,7 +167,10 @@ export async function analyzeData( ) { console.log(`[!] Transition done ${resObj["room"]} -> ${roomID}`); - await redis.decr(`room:${resObj["room"]}`); + const previousRoomID = resObj["room"]; + const pseudoID = resObj["userID"]; + + await redis.decr(`room:${previousRoomID}`); await redis.incr(`room:${roomID}`); userRooms.set(userID, roomID); @@ -151,7 +181,12 @@ export async function analyzeData( resObj["verified"] = false; await redis.set(`user:${userID}`, JSON.stringify(resObj), { PX: config.core.userTTL }); - } else console.log("[!] Transition declined"); + + await writeRoomEvent("leave", pseudoID, previousRoomID, metrics.rssi); + await writeRoomEvent("enter", pseudoID, roomID, metrics.rssi); + } else { + console.log("[!] Transition declined"); + } } finally { running.delete(metrics.tagID); } diff --git a/src/officesense_pi/src/core/bootstrap.ts b/src/officesense_pi/src/core/bootstrap.ts index 3d115f0..de5ac6c 100644 --- a/src/officesense_pi/src/core/bootstrap.ts +++ b/src/officesense_pi/src/core/bootstrap.ts @@ -3,6 +3,7 @@ import { initRedis, initSubRedis } from "../redis/redis.js"; import { initCamera } from "./camera.js"; import * as mqtt from "./mqtt.js"; import { cleanupWorker } from "./transition.js"; +import { initInflux } from "../influx/influx.js"; export async function bootstrap() { try { @@ -17,6 +18,8 @@ export async function bootstrap() { await initSubRedis(); await mqtt.start(); + initInflux(); + cleanupWorker(); await initCamera(); diff --git a/src/officesense_pi/src/influx/influx.ts b/src/officesense_pi/src/influx/influx.ts new file mode 100644 index 0000000..ff286aa --- /dev/null +++ b/src/officesense_pi/src/influx/influx.ts @@ -0,0 +1,22 @@ +import { InfluxDB, type WriteApi } from "@influxdata/influxdb-client"; +import config from "../config/config.js"; + +let writeApi: WriteApi | null = null; + +export function initInflux(): void { + const client = new InfluxDB({ + url: config.influx.url, + token: config.influx.token, + }); + + writeApi = client.getWriteApi(config.influx.org, config.influx.bucket, "ms"); + + writeApi.useDefaultTags({ host: "officesense-pi" }); + + console.log("[InfluxDB] Write API initialized."); +} + +export function getInflux(): WriteApi { + if (!writeApi) throw new Error("[InfluxDB] Not initialized. Call initInflux() first."); + return writeApi; +}