room occupancy impl & race cond fix & local user roomID and psuedoID storage for cleanup and indexing

Core dev

progress.md

@@ -45,3 +45,15 @@
 
 ### Next Steps
 - WiFi self-healing and MQTT reconnection & testing
+
+## Week 4
+### Completed
+- Tag & Scanner CLI for configuration & persistent config
+- Core script (MQTT, location estimation, Store in Redis)
+
+### In Progress
+- NGSI-LD data for public API
+- Admin API with CRUD functions
+
+### Next Steps
+- Add verification with camera in core script

src/officesense_pi/src/api/livedata/livadata.routes.ts

@@ -0,0 +1,10 @@
+import express from "express";
+import { getEntities, getEntitiesOfType, getEntity } from "./livedata.controller.js";
+
+const router = express.Router();
+
+router.get("/entities", getEntities);
+router.get("/entities", getEntitiesOfType);
+router.get("/entities/:urn", getEntity);
+
+export default router;

src/officesense_pi/src/api/livedata/livedata.controller.ts

@@ -0,0 +1,172 @@
+import type { Request, Response, NextFunction } from "express";
+import { getActiveUserData, getRedisRoomData, type RedisRoomData, type RedisUserData } from "./livedata.repository.js";
+import { prisma } from "../../lib/prisma.js";
+
+interface Params {
+    urn: string;
+}
+
+type Relationship = {
+    type: "Relationship";
+    object: string;
+};
+
+type Property<T = any> = {
+    type: "Property";
+    value: T;
+};
+
+interface Entity {
+    id: string;
+    type: string;
+    name: {
+        type: "Property";
+        value: string;
+    };
+}
+
+interface User extends Entity {
+    type: "User";
+    locatedIn: Relationship;
+    rssi: Property<number>;
+    authenticationStatus: Property<string>;
+    observedAt: Property<string>;
+}
+
+interface Room extends Entity {
+    type: "Room";
+    occupancy: Property<number>;
+}
+
+function convertToNGSIUser(user: RedisUserData): User {
+    return {
+        id: `urn:ngsi-ld:User:${user.userID}`,
+        type: "User",
+        name: {
+            type: "Property",
+            value: user.name
+        },
+        locatedIn: {
+            type: "Relationship",
+            object: user.room
+        },
+        rssi: {
+            type: "Property",
+            value: user.rssi,
+        },
+        authenticationStatus: {
+            type: "Property",
+            value: user.verified ? "verified" : "unverified"
+        },
+        observedAt: {
+            type: "Property",
+            value: (new Date(user.timestamp)).toISOString()
+        }
+    }
+}
+
+async function convertToNGSIRoom(room: RedisRoomData): Promise<Room | null> {
+    const roomName = await prisma.room.findUnique({
+        where: { id: room.roomID },
+        select: { name: true }
+    })
+
+    if (!roomName?.name) return null;
+
+    return {
+        id: `urn:ngsi-ld:Room:${room.roomID}`,
+        type: "Room",
+        name: {
+            type: "Property",
+            value: roomName.name
+        },
+        occupancy: {
+            type: "Property",
+            value: room.occupancy
+        }
+    }
+}
+
+async function getNGSIUsers(): Promise<User[]> {
+    let users: User[] = [];
+
+    const result: RedisUserData[] = await getActiveUserData();
+
+    for (const r of result) {
+        const user = convertToNGSIUser(r);
+
+        users.push(user);
+    }
+
+    return users;
+}
+
+async function getNGSIRooms(): Promise<Room[]> {
+    let rooms: Room[] = [];
+
+    const result: RedisRoomData[] = await getRedisRoomData();
+
+    for (const r of result) {
+        const room = await convertToNGSIRoom(r);
+        if (room == null) continue;
+
+        rooms.push(room);
+    }
+
+    return rooms;
+}
+
+export async function getEntities(req: Request, res: Response, next: NextFunction) {
+    if (Object.keys(req.query).length > 0) return next();
+
+    let entities: Entity[] = [];
+
+    entities.push(...await getNGSIUsers());
+    entities.push(...await getNGSIRooms());
+
+    return res.status(200).send(entities);
+}
+
+export async function getEntitiesOfType(req: Request, res: Response) {
+    const { type } = req.query;
+
+    switch (type) {
+        case "user": return res.status(200).send(await getNGSIUsers());
+        case "room": return res.status(200).send(await getNGSIRooms());
+        default:
+            return res.sendStatus(400);
+    }
+}
+
+export async function getEntity(req: Request<Params>, res: Response) {
+    const { urn } = req.params;
+
+    if (!urn || !urn.startsWith("urn:ngsi-ld:")) return res.sendStatus(400);
+
+    const entity = urn.slice(12);
+
+    if (!entity) return res.sendStatus(400);
+
+    if (entity.toLowerCase().startsWith("room:")) {
+        const roomID = entity.slice(5);
+
+        if (!roomID) return res.sendStatus(400);
+
+        const data = await getRedisRoomData(roomID);
+
+        if (!data.length) return res.sendStatus(404);
+
+        return res.status(200).send(await convertToNGSIRoom(data[0]!));
+    } else if (entity.toLowerCase().startsWith("user:")) {
+        const userID = entity.slice(5);
+
+        if (!userID) return res.sendStatus(400);
+
+        const data = await getActiveUserData(userID);
+
+        if (!data.length) return res.sendStatus(404);
+
+        return res.status(200).send(convertToNGSIUser(data[0]!));
+    } else
+        return res.sendStatus(400);
+}

