add share sub 2026-6-25 16:14

src/ems_water_grp/meter_3rd2.py

+# -*- coding:utf-8 -*-
+"""
+DATE：2026/6/25 16:02
+
+"""
+import json
+import asyncio
+import signal
+import pendulum
+import argparse
+from utils.time_format import CST, YMD_Hms
+from gmqtt import Client as MQTTClient
+from ems_water_grp.constants import SCADA_FIELDS_MAP
+from utils.utils import admin_client_id, mqtt_pwd
+from ems_water_grp.helper import load_mtype_from_sm
+from infra.config.settings import SETTING
+from infra.logger.logger import Logger
+
+
+class Meter3rdForward:
+    sub_host = "172.18.4.82"
+    sub_uname = "nyconsumer"
+    sub_pwd = "nyconsumer!321"
+    # sub_topic = "factory/data"
+    sub_topic = "$share/qkup/factory/data"
+
+    pub_username = "pot_emqx_super"
+    pub_topic_pfx = "eems/td/"
+
+    def __init__(self, worker_id):
+        self.worker_id = worker_id
+        self.client_sub = None
+        self.client_pub = None
+
+        self.stop_event = asyncio.Event()
+
+        self.d_last_msg = {}
+
+    @property
+    def pub_pwd(self):
+        return mqtt_pwd(self.pub_username)
+
+    @staticmethod
+    def on_sub_subscribe(client, mid, qos, properties):
+        logger.info(f"Sub to {Meter3rdForward.sub_topic} success, mid: {mid}")
+
+    @staticmethod
+    def on_sub_connect(client, flags, rc, properties):
+        logger.info(f"Connected to {Meter3rdForward.sub_host} success rc:{rc}")
+        client.subscribe(Meter3rdForward.sub_topic, qos=1)
+
+    @staticmethod
+    def on_sub_disconnect(client, packet, exc=None):
+        logger.warning(f"Disconnected from {Meter3rdForward.sub_host} {exc}")
+
+    @staticmethod
+    def on_pub_connect(client, flags, rc, properties):
+        logger.info(f"Connected to {SETTING.mqtt_host} success, rc: {rc}")
+
+    @staticmethod
+    def on_pub_disconnect(client, packet, exc=None):
+        logger.warning(f"Disconnected from EMQX {SETTING.mqtt_host} exc {exc}")
+
+    def on_sub_message(self, client, topic, payload, qos, properties):
+        try:
+            msg_str = payload.decode("utf-8")
+            data = json.loads(msg_str)
+            fId = data.get("fId")
+            if fId:
+                self.d_last_msg[fId] = msg_str
+        except Exception as e:
+            logger.error(f"Decode msg:{payload} error: {e}")
+
+    @staticmethod
+    async def parse_topic(sid):
+        m_type = await load_mtype_from_sm(sid)
+        if m_type:
+            if m_type == 137:
+                return "pcc_ele"
+            elif m_type == 136:
+                return "load_ele"
+        return ""
+
+    async def parse_and_forward(self, fid, raw_payload):
+        """ 重新封装消息，推送至EMQX """
+        try:
+            d_pyds = {}
+            data = json.loads(raw_payload)
+            ts = pendulum.from_timestamp(data["time"], tz=CST).format(YMD_Hms)
+            for item in data["items"]:
+                try:
+                    scada_id = item["n"]
+                    if scada_id in SCADA_FIELDS_MAP[fid]:
+                        tmp = SCADA_FIELDS_MAP[fid][scada_id]
+                        mid = tmp["sid"]
+                        tag = {tmp["field"]: item["v"]}
+                        if mid in d_pyds:
+                            d_pyds[mid]["images"][0]["tags"].update(tag)
+                        else:
+                            d_pyds[mid] = {"cid": fid, "mid": mid, "nm": mid,
+                                           "images": [{"t": ts, "tags": tag}]}
+                except Exception as e:
+                    continue
+
+            cnt = 0
+            for sid, payload in d_pyds.items():
+                # logger.info(f"{sid}: {payload}")
+                topic = await self.parse_topic(sid)
+                if not topic:
+                    logger.error(f"Parse topic {sid} failed, pyd:{payload}")
+                    continue
+
+                topic = f"{self.pub_topic_pfx}{topic}/{sid}"
+                self.client_pub.publish(topic, json.dumps(payload), qos=1)
+                cnt += 1
+
+            logger.info(f"forward fid:{fid} {cnt} msgs")
+        except Exception as e:
+            logger.error(f"parse_and_forward error: {e}")
+
+    async def snapshot_sampler(self):
+        while not self.stop_event.is_set():
+            try:
+                if self.d_last_msg:
+                    cur_batch = dict(self.d_last_msg)
+                    self.d_last_msg.clear()
+                    for fid, raw_msg in cur_batch.items():
+                        if fid in SCADA_FIELDS_MAP:
+                            await self.parse_and_forward(fid, raw_msg)
+
+                await asyncio.sleep(1)
+            except Exception as e:
+                logger.error(f"Processing loop error: {e}")
+                await asyncio.sleep(1)
+
+    async def stop(self):
+        logger.info("Shutting down Meter3rdForward Service...")
+        self.stop_event.set()
+
+        try:
+            if self.client_sub:
+                await self.client_sub.disconnect()
+        except Exception:
+            pass
+
+        try:
+            if self.client_pub:
+                await self.client_pub.disconnect()
+        except Exception:
+            pass
+
+    async def start(self):
+        client_id = admin_client_id(f"Meter3rdForward{self.worker_id}")
+        self.client_sub = MQTTClient(client_id)
+        self.client_sub.set_auth_credentials(self.sub_uname, self.sub_pwd)
+        self.client_sub.on_connect = self.on_sub_connect
+        self.client_sub.on_subscribe = self.on_sub_subscribe
+        self.client_sub.on_disconnect = self.on_sub_disconnect
+        self.client_sub.on_message = self.on_sub_message
+
+        self.client_pub = MQTTClient(admin_client_id("Meter3rdForward"))
+        self.client_pub.set_auth_credentials(self.pub_username, self.pub_pwd)
+        self.client_pub.on_connect = self.on_pub_connect
+        self.client_pub.on_disconnect = self.on_pub_disconnect
+
+        loop = asyncio.get_running_loop()
+        for sig in (signal.SIGINT, signal.SIGTERM):
+            loop.add_signal_handler(sig, self.stop_event.set)
+
+        while not self.stop_event.is_set():
+            try:
+                await asyncio.gather(
+                    self.client_sub.connect(self.sub_host, 1883),
+                    self.client_pub.connect(SETTING.mqtt_host,
+                                            SETTING.mqtt_port)
+                )
+                logger.info("双向 MQTT 服务初始连接全部成功！")
+                break
+            except Exception as e:
+                logger.error(f"MQTT 初始连接失败，30秒后重试:{e}")
+                await asyncio.sleep(30)
+
+        if self.stop_event.is_set():
+            return
+
+        asyncio.create_task(self.snapshot_sampler())
+
+        logger.info("Meter3rdForward 核心服务启动，等待退出信号...")
+        try:
+            await self.stop_event.wait()
+        finally:
+            await self.stop()
+
+
+async def main(worker_id):
+    while True:
+        try:
+            srv = Meter3rdForward(worker_id)
+            await srv.start()
+            break
+        except Exception as e:
+            logger.error(f"转发服务发生致命异常，10秒后尝试重新初始化: {e}")
+            await asyncio.sleep(10)
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "log_num",
+        metavar="log_num",
+        type=str,
+        help="log file number"
+    )
+    args = parser.parse_args()
+    log_num = args.log_num
+
+    log_name = f"meter3rd_forward_{log_num}"
+    Logger.init_logger_path(f"./ems_water_grp", f"{log_name}.log", log_name)
+    logger = Logger.getLogger(log_name)
+
+    try:
+        asyncio.run(main(log_num))
+    except KeyboardInterrupt:
+        pass