feat(meter): 柜级电表读数落库 + 变化留痕（协议 §5.3 要求"平台原样存库"，此前只写 Redis）
## 背景：协议一直有上报，是平台没存
`status_post.cabinet.meter`（电表读数，单位 **Wh**，硬件最小刻度 0.1 度电）在协议 §5.3 里
字段表写得很明确：**「平台原样存库、不再 ×100」**（`docs/protocol/柜控板固件对接文档.md` L565），
报文示例（L535）也一直带着它。
但实现侧此前**只写 Redis**（`device:{imei}` 的 `meter`，TTL 300 秒）⇒ 每 60 秒报一次、
**5 分钟后被清掉**，整柜累计电能在平台侧等于不存在、历史无法回溯。
本提交补上落库 —— 属**履约**，不是新增协议内容。
## 改动
1. **新表 `cabinet_meter_readings`**（`cabinet_id` / `reading_wh` / `collected_at`，FK→cabinets，
索引 `(cabinet_id, id)`）+ **`cabinets.meter_wh` / `meter_at`**（最新读数，供列表/详情廉价读取）。
2. **新模块 `workers/meter.rs`**：
- `should_record(prev, current)` —— **纯函数判据**，且**真正承担生产判据**（不是只给单测用）；
- `record_if_changed(pool, imei, meter_wh)` —— 读上次值 → 纯函数判定 → 带 `WHERE ... <> ?`
的 UPDATE 做**并发认领** → 认领成功才追加留痕行（2~3 次往返，且读数未变时 0 次写入）。
3. **`status_post` 接线**：`meter` 改为按 `Option` 解析，**区分「字段缺失」与「值为 0」**
（原先 `unwrap_or(0)` 会把缺失当 0 落库，等于在序列里插进凭空的零值）。
Redis 那条路径口径不变（缺失即 0），不影响既有实时状态契约。失败只记 WARN。
## 为什么"只在变化时记一条"
`status_post` 固定 60 秒一次 ⇒ 68 柜约 **9.8 万行/天**；而电表读数**只在有电量消耗时才变**。
只记变化 ⇒ 写入量与真实用电事件成正比，且"变化"本身就是要留的信息
（等价于抄表系统的**抄表事件**）。读数**变小也记**（换表/清零/回绕属真实事件，必须留痕），
但不在此做任何补差推算 —— 口径仍是**宁可少算、不可虚增**。
## 单测（6 条，全过）
`test_missing_meter_is_never_recorded`（⛔ 缺失绝不落库成 0）、
`test_first_reading_is_recorded_as_baseline`、`test_unchanged_reading_is_not_recorded`、
`test_changed_reading_is_recorded_including_decrease`、
`test_absent_and_zero_are_treated_differently`（**反证**：缺失与 0 处置必须不同）、
`test_update_guard_sql_shape`（变化闸门与占位符个数守门）。
## 顺带被两个既有守门拦下（都是它们正常工作，责任在我）
1. `test_schema_sql_energy_columns_comment_unit_is_wh` 的 `!SCHEMA_SQL.contains("kWh")`：
我的新表注释写了"硬件0.1kWh×100"命中了禁词。**守卫意图正确**（防有人把列标成 kWh
而实际存 Wh），故**改自己的措辞**（改述为"最小刻度 0.1 度电=100Wh"），**不放宽守卫**。
2. `EXPECTED_TABLE_COUNT` 21 → 22（断言消息本就写着"新增/删除表时请同步更新本断言"）。
## 验收
- `cargo check --all-targets`：**0 error / 0 warning**
- `cargo test --bin pms-api`：**195 passed**（189 + 6 条新单测，未减少）
This commit is contained in:
12451
parent
fe8436f114
commit
e8d7442377
@ -240,6 +240,15 @@ const SCHEMA_SQL: &str = r#"
|
||||
FOREIGN KEY (cabinet_id) REFERENCES cabinets(id) ON DELETE CASCADE
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS cabinet_meter_readings (
|
||||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||||
cabinet_id BIGINT NOT NULL COMMENT '柜子ID',
|
||||
reading_wh BIGINT NOT NULL COMMENT '电表读数(Wh；协议平台单位，硬件最小刻度为 0.1 度电=100Wh，设备已换算后上报)',
|
||||
collected_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '平台收到该读数的时刻',
|
||||
FOREIGN KEY (cabinet_id) REFERENCES cabinets(id) ON DELETE CASCADE,
|
||||
KEY idx_cabinet_meter_readings_cabinet (cabinet_id, id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS user_feedback (
|
||||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||||
user_id BIGINT NOT NULL COMMENT '提交用户ID',
|
||||
@ -820,6 +829,15 @@ async fn add_columns_if_not_exists(pool: &MySqlPool) -> Result<(), sqlx::Error>
|
||||
ensure_column(pool, "cabinets", "latitude", "DECIMAL(10,7) NULL COMMENT '纬度'").await?;
|
||||
ensure_column(pool, "cabinets", "longitude", "DECIMAL(10,7) NULL COMMENT '经度'").await?;
|
||||
|
||||
// cabinets：**最新电表读数**（柜级累计电能）。协议 §5.3 的 `status_post.cabinet.meter`
|
||||
// 明确要求「平台原样存库、不再 ×100」，此前只写了 Redis（TTL 300s）⇒ 上报被清掉、
|
||||
// 历史无法回溯。这里落一份"最新值"便于列表/详情廉价读取；
|
||||
// **变化留痕**在 `cabinet_meter_readings`（只在读数变化时追加，见 `workers/meter.rs`）。
|
||||
ensure_column(pool, "cabinets", "meter_wh", "BIGINT NULL COMMENT '最新电表读数(Wh)；NULL=从未上报过'")
|
||||
.await?;
|
||||
ensure_column(pool, "cabinets", "meter_at", "DATETIME NULL COMMENT '最新电表读数的收到时刻'")
|
||||
.await?;
|
||||
|
||||
// cabinets：设备认证 / 拓扑 / 工作模式（线上曾缺失，必须补齐）
|
||||
ensure_column(
|
||||
pool,
|
||||
@ -1490,7 +1508,9 @@ mod tests {
|
||||
/// 外键引用的关键字（用于从建表语句中抽取被引用表名）
|
||||
const REFERENCES_MARKER: &str = "REFERENCES ";
|
||||
/// 期望的建表数量（表数量变化时该断言会提醒同步更新）
|
||||
const EXPECTED_TABLE_COUNT: usize = 21;
|
||||
/// 建表数量（**新增/删除表时必须同步**，由 est_schema_sql_table_order_has_forward_reference_returns_none 守门）。
|
||||
/// 21 → 22（2026-09-20 新增 cabinet_meter_readings：电表读数变化留痕，见 workers/meter.rs）。
|
||||
const EXPECTED_TABLE_COUNT: usize = 22;
|
||||
|
||||
/// 抽取建表语句中所有 `REFERENCES <表名>` 的被引用表名
|
||||
fn referenced_tables(statement: &str) -> Vec<&str> {
|
||||
|
||||
184
software/api-server/src/workers/meter.rs
Normal file
184
software/api-server/src/workers/meter.rs
Normal file
@ -0,0 +1,184 @@
|
||||
//! 柜级电表读数（`status_post.cabinet.meter`）的落库与**变化留痕**。
|
||||
//!
|
||||
//! # 为什么需要它
|
||||
//!
|
||||
//! 协议 §5.3 的 `status_post` **本来就在上报** `cabinet.meter`（电表读数，单位 **Wh**，
|
||||
//! 硬件 0.1kWh 读数 ×100，示例 `125500` = 125.5 kWh），字段表还明确写着
|
||||
//! 「**平台原样存库、不再 ×100**」（`docs/protocol/柜控板固件对接文档.md` L565）。
|
||||
//!
|
||||
//! 但实现侧此前**只把它写进 Redis**（`device:{imei}` 的 `meter` 字段，TTL 300 秒），
|
||||
//! **没有落库** ⇒ 每 60 秒上报一次、5 分钟后被清掉，**整柜累计电能在平台侧等于不存在、
|
||||
//! 历史也无法回溯**。本模块补上这一步（属**履约**，不是新增协议内容）。
|
||||
//!
|
||||
//! # 与「订单电量」的区别（⛔ 别混）
|
||||
//!
|
||||
//! | | 含义 | 来源 |
|
||||
//! |---|---|---|
|
||||
//! | **`meter`**（本模块） | **整柜累计电能**（电表读数，天然是累计值） | 协议已有，硬件上报 |
|
||||
//! | `energy` | **单订单充电量** | 协议**没有**该字段（待确认第 13 项），平台按 `V×I×Δt` 积分（见 [`super::energy`]） |
|
||||
//!
|
||||
//! ⛔ **整柜累计电量绝不能由订单电量相加得到**（含非充电损耗差异、且订单电量本身是估算）。
|
||||
//!
|
||||
//! # 为什么"只记变化"
|
||||
//!
|
||||
//! `status_post` 固定每 60 秒一次 ⇒ 每柜每天 1440 条。若无条件落库，68 柜即 **≈9.8 万行/天**，
|
||||
//! 而电表读数**只在有电量消耗时才变**。因此只在**读数相对上次发生变化**时追加一行 ——
|
||||
//! 写入量与真实用电事件成正比，且"变化"本身就是我们要的那个信息。
|
||||
//!
|
||||
//! ⚠️ 读数**变小也要记**（换表 / 电表清零 / 计数器回绕都属真实事件，必须留痕），
|
||||
//! 但不在此处做任何"补差"推算 —— 口径是**宁可少算、不可虚增**（与 [`super::energy`] 同）。
|
||||
|
||||
use sqlx::MySqlPool;
|
||||
|
||||
/// 更新 `cabinets` 上的**最新读数**（`meter_wh` / `meter_at`）。
|
||||
///
|
||||
/// `WHERE ... (meter_wh IS NULL OR meter_wh <> ?)` 是**原子闸门**：
|
||||
/// `rows_affected == 1` 表示"本次读数确实发生了变化，且由本语句认领"。
|
||||
/// 并发下同一柜的两条 `status_post` 也只会有一条认领同一个新值（另一条 `<>` 不成立）。
|
||||
const UPDATE_CABINET_METER_SQL: &str = "UPDATE cabinets SET meter_wh = ?, meter_at = NOW() \
|
||||
WHERE imei = ? AND (meter_wh IS NULL OR meter_wh <> ?)";
|
||||
|
||||
/// 追加一条**变化留痕**。`cabinet_id` 由 `imei` 反查，省掉一次 `SELECT` 往返。
|
||||
const INSERT_METER_READING_SQL: &str =
|
||||
"INSERT INTO cabinet_meter_readings (cabinet_id, reading_wh) SELECT id, ? FROM cabinets WHERE imei = ?";
|
||||
|
||||
/// 读上次读数（`imei` 是唯一索引，成本可忽略）。
|
||||
///
|
||||
/// 为什么先读一次：让 [`should_record`] 这个**纯函数判据真正成为生产判据**
|
||||
/// （否则它只被单测使用 = 空壳，判据与实现会各走各的）。
|
||||
const SELECT_CABINET_METER_SQL: &str = "SELECT meter_wh FROM cabinets WHERE imei = ?";
|
||||
|
||||
/// **判据（纯函数，可单测）**：本次读数是否应当落库留痕。
|
||||
///
|
||||
/// - `current == None`：报文**没有** `meter` 字段（或值非法）⇒ **不记录**
|
||||
/// （⛔ 关键：绝不能把"缺失"当成 `0` 写进去 —— 那会在序列里插进假的零值，
|
||||
/// 后续算区间用量时就是一个凭空的巨大差值。历史实现用 `unwrap_or(0)` 正是这个毛病）；
|
||||
/// - `prev == None`：库中尚无记录（首次上报新柜 / 新加列）⇒ **记录**，作为基准；
|
||||
/// - 有变化（含**变小**）⇒ **记录**；
|
||||
/// - 完全相等 ⇒ 不记录（这是绝大多数情况，60 秒一次但读数常常不变）。
|
||||
pub fn should_record(prev: Option<i64>, current: Option<i64>) -> bool {
|
||||
match current {
|
||||
None => false,
|
||||
Some(_) => prev != current,
|
||||
}
|
||||
}
|
||||
|
||||
/// 落库一次读数（**只有变化时才写**）。
|
||||
///
|
||||
/// 返回 `Ok(true)` 表示本次确实记录了变化（`UPDATE` 认领成功且已追加留痕行）。
|
||||
///
|
||||
/// 失败**只向上抛给调用方记日志**：电表留痕是增值信息，绝不能因为库抖动影响
|
||||
/// `status_post` 的主流程（在线状态、通道实时值）。
|
||||
pub async fn record_if_changed(
|
||||
pool: &MySqlPool,
|
||||
imei: &str,
|
||||
meter_wh: Option<i64>,
|
||||
) -> Result<bool, sqlx::Error> {
|
||||
let Some(current) = meter_wh else {
|
||||
// 报文没带 meter（或值非法）⇒ 什么都不做（见 should_record 的说明）
|
||||
return Ok(false);
|
||||
};
|
||||
|
||||
// ① **判据**：读上次读数，交给纯函数 [`should_record`] 决定要不要留痕。
|
||||
// 绝大多数 `status_post`（读数没变）在这里就返回，连 UPDATE 都不发。
|
||||
let prev: Option<Option<i64>> = sqlx::query_scalar(SELECT_CABINET_METER_SQL)
|
||||
.bind(imei)
|
||||
.fetch_optional(pool)
|
||||
.await?;
|
||||
// 三条路径要分清：柜子不存在（None）/ 该列从未写过（Some(None)）/ 有旧值（Some(Some(v))）
|
||||
let prev = prev.flatten();
|
||||
if !should_record(prev, Some(current)) {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
// ② **并发认领**：`should_record` 是"读后判"，与写之间存在竞态窗口
|
||||
// （同一柜两条 status_post 并发）。真正的闸门是这条 SQL 的 `WHERE`：
|
||||
// 只有"库里当下仍与传入值不同"的那一条能 `rows_affected == 1`。
|
||||
let changed = sqlx::query(UPDATE_CABINET_METER_SQL)
|
||||
.bind(current)
|
||||
.bind(imei)
|
||||
.bind(current)
|
||||
.execute(pool)
|
||||
.await?
|
||||
.rows_affected();
|
||||
|
||||
if changed == 0 {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
sqlx::query(INSERT_METER_READING_SQL)
|
||||
.bind(current)
|
||||
.bind(imei)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// ⛔ **最关键的一条**：字段缺失（`None`）绝不能落库成 `0`。
|
||||
/// 若把缺失当 0，序列里会出现凭空的"骤降到 0 再骤升回去"，区间用量会被算成巨大的假值。
|
||||
#[test]
|
||||
fn test_missing_meter_is_never_recorded() {
|
||||
assert!(!should_record(None, None), "一直缺字段 ⇒ 永不记录");
|
||||
assert!(!should_record(Some(125_500), None), "本次缺字段 ⇒ 不记录（不能当 0）");
|
||||
}
|
||||
|
||||
/// 首次上报（库中无值）⇒ 建立基准，必须记录。
|
||||
#[test]
|
||||
fn test_first_reading_is_recorded_as_baseline() {
|
||||
assert!(should_record(None, Some(0)), "首次上报 0 也是合法基准");
|
||||
assert!(should_record(None, Some(125_500)), "首次上报真实读数");
|
||||
}
|
||||
|
||||
/// 读数**未变化** ⇒ 不记录（这是绝大多数情况；60 秒一次但读数常常不变）。
|
||||
#[test]
|
||||
fn test_unchanged_reading_is_not_recorded() {
|
||||
assert!(!should_record(Some(125_500), Some(125_500)));
|
||||
assert!(!should_record(Some(0), Some(0)));
|
||||
}
|
||||
|
||||
/// 读数**变化**（含**变小**：换表 / 清零 / 回绕）⇒ 必须留痕。
|
||||
/// 变小不在这里做任何补差推算 —— 口径是"宁可少算、不可虚增"。
|
||||
#[test]
|
||||
fn test_changed_reading_is_recorded_including_decrease() {
|
||||
assert!(should_record(Some(125_500), Some(125_600)), "正常增长");
|
||||
assert!(should_record(Some(125_500), Some(1_000)), "变小也要留痕（换表/清零/回绕）");
|
||||
assert!(should_record(Some(125_500), Some(0)), "归零也要留痕");
|
||||
assert!(should_record(Some(0), Some(1)), "从 0 起算的变化");
|
||||
}
|
||||
|
||||
/// 反证：`should_record` 对"缺失 vs 0"的处置必须**不同** ——
|
||||
/// 若哪天有人把它简化成 `prev != current`，这条会红。
|
||||
#[test]
|
||||
fn test_absent_and_zero_are_treated_differently() {
|
||||
// 首次（无 prev）时：Some(0) 记录、None 不记录
|
||||
assert_ne!(
|
||||
should_record(None, Some(0)),
|
||||
should_record(None, None),
|
||||
"『首次读到 0』与『没读到』必须区分 —— 否则缺失会被写成 0"
|
||||
);
|
||||
}
|
||||
|
||||
/// SQL 守门：`WHERE` 子句里的"变化闸门"与绑定占位符个数不得被改坏。
|
||||
#[test]
|
||||
fn test_update_guard_sql_shape() {
|
||||
assert!(
|
||||
UPDATE_CABINET_METER_SQL.contains("meter_wh IS NULL OR meter_wh <> ?"),
|
||||
"变化闸门不得移除：它同时承担『首次(IS NULL)』与『有变化(<> ?)』两种情形。当前：{}",
|
||||
UPDATE_CABINET_METER_SQL
|
||||
);
|
||||
assert_eq!(
|
||||
UPDATE_CABINET_METER_SQL.matches('?').count(),
|
||||
3,
|
||||
"占位符个数必须与 bind 顺序一致（新值 / imei / 比较值）"
|
||||
);
|
||||
assert!(
|
||||
INSERT_METER_READING_SQL.contains("SELECT id, ? FROM cabinets WHERE imei = ?"),
|
||||
"留痕插入靠 imei 反查 cabinet_id，别改成需要额外 SELECT"
|
||||
);
|
||||
}
|
||||
}
|
||||
@ -1,6 +1,7 @@
|
||||
//! 后台协程 — 处理 Redis 队列中的设备消息
|
||||
|
||||
pub mod energy;
|
||||
pub mod meter;
|
||||
pub mod log_retention;
|
||||
pub mod node_checker;
|
||||
pub mod order_timeout;
|
||||
|
||||
@ -1036,7 +1036,11 @@ async fn handle_status_post(
|
||||
let cabinet = msg.get("cabinet");
|
||||
let rssi = msg.get("rssi").and_then(|v| v.as_i64()).unwrap_or(0);
|
||||
let temp = cabinet.and_then(|c| c.get("temp")).and_then(|v| v.as_i64()).unwrap_or(0);
|
||||
let meter = cabinet.and_then(|c| c.get("meter")).and_then(|v| v.as_i64()).unwrap_or(0);
|
||||
// ⚠️ 用 `Option` **保留「字段缺失」与「值为 0」的区别**：
|
||||
// 缺失绝不能当成 0 落库（否则电表序列里会插进凭空的零值，见 `workers::meter`）。
|
||||
// Redis 那条路径沿用原口径（缺失即 0），不改既有实时状态契约。
|
||||
let meter_raw = cabinet.and_then(|c| c.get("meter")).and_then(|v| v.as_i64());
|
||||
let meter = meter_raw.unwrap_or(0);
|
||||
|
||||
let raw_status1 = cabinet.and_then(|c| c.get("status1")).and_then(|v| v.as_str());
|
||||
let raw_status2 = cabinet.and_then(|c| c.get("status2")).and_then(|v| v.as_str());
|
||||
@ -1054,6 +1058,24 @@ async fn handle_status_post(
|
||||
// 1) 设备级 Hash：在线标记 + 信号 + 柜体状态（TTL 300 秒）
|
||||
write_device_state(redis, dev_id, rssi, status1.as_deref(), status2.as_deref(), temp, meter).await;
|
||||
|
||||
// 1b) **电表读数落库 + 变化留痕**（协议 §5.3 字段表要求「平台原样存库」；
|
||||
// 此前只写 Redis，5 分钟后被 TTL 清掉 ⇒ 整柜累计电能留不下来）。
|
||||
// 只在读数**变化**时写（60 秒一次但读数常常不变）；
|
||||
// 失败只记 WARN：留痕是增值信息，绝不能影响 status_post 的主流程。
|
||||
match crate::workers::meter::record_if_changed(mysql, dev_id, meter_raw).await {
|
||||
Ok(true) => tracing::info!(
|
||||
"[status_post] 电表读数变化已留痕 dev_id={} meter={}Wh",
|
||||
dev_id,
|
||||
meter_raw.unwrap_or(0)
|
||||
),
|
||||
Ok(false) => {}
|
||||
Err(e) => tracing::warn!(
|
||||
"[status_post] 电表读数留痕失败（不影响状态上报）dev_id={} err={}",
|
||||
dev_id,
|
||||
e
|
||||
),
|
||||
}
|
||||
|
||||
// 2) 供电恢复即清除停电标记：`cabinet.status1` bit0 = 0（正常）→ pow_fail_dc/ac 置 0；
|
||||
// bit0 = 1 时**不动**（停电事件由 pow_fail_dc / pow_fail_ac 单独置 1，协议 §5.4）
|
||||
clear_power_fail_if_normal(redis, dev_id, raw_status1).await;
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user