influxdb implementation

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;
+export default config;

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<string>();
 const userRooms = new Map<string, string>();
 const userPseudo = new Map<string, string>();
 
+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) {
+    if (!lastRoomID) console.log("[!] Key not found in local map");
-        console.log("[!] Key not found in local map");
+    else {
-    } 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);
     }

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();

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;
+}