src/services/EmulatorService.ts - develop - h0tnanny/IotPlatform

src/services/EmulatorService.ts
577 строк18 KB
h0tnanny
Парсинг данных как строчки
20 фев 2026, 19:31
20 фев 2026, 19:31fbb7684
import { Express } from 'express';
import dgram from 'dgram';
import mqtt from 'mqtt';
import coap from 'coap';
import {
  EmulatorRepository,
  EmulatedDeviceRow,
  EmulatedDeviceVarRow,
} from '../repositories/EmulatorRepository';
import { EquipmentRepository } from '../repositories/EquipmentRepository';

export interface RunningDevice {
  device: EmulatedDeviceRow;
  variables: EmulatedDeviceVarRow[];
  intervalId?: NodeJS.Timeout;
  udpSocket?: dgram.Socket;
  mqttClient?: mqtt.MqttClient;
  coapServer?: ReturnType<typeof coap.createServer>;
  currentValues: Record<string, unknown>;
}

/**
 * Сервис эмуляции устройств.
 * Управляет жизненным циклом эмулируемых устройств:
 * - Server mode: отвечает на HTTP GET / UDP / MQTT / CoAP запросы сгенерированными данными
 *   (настройки MQTT/CoAP берутся из привязанного equipment)
 * - Client mode: периодически отправляет HTTP POST / UDP датаграммы на целевой endpoint
 */
export class EmulatorService {
  private readonly devices: Map<string, RunningDevice> = new Map();
  private readonly repository: EmulatorRepository;
  private readonly equipmentRepository: EquipmentRepository;
  private readonly port: number;
  private routeRegistered = false;

  constructor(
    private readonly app: Express,
    port?: number
  ) {
    this.repository = new EmulatorRepository();
    this.equipmentRepository = new EquipmentRepository();
    this.port = port ?? parseInt(process.env.PORT || '3000', 10);
  }

  /**
   * Инициализация: загрузка устройств из БД, регистрация маршрута, запуск работающих
   */
  async initialize(): Promise<void> {
    this.registerHttpRoute();

    try {
      const allDevices = await this.repository.findAll();
      console.log(`[Emulator] Загружено ${allDevices.length} эмулируемых устройств`);

      for (const { device, variables } of allDevices) {
        this.devices.set(device.id, {
          device,
          variables,
          currentValues: {},
        });

        if (device.is_running) {
          await this.startDevice(device.id);
        }
      }
    } catch (error) {
      console.error('[Emulator] Ошибка загрузки устройств:', error);
    }
  }

  /**
   * Регистрирует единый HTTP GET маршрут для всех server-mode устройств
   */
  private registerHttpRoute(): void {
    if (this.routeRegistered) return;
    this.routeRegistered = true;

    this.app.get('/emulator/devices/:deviceId/data', (req, res) => {
      const { deviceId } = req.params;
      const running = this.devices.get(deviceId);

      if (!running || !running.device.is_running || running.device.mode !== 'server') {
        res.status(404).json({
          error: {
            code: 'DEVICE_NOT_FOUND',
            message: 'Эмулируемое устройство не найдено или не запущено',
          },
        });
        return;
      }

      const values = this.generateAllValues(running);
      running.currentValues = values;
      res.json(values);
    });
  }

  // --- Value Generation ---

  /**
   * Генерирует значение для переменной по её типу и gen_config
   */
  private generateValue(variable: EmulatedDeviceVarRow): unknown {
    const config = variable.gen_config;

    switch (variable.type) {
      case 'Number': {
        const min = (config.min as number) ?? 0;
        const max = (config.max as number) ?? 100;
        const step = (config.step as number) ?? 1;
        const range = Math.floor((max - min) / step);
        return min + Math.floor(Math.random() * (range + 1)) * step;
      }
      case 'Double': {
        const min = (config.min as number) ?? 0;
        const max = (config.max as number) ?? 100;
        const precision = (config.precision as number) ?? 2;
        return parseFloat((min + Math.random() * (max - min)).toFixed(precision));
      }
      case 'Boolean': {
        const probability = (config.probability as number) ?? 0.5;
        return Math.random() < probability;
      }
      case 'String': {
        const values = (config.values as string[]) ?? ['value'];
        return values[Math.floor(Math.random() * values.length)];
      }
      default:
        return null;
    }
  }

