Server-py/sri-server-bg02/lib/MessageListener.py

from hub import methods, Global

import threading
import time
import json
import os

# --- for linux
# save_dir = f"/home/server/logs"

# --- for windows
# save_dir = r'C:\SRI-DINO.Server-py\logs'  # sri内网测试环境
save_dir = r'C:\logs'  # sri内网测试环境


# last_log_key = str()

class MessageListener(object):
    """
    """
    last_log_key = str()

    @staticmethod
    def method_v001(client, userdata, message):
        """消息处理方法"""
        # --- log ---
        # print(f'MessageListener24: #message.payload: {message.payload}', flush=True)
        # methods.debug_log(f"MessageListener24", f"#message.payload: {message.payload}")

        # --- check ---
        # if not methods.is_dir(save_dir):
        #     out = methods.run_command(f'mkdir -p {save_dir}', callback=True)
        #     methods.debug_log('MessageListener46', f"#out: {out}")

        # --- save log ---
        file_name = methods.now_string('v001-%Y-%m-%d-%H.log')
        log_file_path = os.path.join(save_dir, file_name)
        log_dict = json.loads(message.payload)
        log_list = [

            f"timestamp: {methods.ts_to_string(int(log_dict.get('timestamp')) / 1000)}",  # 时间
            f"userID: {log_dict.get('userID')}",  # 用户id
            f"VehicleID: {log_dict.get('VehicleID')}",  # 车id

            f"directSwitch: {log_dict.get('baseControl')[0].get('directSwitch')}",  # 前后切换
            f"eStop: {log_dict.get('baseControl')[0].get('eStop')}",  # 急停开关
            f"gearCaontrol: {log_dict.get('baseControl')[0].get('gearCaontrol')}",  # 车辆档位控制
            f"hazardLight: {log_dict.get('baseControl')[0].get('hazardLight')}",  # 双闪灯
            f"keyStatus: {log_dict.get('baseControl')[0].get('keyStatus')}",  # 钥匙状态
            f"parkControl: {log_dict.get('baseControl')[0].get('parkControl')}",  # 驻车控制
            f"silencedAlarm: {log_dict.get('baseControl')[0].get('silencedAlarm')}",  # 消报警音按钮
            f"travelLight: {log_dict.get('baseControl')[0].get('travelLight')}",  # 行驶灯光
            f"travelMode: {log_dict.get('baseControl')[0].get('travelMode')}",  # 驾驶模式
            f"vehicleHorn: {log_dict.get('baseControl')[0].get('vehicleHorn')}",  # 喇叭

            f"accPedalF: {log_dict.get('driveControl')[0].get('accPedal')[0].get('accPedalF')}",  # 油门踏板.手油门
            f"accPedalH: {log_dict.get('driveControl')[0].get('accPedal')[0].get('accPedalH')}",  # 油门踏板.脚油门

            f"accPedalH: {log_dict.get('driveControl')[0].get('brakePedal')}",  # 刹车踏板
            f"steeringWheel: {log_dict.get('driveControl')[0].get('steeringWheel')}",  # 方向盘转速
            f"turnMode: {log_dict.get('driveControl')[0].get('turnMode')}",  # 转向模式
            f"turnSignal: {log_dict.get('driveControl')[0].get('turnSignal')}",  # 转向灯

            f"errAccPedal: {log_dict.get('errCode')[0].get('errAccPedal')}",  # 油门信号故障
            f"errBasOperation: {log_dict.get('errCode')[0].get('errBasOperation')}",  # 基本操作故障
            f"errBrakePedal: {log_dict.get('errCode')[0].get('errBrakePedal')}",  # 刹车信号故障
            f"errEndTool: {log_dict.get('errCode')[0].get('errEndTool')}",  # 末端工具控制故障
            f"errHandle: {log_dict.get('errCode')[0].get('errHandle')}",  # 手柄信号故障
            f"errOther: {log_dict.get('errCode')[0].get('errOther')}",  # 其他故障
            f"errSteeringWheel: {log_dict.get('errCode')[0].get('errSteeringWheel')}",  # 转向故障

            f"baseLegControl: {log_dict.get('taskControl')[0].get('baseLegControl')}",  # 支腿动作控制
            f"baseLegSwitch: {log_dict.get('taskControl')[0].get('baseLegSwitch')}",  # 支腿选择开关
            f"bypassSwitch: {log_dict.get('taskControl')[0].get('bypassSwitch')}",  # 旁通开关
            f"cabLift: {log_dict.get('taskControl')[0].get('cabLift')}",  # 驾驶室升降机构
            f"coopSignal: {log_dict.get('taskControl')[0].get('coopSignal')}",  # 协同作业信号
            f"enableHydraulic: {log_dict.get('taskControl')[0].get('enableHydraulic')}",  # 液压使能开关
            f"endJoint: {log_dict.get('taskControl')[0].get('endJoint')}",  # 末端关节控制
            f"esCabLift: {log_dict.get('taskControl')[0].get('esCabLift')}",  # 驾驶室应急下降开关
            f"suckerSelect: {log_dict.get('taskControl')[0].get('suckerSelect')}",  # 吸盘选择开关
            f"taskJoint_1: {log_dict.get('taskControl')[0].get('taskJoint_1')}",  # 第一个作业关节控制
            f"taskJoint_2: {log_dict.get('taskControl')[0].get('taskJoint_2')}",  # 第二个作业关节控制
            f"taskJoint_3: {log_dict.get('taskControl')[0].get('taskJoint_3')}",  # 第三个作业关节控制
            f"toolControl: {log_dict.get('taskControl')[0].get('toolControl')}",  # 末端工具控制
            f"workLight: {log_dict.get('taskControl')[0].get('workLight')}",  # 工作灯

        ]

        # --- check ---
        log_key = '-'.join(log_list[1:])
        if MessageListener.last_log_key and MessageListener.last_log_key == log_key:
            # methods.debug_log(f"MessageListener24", f"#86: {MessageListener.last_log_key == log_key}")
            return

        # --- update --
        MessageListener.last_log_key = log_key

        # --- save ---
        methods.debug_log(f"MessageListener24", f"#log_list: {log_list}")
        methods.write_text(log_file_path, ' | '.join(log_list) + '\n', 'a')

    @classmethod
    def subscribe_v001(cls):
        """"""
        Global.emqx.start_subscribe_loop(
            decorate_method=MessageListener.method_v001,
            subscribe_topic='Vehicle/ControlVehicle/Veh001'
        )

    @staticmethod
    def method_c001(client, userdata, message):
        """消息处理方法"""
        # --- log ---
        methods.debug_log(f"MessageListener|114", f"#message.payload: {message.payload}")

    def subscribe_c001(cls):
        """"""
        Global.emqx.start_subscribe_loop(
            decorate_method=MessageListener.method_c001,
            subscribe_topic='Cockpit/CanBus/CanId001/Cop001'
        )

    @classmethod
    def run(cls, background_is=True):
        thread_list = [
            threading.Thread(target=cls.subscribe_v001),
        ]
        for thread in thread_list:
            thread.setDaemon(True)
            thread.start()
        if background_is:
            return
        for thread in thread_list:
            thread.join()


if __name__ == '__main__':
    # --- test ---
    MessageListener.run()