src/officesense_pi/src/api/livedata/livedata.repository.ts

@@ -0,0 +1,106 @@
+import { prisma } from "../../lib/prisma.js";
+import { getRedis } from "../../redis/redis.js";
+
+export interface RedisUserData {
+    userID: string;
+    name: string;
+    rssi: number;
+    room: string;
+    verified: boolean;
+    timestamp: number;
+}
+
+export interface RedisRoomData {
+    roomID: string;
+    occupancy: number;
+}
+
+export async function getActiveUserData(userID?: string): Promise<RedisUserData[]> {
+    const redis = getRedis();
+
+    if (!userID) {
+        let cursor = "0";
+        let result: RedisUserData[] = [];
+
+        do {
+            const res = await redis.scan(cursor, {
+                MATCH: "user:*",
+                COUNT: 100,
+            });
+
+            cursor = res.cursor;
+
+            const values = await Promise.all(
+                res.keys.map((key) => redis.get(key))
+            );
+
+            result.push(...values
+                .filter((v): v is string => v !== null)
+                .map((v) => JSON.parse(v))
+            );
+
+        } while (cursor !== "0");
+
+        return result;
+    }
+
+    const _userID = await redis.get(`_user:${userID}`);
+
+    if (!_userID) return [];
+
+    const res = await redis.get(`user:${_userID}`);
+
+    if (!res) return [];
+
+    const resObj: RedisUserData = JSON.parse(res);
+
+    return [resObj];
+}
+
+export async function getRedisRoomData(roomID?: string): Promise<RedisRoomData[]> {
+    const redis = getRedis();
+
+    if (!roomID) {
+        let cursor = "0";
+        let result: RedisRoomData[] = [];
+
+        do {
+            const res = await redis.scan(cursor, {
+                MATCH: "room:*",
+                COUNT: 100,
+            });
+
+            cursor = res.cursor;
+
+            const values = await Promise.all(
+                res.keys.map(async (key) => ({
+                    roomID: key,
+                    occupancy: await redis.get(key),
+                }))
+            );
+
+            result.push(...values
+                .filter((v) => v.occupancy !== null)
+                .map((v) => ({
+                    roomID: v.roomID.slice(5),
+                    occupancy: Number(v.occupancy),
+                }))
+            );
+
+        } while (cursor !== "0");
+
+        return result;
+    }
+
+    const res = await redis.get(`room:${roomID}`);
+
+    if (!res) return [];
+
+    return [{ roomID: roomID, occupancy: Number(res) }];
+}
+
+export async function getRoomIDs(): Promise<{ id: string; }[]> {
+    return await prisma.room.findMany({
+        select: { id: true }
+    });
+}