  /**
   * Генерирует все значения для устройства
   */
  private generateAllValues(device: RunningDevice): Record<string, unknown> {
    const result: Record<string, unknown> = {};
    for (const v of device.variables) {
      result[v.name] = this.generateValue(v);
    }
    return result;
  }

  // --- Device Lifecycle ---

  /**
   * Запускает эмуляцию устройства
   */
  async startDevice(deviceId: string): Promise<void> {
    const running = this.devices.get(deviceId);
    if (!running) {
      throw new Error(`Устройство ${deviceId} не найдено`);
    }

    if (running.device.is_running && (running.intervalId || running.udpSocket)) {
      return; // Уже запущено
    }

    running.device = { ...running.device, is_running: true };
    running.currentValues = this.generateAllValues(running);

    if (running.device.mode === 'server') {
      await this.startServerMode(running);
    } else {
      this.startClientMode(running);
    }

    try {
      await this.repository.updateRunningStatus(deviceId, true);
    } catch (error) {
      console.error(`[Emulator] Ошибка обновления статуса ${deviceId}:`, error);
    }

    console.log(
      `[Emulator] Устройство "${running.device.name}" (${running.device.mode}) запущено`
    );
  }

  /**
   * Останавливает эмуляцию устройства
   */
  async stopDevice(deviceId: string): Promise<void> {
    const running = this.devices.get(deviceId);
    if (!running) {
      throw new Error(`Устройство ${deviceId} не найдено`);
    }

    this.cleanupDevice(running);
    running.device = { ...running.device, is_running: false };

    try {
      await this.repository.updateRunningStatus(deviceId, false);
    } catch (error) {
      console.error(`[Emulator] Ошибка обновления статуса ${deviceId}:`, error);
    }

    console.log(`[Emulator] Устройство "${running.device.name}" остановлено`);
  }

  /**
   * Останавливает все устройства (graceful shutdown)
   */
  stopAll(): void {
    for (const running of this.devices.values()) {
      this.cleanupDevice(running);
    }
    console.log('[Emulator] Все устройства остановлены');
  }

  private cleanupDevice(running: RunningDevice): void {
    if (running.intervalId) {
      clearInterval(running.intervalId);
      running.intervalId = undefined;
    }
    if (running.udpSocket) {
      try {
        running.udpSocket.close();
      } catch (_e) {
        // Socket already closed
      }
      running.udpSocket = undefined;
    }
    if (running.mqttClient) {
      try {
        running.mqttClient.end();
      } catch (_e) {
        // Client already closed
      }
      running.mqttClient = undefined;
    }
    if (running.coapServer) {
      try {
        running.coapServer.close();
      } catch (_e) {
        // Server already closed
      }
      running.coapServer = undefined;
    }
  }

  // --- Server Mode ---

  /**
   * Server mode: определяет протокол из привязанного equipment.
   * - HTTP: маршрут уже зарегистрирован (wildcard)
   * - UDP: создаём сокет на указанном порту
   * - MQTT: берём broker URL и topic из equipment, публикуем данные по таймеру
   * - CoAP: берём endpoint из equipment, запускаем CoAP-сервер на нужном порту
   */
  private async startServerMode(running: RunningDevice): Promise<void> {
    const { udp_port, target_equipment_id } = running.device;

    // Если есть привязка к equipment — определяем протокол оттуда
    if (target_equipment_id) {
      const equipment = await this.equipmentRepository.findById(target_equipment_id);

      if (equipment?.protocol === 'mqtt' && equipment.mqttBrokerUrl && equipment.mqttTopic) {
        this.startMqttPublisher(running, equipment.mqttBrokerUrl, equipment.mqttTopic);
        return;
      }

      if (equipment?.protocol === 'coap') {
        const coapPort = this.extractPortFromCoapEndpoint(equipment.endpoint);
        this.startCoapServer(running, coapPort);
        return;
      }
    }

    // Fallback: UDP если указан порт, иначе HTTP (wildcard)
    if (udp_port) {
      this.startUdpServer(running, udp_port);
    }
    // HTTP server mode работает через зарегистрированный wildcard маршрут
  }

