3T-UEM物联网数据采集架构与云端核算引擎实现方案

# 表计数据采集模型
class MeterDataModel:
    """统一表计数据对象"""
    def __init__(self):
        self.meter_id = None        # 表计唯一标识
        self.meter_type = None      # electric/water/gas/air/cold/heat
        self.protocol = None        # DLT645/CJT188/Modbus/OC
        self.communication = None   # 4G/RS485/MBUS/Cat.1
        self.reading = None          # 当前读数
        self.unit = None             # kWh/m3/GJ/kPa
        self.timestamp = None        # 采集时间戳
        self.quality = None          # verified/corrected/pending
        self.source = None           # direct(通信直读) / ocr(图像识别)

class CollectionScheduler:
    """采集调度引擎"""
    ELECTRIC_METER = {
        "mode": "hourly_active_report",
        "retry": "透传补抄 (每小时05分前未收到)",
        "ensures": "日冻结数据完整性"
    }
    WATER_METER_4G = {
        "mode": "daily_freeze_report",
        "time": "00:00后",
        "content": "总用水量+24h整点用量",
        "realtime": False,
        "manual_trigger": "表端触摸唤醒"
    }
    OCR_TERMINAL = {
        "mode": "scheduled_capture",
        "default_freq": "每日1次",
        "upload": "4G/Cat.1",
        "fallback": "蓝牙4.2近场",
        "pipeline": "拍照->缓存->上传->AI识别->标准化入库"
    }

# OCR终端与通信直读数据归一化
class OCRDataNormalizer:
    """将OCR识别结果归一为与通信直读等价的数据对象"""

    def normalize(self, ocr_result):
        """
        OCR识别结果 -> 标准化数据对象
        与通信直读表计在应用层完全等价
        """
        return {
            "meter_id": ocr_result["meter_id"],
            "meter_type": ocr_result["meter_type"],
            "reading": ocr_result["recognized_value"],
            "unit": self._get_unit(ocr_result["meter_type"]),
            "timestamp": ocr_result["capture_time"],
            "source": "ocr",
            "quality": self._check_confidence(ocr_result),
            "original_image": ocr_result["image_url"]  # 原始图像留存
        }

    def _check_confidence(self, ocr_result):
        """置信度校验：低于阈值不进入计费闭环"""
        if ocr_result["confidence"] < 0.85:
            return "pending_manual"  # 标记待人工确认
        return "verified"

# 多协议适配层
class ProtocolAdapter:
    """多协议适配：将异构表计数据统一为标准数据对象"""

    SUPPORTED_PROTOCOLS = {
        "DLT_645_2007": "电表RS485协议",
        "CJ_T_188_2004": "水表M-BUS协议",
        "Q_GDW_376_1_2013": "集中器上行协议",
        "MODBUS_RTU": "工业通用协议",
        "COAP": "OCR摄像终端协议",
        "MQTT": "IoT云平台协议"
    }

    def adapt(self, raw_frame, protocol):
        """将原始报文解析为标准化数据对象"""
        parser = self._get_parser(protocol)
        parsed = parser.parse(raw_frame)
        return self._to_standard_object(parsed, protocol)

# 腾讯云IoT Hub物模型对接
class TencentIoTHubConnector:
    """表计数据通过IoT Hub上报腾讯云"""

    def __init__(self, product_id, device_name, secret):
        self.hub = IoTHubClient(product_id, device_name, secret)

    def report_meter_data(self, meter_data):
        """上报表计数据到IoT Hub"""
        payload = {
            "type": "report",
            "timestamp": meter_data["timestamp"],
            "params": {
                "meter_id": meter_data["meter_id"],
                "meter_type": meter_data["meter_type"],
                "reading": meter_data["reading"],
                "unit": meter_data["unit"],
                "quality": meter_data["quality"],
                "source": meter_data["source"]
            }
        }
        # 物模型属性上报
        self.hub.publish_property(payload)

    def on_control_command(self, callback):
        """接收云端下发控制指令（如远程断电/参数配置）"""
        self.hub.on_control(callback)

