feat: 扩展心跳消息支持电力与空调设备数组字段

新增 Kafka 消息中 electricity[] 和 air_conditioner[] 数组字段支持，用于存储电力与空调设备明细数据。数据库表新增对应数组列并创建 GIN 索引优化查询性能，processor 实现数组字段校验与聚合转换逻辑。主要变更： - Kafka 消息规范新增 electricity 和 air_conditioner 数组字段定义 - 数据库 heartbeat_events 表新增 14 个数组列并创建 4 个 GIN 索引 - processor 实现数组字段解析、校验及聚合转换逻辑 - 更新相关文档与测试用例，确保端到端功能完整
2026-01-16 14:45:36 +08:00
parent 7d5b9c50ea
commit 455185ac5d
16 changed files with 693 additions and 8 deletions
--- a/src/db/databaseManager.js
+++ b/src/db/databaseManager.js
@@ -95,6 +95,21 @@ class DatabaseManager {
          device_count int2        NOT NULL,
          comm_seq     int4        NOT NULL,

+          elec_address text[],
+          air_address  text[],
+          voltage      double precision[],
+          ampere       double precision[],
+          power        double precision[],
+          phase        text[],
+          energy       double precision[],
+          sum_energy   double precision[],
+          state        int2[],
+          model        int2[],
+          speed        int2[],
+          set_temp     int2[],
+          now_temp     int2[],
+          solenoid_valve int2[],
+
          extra        jsonb,

          CONSTRAINT heartbeat_events_pk PRIMARY KEY (ts_ms, id),
@@ -112,12 +127,31 @@ class DatabaseManager {
        )
        PARTITION BY RANGE (ts_ms);

+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS elec_address text[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS air_address text[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS voltage double precision[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS ampere double precision[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS power double precision[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS phase text[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS energy double precision[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS sum_energy double precision[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS state int2[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS model int2[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS speed int2[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS set_temp int2[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS now_temp int2[];
+        ALTER TABLE heartbeat.heartbeat_events ADD COLUMN IF NOT EXISTS solenoid_valve int2[];
+
        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_hotel_id ON heartbeat.heartbeat_events (hotel_id);
        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_power_state ON heartbeat.heartbeat_events (power_state);
        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_guest_type ON heartbeat.heartbeat_events (guest_type);
        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_device_id ON heartbeat.heartbeat_events (device_id);
        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_service_mask_brin ON heartbeat.heartbeat_events USING BRIN (service_mask);
        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_hotel_ts ON heartbeat.heartbeat_events (hotel_id, ts_ms);
+        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_elec_address_gin ON heartbeat.heartbeat_events USING GIN (elec_address);
+        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_air_address_gin ON heartbeat.heartbeat_events USING GIN (air_address);
+        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_state_gin ON heartbeat.heartbeat_events USING GIN (state);
+        CREATE INDEX IF NOT EXISTS idx_heartbeat_events_model_gin ON heartbeat.heartbeat_events USING GIN (model);

        -- 分区预创建函数（按 Asia/Shanghai 自然日）
        CREATE OR REPLACE FUNCTION heartbeat.day_start_ms_shanghai(p_day date)
@@ -366,6 +400,20 @@ class DatabaseManager {
      'carbon_state',
      'device_count',
      'comm_seq',
+      'elec_address',
+      'air_address',
+      'voltage',
+      'ampere',
+      'power',
+      'phase',
+      'energy',
+      'sum_energy',
+      'state',
+      'model',
+      'speed',
+      'set_temp',
+      'now_temp',
+      'solenoid_valve',
      'extra',
    ];

@@ -387,6 +435,20 @@ class DatabaseManager {
          e.carbon_state,
          e.device_count,
          e.comm_seq,
+          Array.isArray(e.elec_address) ? e.elec_address : null,
+          Array.isArray(e.air_address) ? e.air_address : null,
+          Array.isArray(e.voltage) ? e.voltage : null,
+          Array.isArray(e.ampere) ? e.ampere : null,
+          Array.isArray(e.power) ? e.power : null,
+          Array.isArray(e.phase) ? e.phase : null,
+          Array.isArray(e.energy) ? e.energy : null,
+          Array.isArray(e.sum_energy) ? e.sum_energy : null,
+          Array.isArray(e.state) ? e.state : null,
+          Array.isArray(e.model) ? e.model : null,
+          Array.isArray(e.speed) ? e.speed : null,
+          Array.isArray(e.set_temp) ? e.set_temp : null,
+          Array.isArray(e.now_temp) ? e.now_temp : null,
+          Array.isArray(e.solenoid_valve) ? e.solenoid_valve : null,
          e.extra ?? null
        );
        const row = columns.map((_, colIndex) => `$${base + colIndex + 1}`).join(', ');
--- a/src/processor/heartbeatProcessor.js
+++ b/src/processor/heartbeatProcessor.js
@@ -124,11 +124,31 @@ class HeartbeatProcessor {
    }
    if (typeof normalized.ip !== 'string' || normalized.ip.length === 0) return false;

+    if (normalized.electricity !== undefined && normalized.electricity !== null) {
+      if (!Array.isArray(normalized.electricity)) return false;
+      for (const item of normalized.electricity) {
+        if (!item || typeof item !== 'object' || Array.isArray(item)) {
+          return false;
+        }
+      }
+    }
+
+    if (normalized.air_conditioner !== undefined && normalized.air_conditioner !== null) {
+      if (!Array.isArray(normalized.air_conditioner)) return false;
+      for (const item of normalized.air_conditioner) {
+        if (!item || typeof item !== 'object' || Array.isArray(item)) {
+          return false;
+        }
+      }
+    }
+
    return true;
  }

  transformData(data) {
-    return this.normalizeHeartbeat(data);
+    const n = this.normalizeHeartbeat(data);
+    const ext = this.aggregateArrays(n);
+    return { ...n, ...ext };
  }

  async processBatch() {
@@ -377,6 +397,8 @@ class HeartbeatProcessor {
      device_count: pick(['device_count', 'deviceCount', 'DeviceCount']),
      comm_seq: pick(['comm_seq', 'commSeq', 'CommSeq']),
      extra: pick(['extra', 'Extra']),
+      electricity: pick(['electricity', 'Electricity']),
+      air_conditioner: pick(['air_conditioner', 'airConditioner', 'AirConditioner']),
    };

    const toTrimmedStringOrUndefined = (v) => {
@@ -447,7 +469,9 @@ class HeartbeatProcessor {
          'carbon_state','carbonState','CarbonState',
          'device_count','deviceCount','DeviceCount',
          'comm_seq','commSeq','CommSeq',
-          'extra','Extra'
+          'extra','Extra',
+          'electricity','Electricity',
+          'air_conditioner','airConditioner','AirConditioner'
        ].includes(k)
      ) {
        continue;
@@ -457,6 +481,52 @@ class HeartbeatProcessor {

    return normalized;
  }
+
+  aggregateArrays(n) {
+    const toStrOrNull = (v) => {
+      if (v === undefined || v === null) return null;
+      const s = String(v).trim();
+      return s.length === 0 ? null : s;
+    };
+    const toNumOrNull = (v) => {
+      if (v === undefined || v === null) return null;
+      if (typeof v === 'number') return Number.isFinite(v) ? v : null;
+      const s = String(v).trim();
+      if (s.length === 0) return null;
+      const num = Number(s);
+      return Number.isFinite(num) ? num : null;
+    };
+    const toIntOrNull = (v) => {
+      if (v === undefined || v === null) return null;
+      if (typeof v === 'number') return Number.isFinite(v) ? Math.trunc(v) : null;
+      const s = String(v).trim();
+      if (!/^-?\d+$/.test(s)) return null;
+      const num = Number(s);
+      return Number.isFinite(num) ? Math.trunc(num) : null;
+    };
+    const out = {};
+    const elec = Array.isArray(n.electricity) ? n.electricity : null;
+    if (elec && elec.length) {
+      out.elec_address = elec.map((x) => toStrOrNull(x?.address));
+      out.voltage = elec.map((x) => toNumOrNull(x?.voltage));
+      out.ampere = elec.map((x) => toNumOrNull(x?.ampere));
+      out.power = elec.map((x) => toNumOrNull(x?.power));
+      out.phase = elec.map((x) => toStrOrNull(x?.phase));
+      out.energy = elec.map((x) => toNumOrNull(x?.energy));
+      out.sum_energy = elec.map((x) => toNumOrNull(x?.sum_energy));
+    }
+    const ac = Array.isArray(n.air_conditioner) ? n.air_conditioner : null;
+    if (ac && ac.length) {
+      out.air_address = ac.map((x) => toStrOrNull(x?.address));
+      out.state = ac.map((x) => toIntOrNull(x?.state));
+      out.model = ac.map((x) => toIntOrNull(x?.model));
+      out.speed = ac.map((x) => toIntOrNull(x?.speed));
+      out.set_temp = ac.map((x) => toIntOrNull(x?.set_temp));
+      out.now_temp = ac.map((x) => toIntOrNull(x?.now_temp));
+      out.solenoid_valve = ac.map((x) => toIntOrNull(x?.solenoid_valve));
+    }
+    return out;
+  }
 }

 export { HeartbeatProcessor };