  /**
   * Извлекает порт из CoAP endpoint (coap://host:port/path)
   */
  private extractPortFromCoapEndpoint(endpoint: string): number {
    try {
      const url = new URL(endpoint);
      return parseInt(url.port, 10) || 5683;
    } catch {
      return 5683;
    }
  }

  private startUdpServer(running: RunningDevice, port: number): void {
    const socket = dgram.createSocket('udp4');

    socket.on('message', (_msg, rinfo) => {
      if (!running.device.is_running) return;

      const values = this.generateAllValues(running);
      running.currentValues = values;
      const response = Buffer.from(JSON.stringify(values));

      socket.send(response, rinfo.port, rinfo.address, (err) => {
        if (err) {
          console.error(
            `[Emulator] UDP ошибка отправки для "${running.device.name}":`,
            err
          );
        }
      });
    });

    socket.on('error', (err) => {
      console.error(
        `[Emulator] UDP ошибка сокета для "${running.device.name}":`,
        err
      );
    });

    socket.bind(port, () => {
      console.log(
        `[Emulator] UDP сервер "${running.device.name}" слушает порт ${port}`
      );
    });

    running.udpSocket = socket;
  }

  /**
   * MQTT publisher: подключается к брокеру и периодически публикует
   * сгенерированные данные в указанный топик
   */
  private startMqttPublisher(
    running: RunningDevice,
    brokerUrl: string,
    topic: string
  ): void {
    const client = mqtt.connect(brokerUrl);

    client.on('connect', () => {
      console.log(
        `[Emulator] MQTT publisher "${running.device.name}" подключён к ${brokerUrl}, топик: ${topic}`
      );

      // Периодическая публикация данных
      const interval = running.device.push_interval || 5000;
      running.intervalId = setInterval(() => {
        if (!running.device.is_running) return;

        const values = this.generateAllValues(running);
        running.currentValues = values;
        client.publish(topic, JSON.stringify(values), (err) => {
          if (err) {
            console.error(
              `[Emulator] MQTT ошибка публикации для "${running.device.name}":`,
              err
            );
          }
        });
      }, interval);
    });

    client.on('error', (err) => {
      console.error(
        `[Emulator] MQTT ошибка для "${running.device.name}":`,
        err.message
      );
    });

    running.mqttClient = client;
  }

  /**
   * CoAP server: запускает CoAP-сервер, отвечающий на GET-запросы
   * сгенерированными данными (аналогично HTTP server mode)
   */
  private startCoapServer(running: RunningDevice, port: number): void {
    const server = coap.createServer(
      (_req: { url: string; method: string }, res: { end: (payload: string) => void }) => {
        if (!running.device.is_running) {
          res.end(JSON.stringify({ error: 'Устройство не запущено' }));
          return;
        }

        const values = this.generateAllValues(running);
        running.currentValues = values;
        res.end(JSON.stringify(values));
      }
    );

    server.listen(port, () => {
      console.log(
        `[Emulator] CoAP сервер "${running.device.name}" слушает порт ${port}`
      );
    });

    server.on('error', (err: Error) => {
      console.error(
        `[Emulator] CoAP ошибка для "${running.device.name}":`,
        err.message
      );
    });

    running.coapServer = server;
  }

  // --- Client Mode ---