src/officesense_pi/src/core/analyze.ts

@@ -1,7 +1,8 @@
-import { getRedis } from "./redis.js";
+import { getRedis } from "../redis/redis.js";
 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";
 
 const adjectives = [
     "Crazy",
@@ -23,6 +24,31 @@ const nouns = [
     "Phoenix"
 ];
 
+const running = new Set<string>();
+const userRooms = new Map<string, string>();
+const userPseudo = new Map<string, string>();
+
+export async function updateRoomOccupancyListener(message: string, channel: string) {
+    console.log(`[Redis] Key ${message} expired.`);
+
+    const id = message.startsWith("user:") ? message.slice(5) : message;
+    const lastRoomID = userRooms.get(id);
+    const userPseudoID = userPseudo.get(id);
+
+    const redis = getRedis();
+
+    if (!lastRoomID) {
+        console.log("[!] Key not found in local mapι")
+    } else {
+        await redis.decr(`room:${lastRoomID}`);
+    }
+
+    await redis.del(`_user:${userPseudoID}`);
+
+    userRooms.delete(id);
+    userPseudo.delete(id);
+}
+
 function generateNickname() {
     const adjective =
         adjectives[Math.floor(Math.random() * adjectives.length)];
@@ -46,77 +72,102 @@ export async function analyzeData(roomID: string, metrics: {
     tagID: string,
     rssi: number
 }) {
-    console.log(`\n=======================================================\n` +
-        `ROOM_ID: ${roomID}\nTAG_ID: ${metrics.tagID}\nRSSI: ${metrics.rssi}\n` +
-        `=======================================================\n`);
-
-    const userID = (
-        await prisma.tag.findUnique({
-            where: { id: metrics.tagID },
-            select: { userId: true },
-        })
-    )?.userId;
-
-    console.log(`[SQLite] Resolved USER_ID: ${userID}`);
-
-    if (!userID)
-        return;
-
-    const redis = getRedis();
-    const res = await redis.get(`user:${userID}`);
-
-    if (!res) {
-        const user = newUser();
-        console.log(`[Redis] Created USER: ${user.psuedoName} USER_ID: ${user.pseudoID}`);
-
-        // trigger camera
-
-        await redis.set(`user:${userID}`, JSON.stringify({
-            userID: user.pseudoID,
-            name: user.psuedoName,
-            rssi: metrics.rssi,
-            room: roomID,
-            verified: false,
-            timestamp: Date.now()
-        }), { PX: config.core.userTTL });
-
-        console.log("[Redis] Stored new user in redis.");
-
+    if (running.has(metrics.tagID)) {
+        console.log(`[!] Skipping TAG_ID: ${metrics.tagID} already in process.`);
         return;
     }
 
-    let resObj = JSON.parse(res);
-    console.log(`[Redis] User exists in redis in ROOM_ID: ${resObj["room"]}`);
+    running.add(metrics.tagID);
 
-    if (roomID == resObj["room"]) {
-        console.log("[Redis] Same room, updating redis...");
-        resObj["rssi"] = metrics.rssi;
-        resObj["timestamp"] = Date.now()
+    try {
+        console.log(`\n=======================================================\n` +
+            `ROOM_ID: ${roomID}\nTAG_ID: ${metrics.tagID}\nRSSI: ${metrics.rssi}\n` +
+            `=======================================================\n`);
 
-        await redis.set(`user:${userID}`, JSON.stringify(resObj), { PX: config.core.userTTL });
+        const userID = (
+            await prisma.tag.findUnique({
+                where: { id: metrics.tagID },
+                select: { userId: true },
+            })
+        )?.userId;
 
-        return;
+        console.log(`[SQLite] Resolved USER_ID: ${userID}`);
+
+        if (!userID)
+            return;
+
+        const redis = getRedis();
+        const res = await redis.get(`user:${userID}`);
+
+        if (!res) {
+            const user = newUser();
+            console.log(`[Redis] Created USER: ${user.psuedoName} USER_ID: ${user.pseudoID}`);
+
+            // trigger camera
+
+            await redis.set(`user:${userID}`, JSON.stringify({
+                userID: user.pseudoID,
+                name: user.psuedoName,
+                rssi: metrics.rssi,
+                room: roomID,
+                verified: false,
+                timestamp: Date.now()
+            }), { PX: config.core.userTTL });
+
+            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.");
+
+            return;
+        }
+
+        let resObj: RedisUserData = JSON.parse(res);
+        console.log(`[Redis] User exists in redis in ROOM_ID: ${resObj["room"]}`);
+
+        if (roomID == resObj["room"]) {
+            console.log("[Redis] Same room, updating redis...");
+            resObj["rssi"] = metrics.rssi;
+            resObj["timestamp"] = Date.now()
+
+            await redis.set(`user:${userID}`, JSON.stringify(resObj), { PX: config.core.userTTL });
+
+            return;
+        }
+
+        let transition = transitions.get(userID);
+
+        // update room occupancy
+
+        if (!transition) {
+            console.log("[!] Starting new transition");
+            transition = new RoomTransition();
+            transitions.set(userID, transition);
+        }
+
+        if (transition.shouldTransitionTo(roomID, metrics.rssi, resObj["rssi"], resObj["timestamp"])) {
+            // trigger camera
+
+            console.log(`[!] Transition done ${resObj["room"]} -> ${roomID}`);
+
+            await redis.decr(`room:${resObj["room"]}`);
+            await redis.incr(`room:${roomID}`);
+
+            userRooms.set(userID, roomID);
+
+            resObj["rssi"] = metrics.rssi;
+            resObj["room"] = roomID;
+            resObj["timestamp"] = Date.now()
+
+
+            await redis.set(`user:${userID}`, JSON.stringify(resObj), { PX: config.core.userTTL });
+        } else
+            console.log("[!] Transition declined");
+    } finally {
+        running.delete(metrics.tagID);
     }
-
-    let transition = transitions.get(userID);
-
-    if (!transition) {
-        console.log("[!] Starting new transition");
-        transition = new RoomTransition();
-        transitions.set(userID, transition);
-    }
-
-    if (transition.shouldTransitionTo(roomID, metrics.rssi, resObj["rssi"], resObj["timestamp"])) {
-        // trigger camera
-
-        console.log(`[!] Transition done ${resObj["room"]} -> ${roomID}`);
-
-        resObj["rssi"] = metrics.rssi;
-        resObj["room"] = roomID;
-        resObj["timestamp"] = Date.now()
-
-
-        await redis.set(`user:${userID}`, JSON.stringify(resObj), { PX: config.core.userTTL });
-    } else
-        console.log("[!] Transition declined");
 }

src/officesense_pi/src/core/bootstrap.ts

@@ -1,16 +1,27 @@
-import { initRedis } from "./redis.js";
+import { getRoomIDs } from "../api/livedata/livedata.repository.js";
+import { initRedis, initSubRedis } from "../redis/redis.js";
 import * as mqtt from "./mqtt.js";
 import { cleanupWorker } from "./transition.js";
 
 export async function bootstrap() {
     try {
-        await initRedis();
+        const redis = await initRedis();
+
+        await redis.flushDb();
+        console.log(`[!] Flushed Redis.`);
+
+        const rooms = await getRoomIDs();
+        for (const { id } of rooms)
+            await redis.set(`room:${id}`, 0);
+
+        await initSubRedis();
         await mqtt.start();
+
         cleanupWorker();
 
-        console.log("Core started.");
+        console.log("[!] Core started.");
     } catch (err: any) {
-        console.log("Bootstrap failed:", err.message);
+        console.log("[!] Bootstrap failed:", err.message);
         process.exit(1);
     }
 }

src/officesense_pi/src/core/mqtt.ts

@@ -13,13 +13,13 @@ function initMqtt(): Promise<MqttClient> {
         );
 
         client.once("connect", () => {
-            console.log("MQTT connected.");
+            console.log("[MQTT] MQTT connected.");
 
             client.subscribe(config.mqtt.topic, (err) => {
                 if (err) {
                     reject(err);
                 } else {
-                    console.log("MQTT subscribed.");
+                    console.log("[MQTT] MQTT subscribed.");
                     resolve(client);
                 }
             });
@@ -43,6 +43,6 @@ export async function start() {
             console.log(err.message);
         })
     } catch (err: any) {
-        console.log("MQTT connection failed:", err.message);
+        console.log("[MQTT] MQTT connection failed:", err.message);
     }
 }

src/officesense_pi/src/core/redis.ts

@@ -1,28 +0,0 @@
-import { createClient, type RedisClientType } from "redis";
-import config from "../config/config.js";
-
-let redisClient: RedisClientType;
-
-export async function initRedis(): Promise<RedisClientType> {
-    redisClient = createClient({
-        url: `redis://${config.redis.host}:${config.redis.port}`
-    });
-
-    redisClient.on("error", (err) => {
-        console.log("Redis error:", err.message);
-    });
-
-    await redisClient.connect();
-
-    console.log("Redis connected.");
-
-    return redisClient;
-}
-
-export function getRedis(): RedisClientType {
-    if (!redisClient) {
-        throw new Error("Redis not initialized");
-    }
-
-    return redisClient;
-}

src/officesense_pi/src/core/transition.ts

@@ -72,7 +72,7 @@ export class RoomTransition {
 
         const enoughTime = now - (this.candidate.since ?? now) >= config.core.debounceMS;
         const enoughConfirmations = this.candidate.samples >= config.core.minSamples;
-        
+
         console.log(`[TRANSITION] enoughTime = ${enoughTime}`);
         console.log(`[TRANSITION] enoughConfirmations = ${enoughConfirmations}`);
         console.log(`[TRANSITION] signalLost = ${signalLost}`);

src/officesense_pi/src/index.ts

@@ -1,12 +1,14 @@
 import express from "express";
 import lookupRoutes from "./api/lookup/lookup.routes.js";
+import livedataRoutes from "./api/livedata/livadata.routes.js"
 import config from "./config/config.js";
 import { bootstrap } from "./core/bootstrap.js";
 
 (async () => {
     const app = express();
 
-    app.use("/", lookupRoutes);
+    app.use("/lookup", lookupRoutes);
+    app.use("/livedata", livedataRoutes);
 
     app.listen(config.api.port, config.api.address, () => {
         console.log(`Server listening on ${config.api.address}:${config.api.port}`);

src/officesense_pi/src/redis/redis.ts

@@ -0,0 +1,48 @@
+import { createClient, type RedisClientType } from "redis";
+import config from "../config/config.js";
+import { updateRoomOccupancyListener } from "../core/analyze.js";
+
+let redisClient: RedisClientType;
+let subRedisClient: RedisClientType;
+
+export async function initSubRedis(): Promise<void> {
+    subRedisClient = createClient({
+        url: `redis://${config.redis.host}:${config.redis.port}`
+    });
+
+    subRedisClient.on("error", (err) => {
+        console.log("[Redis] SubRedis error:", err.message);
+    });
+
+    await subRedisClient.connect();
+
+    console.log("[Redis] SubRedis connected.");
+
+    subRedisClient.pSubscribe("__keyevent@0__:expired", updateRoomOccupancyListener);
+
+    console.log("[Redis] SubRedis listener set.");
+}
+
+export async function initRedis(): Promise<RedisClientType> {
+    redisClient = createClient({
+        url: `redis://${config.redis.host}:${config.redis.port}`
+    });
+
+    redisClient.on("error", (err) => {
+        console.log("[Redis] Redis error:", err.message);
+    });
+
+    await redisClient.connect();
+
+    console.log("[Redis] Redis connected.");
+
+    return redisClient;
+}
+
+export function getRedis(): RedisClientType {
+    if (!redisClient) {
+        throw new Error("[Redis] Redis not initialized");
+    }
+
+    return redisClient;
+}