# 云端核算引擎核心流程
class CloudCalculationEngine:
    """全剖面通用核算骨架"""

    def execute(self, billing_period):
        # Step 1: 总表对标
        t1_data = self.fetch_t1_readings(billing_period)
        external_bills = self.fetch_external_bills(billing_period)
        self.verify_against_external(t1_data, external_bills)

        # Step 2: S2数据清洗
        raw_data = self.fetch_all_readings(billing_period)
        cleaned = self.s2_clean(raw_data)

        # Step 3: 损耗核算
        total_loss = self.calc_total_loss(t1_data, cleaned["t2"])

        # Step 4: 单元拆分
        building_losses = self.calc_building_losses(cleaned["t2"], cleaned["t3"])

        # Step 5: 剖面算法
        profile = self.get_active_profile()
        result = self.run_profile_algorithm(profile, cleaned, building_losses)

        # Step 6: 存证
        self.store_evidence_chain(result)
        return result

    def s2_clean(self, readings):
        """S2数据完善与清洗"""
        for r in readings:
            if self._is_jump(r):      # 跳数检测
                r = self._fix_jump(r)
            if r["reading"] < 0:      # 负值修正
                r["reading"] = abs(r["reading"])
            if self._is_spike(r):     # 瞬时突变
                r["quality"] = "spike"
            if r.get("source") == "ocr":
                if r.get("confidence", 1.0) < 0.85:
                    r["quality"] = "pending_manual"
        return readings

# 多能种数据采集模型（与水电完全同构）
MULTI_ENERGY_METERS = {
    "electric": {"unit": "kWh", "protocol": "DLT645", "t1_vs": "供电局账单"},
    "water": {"unit": "m3", "protocol": "CJT188", "t1_vs": "自来水公司账单"},
    "gas": {"unit": "m3", "protocol": "Modbus", "t1_vs": "燃气公司账单"},
    "compressed_air": {"unit": "m3", "protocol": "Modbus", "t1_vs": "空压站总表"},
    "cold": {"unit": "kWh", "protocol": "Modbus", "t1_vs": "冷站总表"},
    "heat": {"unit": "GJ", "protocol": "Modbus", "t1_vs": "换热站总表"}
}

# 每种介质独立跑一组校验等式
def multi_energy_verification(meter_data_by_type):
    """多能流并行校验"""
    results = {}
    for energy_type, data in meter_data_by_type.items():
        t1 = data["t1"]
        t2_sum = sum(data["t2"])
        total_loss = t1 - t2_sum
        results[energy_type] = {
            "total": t1,
            "t2_sum": t2_sum,
            "loss": total_loss,
            "loss_rate": total_loss / t1 if t1 > 0 else 0
        }
    return results

# 三层存证体系
class EvidenceChain:
    """采集存证 -> 核算存证 -> 公示存证"""

    LAYERS = {
        "collection": {
            "content": "表计原始读数+时间戳（含OCR原始图像）",
            "purpose": "防篡改基础",
            "reports": ["抄表数据", "电力数据", "OCR原始图像"]
        },
        "calculation": {
            "content": "分摊/单耗计算过程全程留痕",
            "purpose": "公式可回溯",
            "reports": ["电费账单", "水费账单", "公摊账单", "账目盘点"]
        },
        "publication": {
            "content": "月度账单推送+签收记录",
            "purpose": "司法凭证",
            "reports": ["收费通知单", "低余额提醒", "欠费统计"]
        }
    }

    def store(self, layer, data):
        """存证入库，全程可追溯"""
        record = {
            "layer": layer,
            "data": data,
            "timestamp": datetime.now(),
            "hash": hashlib.sha256(str(data).encode()).hexdigest()
        }
        self.db.insert(record)