  /**
   * Client mode: по таймеру отправляет данные на целевой equipment endpoint
   */
  private startClientMode(running: RunningDevice): void {
    const equipmentId = running.device.target_equipment_id;
    if (!equipmentId) {
      console.error(
        `[Emulator] Устройство "${running.device.name}" не привязано к Equipment`
      );
      return;
    }

    const interval = running.device.push_interval;

    running.intervalId = setInterval(async () => {
      if (!running.device.is_running) return;

      const values = this.generateAllValues(running);
      running.currentValues = values;

      try {
        // Определяем, используем HTTP или UDP
        const equipment = this.findEquipmentProtocol(equipmentId);

        const orderedKeys =
          running.device.payload_simulation_format === 'positional'
            ? running.variables.map((v) => v.name)
            : undefined;
        if (equipment === 'udp') {
          await this.sendUdpData(
            equipmentId,
            values,
            running.device.payload_simulation_format,
            orderedKeys
          );
        } else {
          await this.sendHttpData(
            equipmentId,
            values,
            running.device.payload_simulation_format,
            orderedKeys
          );
        }
      } catch (error) {
        console.error(
          `[Emulator] Ошибка отправки данных для "${running.device.name}":`,
          error
        );
      }
    }, interval);
  }

  private async sendHttpData(
    equipmentId: string,
    values: Record<string, unknown>,
    payloadFormat?: 'json' | 'key_value' | 'positional' | null,
    orderedKeys?: string[]
  ): Promise<void> {
    const url = `http://localhost:${this.port}/datasource/${equipmentId}`;

    let body: string;
    let contentType: string;
    if (payloadFormat === 'key_value') {
      body = Object.entries(values)
        .map(([k, v]) => `${k}=${encodeURIComponent(String(v ?? ''))}`)
        .join('&');
      contentType = 'text/plain';
    } else if (payloadFormat === 'positional' && orderedKeys?.length) {
      body = orderedKeys.map((k) => String(values[k] ?? '')).join(' ');
      contentType = 'text/plain';
    } else {
      body = JSON.stringify(values);
      contentType = 'application/json';
    }

    const response = await fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': contentType },
      body,
    });

    if (!response.ok) {
      console.warn(
        `[Emulator] HTTP POST ${url} вернул ${response.status}`
      );
    }
  }

  private async sendUdpData(
    equipmentId: string,
    values: Record<string, unknown>,
    payloadFormat?: 'json' | 'key_value' | 'positional' | null,
    orderedKeys?: string[]
  ): Promise<void> {
    await this.sendHttpData(equipmentId, values, payloadFormat, orderedKeys);
  }

  /**
   * Проверяет протокол целевого equipment (заглушка — всегда http,
   * т.к. client mode всегда отправляет через REST endpoint платформы)
   */
  private findEquipmentProtocol(_equipmentId: string): 'http' | 'udp' {
    return 'http';
  }

  // --- Data Access ---

  /**
   * Добавляет устройство в память (после создания через API)
   */
  addDevice(device: EmulatedDeviceRow, variables: EmulatedDeviceVarRow[]): void {
    this.devices.set(device.id, {
      device,
      variables,
      currentValues: {},
    });
  }

  /**
   * Обновляет устройство в памяти
   */
  updateDevice(device: EmulatedDeviceRow, variables: EmulatedDeviceVarRow[]): void {
    const existing = this.devices.get(device.id);
    if (existing) {
      this.cleanupDevice(existing);
    }
    this.devices.set(device.id, {
      device,
      variables,
      currentValues: existing?.currentValues ?? {},
    });
  }

  /**
   * Удаляет устройство из памяти
   */
  removeDevice(deviceId: string): void {
    const existing = this.devices.get(deviceId);
    if (existing) {
      this.cleanupDevice(existing);
      this.devices.delete(deviceId);
    }
  }

  /**
   * Возвращает все устройства
   */
  getAllDevices(): RunningDevice[] {
    return Array.from(this.devices.values());
  }

  /**
   * Возвращает устройство по ID
   */
  getDevice(deviceId: string): RunningDevice | undefined {
    return this.devices.get(deviceId);
  }

  /**
   * Возвращает текущие значения устройства
   */
  getDeviceValues(deviceId: string): Record<string, unknown> | null {
    const running = this.devices.get(deviceId);
    if (!running) return null;

    if (running.device.is_running) {
      const values = this.generateAllValues(running);
      running.currentValues = values;
      return values;
    }

    return running.currentValues;
  }
}