From 124ef542b0b9f2879acc4433de44121207f36f56 Mon Sep 17 00:00:00 2001 From: Kostas Drakontidis Date: Mon, 25 May 2026 05:53:53 +0300 Subject: [PATCH 1/7] moved redis & added subredis for expire notifications --- src/officesense_pi/src/core/redis.ts | 28 ---------------- src/officesense_pi/src/redis/redis.ts | 48 +++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 28 deletions(-) delete mode 100644 src/officesense_pi/src/core/redis.ts create mode 100644 src/officesense_pi/src/redis/redis.ts diff --git a/src/officesense_pi/src/core/redis.ts b/src/officesense_pi/src/core/redis.ts deleted file mode 100644 index c9b1944..0000000 --- a/src/officesense_pi/src/core/redis.ts +++ /dev/null @@ -1,28 +0,0 @@ -import { createClient, type RedisClientType } from "redis"; -import config from "../config/config.js"; - -let redisClient: RedisClientType; - -export async function initRedis(): Promise { - 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; -} \ No newline at end of file diff --git a/src/officesense_pi/src/redis/redis.ts b/src/officesense_pi/src/redis/redis.ts new file mode 100644 index 0000000..d9826d7 --- /dev/null +++ b/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 { + 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 { + 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; +} \ No newline at end of file From 93c65a948b3973977bcb8d0cf93d1a23ee5afc6b Mon Sep 17 00:00:00 2001 From: Kostas Drakontidis Date: Mon, 25 May 2026 05:54:17 +0300 Subject: [PATCH 2/7] added routes --- src/officesense_pi/src/index.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/officesense_pi/src/index.ts b/src/officesense_pi/src/index.ts index e0e57d8..4f9f1a0 100644 --- a/src/officesense_pi/src/index.ts +++ b/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}`); From c8da529ce3dac895e200416506a5410c11b2d362 Mon Sep 17 00:00:00 2001 From: Kostas Drakontidis Date: Mon, 25 May 2026 05:54:25 +0300 Subject: [PATCH 3/7] whitespace --- src/officesense_pi/src/core/transition.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/officesense_pi/src/core/transition.ts b/src/officesense_pi/src/core/transition.ts index f7c33d0..7be7f29 100644 --- a/src/officesense_pi/src/core/transition.ts +++ b/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}`); From 66ab3fa6ca33b486632d6c94736efce90c75624b Mon Sep 17 00:00:00 2001 From: Kostas Drakontidis Date: Mon, 25 May 2026 05:54:37 +0300 Subject: [PATCH 4/7] added debug prefix --- src/officesense_pi/src/core/mqtt.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/officesense_pi/src/core/mqtt.ts b/src/officesense_pi/src/core/mqtt.ts index 11fe6a3..747a1f4 100644 --- a/src/officesense_pi/src/core/mqtt.ts +++ b/src/officesense_pi/src/core/mqtt.ts @@ -13,13 +13,13 @@ function initMqtt(): Promise { ); 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); } } From c2b947a197d2fbe6cca63e7223ab4d67492d8f21 Mon Sep 17 00:00:00 2001 From: Kostas Drakontidis Date: Mon, 25 May 2026 05:55:04 +0300 Subject: [PATCH 5/7] flush redis on startup and init subredis --- src/officesense_pi/src/core/bootstrap.ts | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/src/officesense_pi/src/core/bootstrap.ts b/src/officesense_pi/src/core/bootstrap.ts index a089428..551a67d 100644 --- a/src/officesense_pi/src/core/bootstrap.ts +++ b/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); } } \ No newline at end of file From 46016d966d94f34e700bc6caf77e4e3f67e73da1 Mon Sep 17 00:00:00 2001 From: Kostas Drakontidis Date: Mon, 25 May 2026 05:55:43 +0300 Subject: [PATCH 6/7] ngsi-ld public api code --- .../src/api/livedata/livadata.routes.ts | 10 + .../src/api/livedata/livedata.controller.ts | 172 ++++++++++++++++++ .../src/api/livedata/livedata.repository.ts | 106 +++++++++++ 3 files changed, 288 insertions(+) create mode 100644 src/officesense_pi/src/api/livedata/livadata.routes.ts create mode 100644 src/officesense_pi/src/api/livedata/livedata.controller.ts create mode 100644 src/officesense_pi/src/api/livedata/livedata.repository.ts diff --git a/src/officesense_pi/src/api/livedata/livadata.routes.ts b/src/officesense_pi/src/api/livedata/livadata.routes.ts new file mode 100644 index 0000000..1ada5c2 --- /dev/null +++ b/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; diff --git a/src/officesense_pi/src/api/livedata/livedata.controller.ts b/src/officesense_pi/src/api/livedata/livedata.controller.ts new file mode 100644 index 0000000..ae0a9bc --- /dev/null +++ b/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 = { + 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; + authenticationStatus: Property; + observedAt: Property; +} + +interface Room extends Entity { + type: "Room"; + occupancy: Property; +} + +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 { + 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 { + 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 { + 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, 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); +} diff --git a/src/officesense_pi/src/api/livedata/livedata.repository.ts b/src/officesense_pi/src/api/livedata/livedata.repository.ts new file mode 100644 index 0000000..fc3ebbd --- /dev/null +++ b/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 { + 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 { + 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 } + }); +} \ No newline at end of file From 80ff81b1e0f71340896a4b5511fabe101c992d0f Mon Sep 17 00:00:00 2001 From: Kostas Drakontidis Date: Mon, 25 May 2026 05:57:35 +0300 Subject: [PATCH 7/7] room occupancy impl & race cond fix & local user roomID and psuedoID storage for cleanup and indexing --- src/officesense_pi/src/core/analyze.ts | 185 ++++++++++++++++--------- 1 file changed, 118 insertions(+), 67 deletions(-) diff --git a/src/officesense_pi/src/core/analyze.ts b/src/officesense_pi/src/core/analyze.ts index 4462be7..5159c1b 100644 --- a/src/officesense_pi/src/core/analyze.ts +++ b/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(); +const userRooms = new Map(); +const userPseudo = new Map(); + +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"); } \ No newline at end of file