import { PrismaClient, type MeasurementPoint } from '../generated/prisma'; import { EventEmitter } from 'events'; import MQTT from '../adapter/mqtt'; import { getMQTTClient, BASE_TOPIC } from './mqttService'; import { AlarmManagement } from './AlarmManagement'; import { VolatileAlert } from '../store/AlarmStateStore'; export interface SensorReading { sensorId: string; value: number; timestamp: Date; } export interface AlertConfig { warningThreshold: number; alarmThreshold: number; warningDelayMs: number; } export class BedService extends EventEmitter { private prisma: PrismaClient; private mqtt: MQTT | null = null; private alarmManagement: AlarmManagement; private baseTopic = `${BASE_TOPIC}pressure`; constructor() { super(); this.prisma = new PrismaClient(); this.alarmManagement = new AlarmManagement(); this.setupAlarmEventListeners(); } private setupAlarmEventListeners(): void { // Forward alarm events this.alarmManagement.on('alertCreated', (alert: VolatileAlert) => { this.emit('alert', alert); }); this.alarmManagement.on('alertUpdated', (alert: VolatileAlert) => { this.emit('alert', alert); }); this.alarmManagement.on('alertRemoved', (alert: VolatileAlert) => { this.emit('alertRemoved', alert); }); } async initialize(mqttConfig?: { host?: string; port?: number; username?: string; password?: string }): Promise { try { // Use mqttService to get initialized MQTT client this.mqtt = await getMQTTClient(mqttConfig); // Subscribe to sensor data topic await this.mqtt.subscribe(`${this.baseTopic}/+/data`, (topic, message) => { this.handleSensorData(topic, message); }); console.log('BedService initialized successfully'); this.emit('initialized'); } catch (error) { console.error('Failed to initialize BedService:', error); throw error; } } private handleSensorData(topic: string, message: string): void { try { // Extract sensor ID from topic: bed/pressure/{sensorId}/data const sensorId = topic.split('/')[2]; const data = JSON.parse(message); const reading: SensorReading = { sensorId, value: data.value, timestamp: new Date(data.timestamp || Date.now()) }; this.processSensorReading(reading); } catch (error) { console.error('Error processing MQTT sensor data:', error); this.emit('error', error); } } private async processSensorReading(reading: SensorReading): Promise { try { // Find measurement point const measurementPoint = await this.prisma.measurementPoint.findUnique({ where: { sensorId: reading.sensorId } }); if (!measurementPoint) { console.warn(`Unknown sensor ID: ${reading.sensorId}`); return; } // Store sensor data await this.prisma.measurementPointData.create({ data: { measurementPointId: measurementPoint.id, value: reading.value, timestamp: reading.timestamp, time: reading.timestamp.toISOString() } }); // Let alarm management handle alerts this.alarmManagement.processSensorReading(reading.sensorId, reading.value, reading.timestamp); this.emit('sensorReading', reading); } catch (error) { console.error('Error processing sensor reading:', error); this.emit('error', error); } } // Public API methods async createMeasurementPoint(data: { sensorId: string; label: string; zone: string; x: number; y: number; pin: number; warningThreshold: number; alarmThreshold: number; warningDelayMs: number; }): Promise { return this.prisma.measurementPoint.create({ data }); } async getMeasurementPoints(): Promise { const points = await this.prisma.measurementPoint.findMany({ orderBy: { zone: 'asc' } }); // Update alarm management with current measurement points const pointsRecord: Record = {}; points.forEach(point => { pointsRecord[point.id] = { id: point.id, sensorId: point.sensorId, label: point.label, zone: point.zone, x: point.x, y: point.y, pin: point.pin, warningThreshold: point.warningThreshold, alarmThreshold: point.alarmThreshold, warningDelayMs: point.warningDelayMs, currentValue: 0, lastUpdateTime: new Date(), status: 'offline' as const }; }); this.alarmManagement.updateMeasurementPoints(pointsRecord); return points; } async getMeasurementPointData(sensorId: string, limit = 100) { const measurementPoint = await this.prisma.measurementPoint.findUnique({ where: { sensorId } }); if (!measurementPoint) { throw new Error(`Measurement point not found for sensor: ${sensorId}`); } return this.prisma.measurementPointData.findMany({ where: { measurementPointId: measurementPoint.id }, orderBy: { timestamp: 'desc' }, take: limit }); } // Get active alerts from alarm management (volatile) getActiveAlerts(): VolatileAlert[] { return this.alarmManagement.getActiveAlerts(); } // Acknowledge alert in volatile store acknowledgeAlert(alertId: string): boolean { return this.alarmManagement.acknowledgeAlert(alertId); } // Silence alert in volatile store silenceAlert(alertId: string): boolean { return this.alarmManagement.silenceAlert(alertId); } async updateAlertConfig(sensorId: string, config: AlertConfig): Promise { return this.prisma.measurementPoint.update({ where: { sensorId }, data: { warningThreshold: config.warningThreshold, alarmThreshold: config.alarmThreshold, warningDelayMs: config.warningDelayMs } }); } async disconnect(): Promise { // Cleanup alarm management this.alarmManagement.cleanup(); // Disconnect MQTT if (this.mqtt) { await this.mqtt.disconnect(); } // Disconnect Prisma await this.prisma.$disconnect(); this.emit('disconnected'); } }