Merge pull request #15 from dkwstas/public_api

Public api

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,6 +72,14 @@ export async function analyzeData(roomID: string, metrics: {
     tagID: string,
     rssi: number
 }) {
+    if (running.has(metrics.tagID)) {
+        console.log(`[!] Skipping TAG_ID: ${metrics.tagID} already in process.`);
+        return;
+    }
+
+    running.add(metrics.tagID);
+
+    try {
         console.log(`\n=======================================================\n` +
             `ROOM_ID: ${roomID}\nTAG_ID: ${metrics.tagID}\nRSSI: ${metrics.rssi}\n` +
             `=======================================================\n`);
@@ -80,12 +114,19 @@ export async function analyzeData(roomID: string, metrics: {
                 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 = JSON.parse(res);
+        let resObj: RedisUserData = JSON.parse(res);
         console.log(`[Redis] User exists in redis in ROOM_ID: ${resObj["room"]}`);
 
         if (roomID == resObj["room"]) {
@@ -100,6 +141,8 @@ export async function analyzeData(roomID: string, metrics: {
 
         let transition = transitions.get(userID);
 
+        // update room occupancy
+
         if (!transition) {
             console.log("[!] Starting new transition");
             transition = new RoomTransition();
@@ -111,6 +154,11 @@ export async function analyzeData(roomID: string, metrics: {
 
             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()
@@ -119,4 +167,7 @@ export async function analyzeData(roomID: string, metrics: {
             await redis.set(`user:${userID}`, JSON.stringify(resObj), { PX: config.core.userTTL });
         } else
             console.log("[!] Transition declined");
+    } finally {
+        running.delete(metrics.tagID);
+    }
 }

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