DataCollectionDevice/converters/modbus_converter.py

"""
@File  : modbus_converter.py
@Author: lee
@Date  : 2022/7/12/0012 8:55:13
@Desc  : 此类为mosbus协议通用解析器，用于：insitu水质传感器，新气象传感器，新水质传感器
"""
import binascii
import math
import struct

from converter import Converter
from event_storage import EventStorage
from logging_config import modbus_converter as logger
from tools.format_value import format_value


class ModbusConverter(Converter):
    def __init__(self, name):
        self._storage = EventStorage()
        self._name = name

    def convert(self, config, data):
        """
        data: [1, {112: 16801, 113: 50856}]
              [站号， {地址：值，地址：值 。。。}]
        """
        if data:
            device_id = data[0]
            data = data[1]
            format_data_dict = {}  # 列表数据转换字典数字
            # [2, {37: 16646, 38: 30230}]
            try:
                for index in config:
                    if int(index["device_id"]) == int(device_id):
                        # addr_type : 'D' 或 'X';
                        # addr_list: [10] 或 [10, 5]
                        addr_type, addr_list = addr_parsing(index['address'])
                        register_addr = addr_list[0]  # 数据地址  example:'X10.5' -> 10
                        # parameter_id = addr_list[1]
                        # print(parameter_id)
                        if register_addr > 40000:
                            register_addr = register_addr - 40001
                        # if register_addr in data and data[register_addr + 4] == parameter_id:
                        if register_addr in data:
                            logger.info(f"----------------------------")
                            if addr_type == 'X':
                                bit_offset_addr = addr_list[1]  # 位偏移地址  example:'X10.5' -> 5
                                if index['data_type'] == "BOOL":
                                    data_bin = bin(65536 | data[register_addr])  # 和65536做或运算，为了留开头0
                                    data_bin = list(data_bin[3:][::-1])  # 去除开头字符 0  b  1并反转字符串
                                    data_bin = list(map(int, data_bin))  # 字符列表转整型列表
                                    if index['modbus_mode'] == "BA":
                                        data_bin = data_bin[8:16] + data_bin[0:8]
                                    if index['negation'] == 1:
                                        return_data = int(bool(1 - data_bin[bit_offset_addr]))  # 取反运算
                                    else:
                                        return_data = data_bin[bit_offset_addr]
                            elif addr_type == 'D':
                                if index['data_type'] == "FLOAT16":
                                    return_data = data[register_addr]
                                elif index['data_type'] == "INT16":
                                    if index['negation'] == 1:
                                        return_data = int(bool(1 - data[register_addr]))  # 取反运算
                                    else:
                                        if data[register_addr] > 32767:
                                            return_data = data[register_addr] - 65536
                                        else:
                                            return_data = data[register_addr]
                                elif index['data_type'] == "INT32":
                                    return_data_H = data[register_addr]
                                    return_data_L = data[register_addr + 1]
                                    if index['modbus_mode'] == "CDAB":
                                        return_data = data[register_addr] * 65536 + data[register_addr + 1]
                                    else:
                                        return_data = data[register_addr + 1] * 65536 + data[register_addr]
                                elif index['data_type'] == "FLOAT32":
                                    t1 = hex(data[register_addr])[2:]
                                    t2 = hex(data[register_addr + 1])[2:]
                                    if len(t1) < 4:
                                        t1 = (4 - len(t1)) * "0" + t1
                                    if len(t2) < 4:
                                        t2 = (4 - len(t2)) * "0" + t2
                                    if index['modbus_mode'] == "CDAB":
                                        return_data = struct.unpack('>f', binascii.unhexlify((t2 + t1).replace(' ', '')))[0]
                                    elif index['modbus_mode'] == "DCBA":
                                        t1_l, t1_r = t1[0:2], t1[2:4]
                                        t2_l, t2_r = t2[0:2], t2[2:4]
                                        t1, t2 = t1_r + t1_l, t2_r + t2_l
                                        return_data = struct.unpack('>f', binascii.unhexlify((t2 + t1).replace(' ', '')))[0]
                                    else:
                                        return_data = struct.unpack('>f', binascii.unhexlify((t1 + t2).replace(' ', '')))[0]
                                    # 电导率转盐度
                                    if index['io_point_name'] == "盐度" and self._name == 'shuzhi_new':
                                        return_data = self.cal_salty(return_data)

                                elif index['data_type'] == "BELZ_FLOAT32":
                                    llj_data = []
                                    for x in range(0, 6, 2):
                                        t1 = hex(data[register_addr + x])[2:]
                                        t2 = hex(data[register_addr + x + 1])[2:]
                                        if len(t1) < 4:
                                            t1 = (4 - len(t1)) * "0" + t1
                                        if len(t2) < 4:
                                            t2 = (4 - len(t2)) * "0" + t2
                                        t = struct.unpack('>f', binascii.unhexlify((t1 + t2).replace(' ', '')))[0]
                                        llj_data.append(t)
                                    return_data = llj_data[0] * 10 ** 6 + llj_data[1] + llj_data[2] / 10 ** 6  # 总累计量
                                elif index['data_type'] == "UINT16":
                                    return_data = data[register_addr]
                            name = 'c' + str(index['serial_number'])
                            # 格式化数据
                            format_data_dict[name] = format_value(index, return_data)
                            logger.info(f"[{self._name}]: {index['io_point_name']}: {format_data_dict}")
                return format_data_dict
            except Exception as e:
                logger.error(f"{self._name}:{repr(e)}")
                return "error"

    def cal_salty(self, conductivity):
        """盐度转换:需要温度和深度参数"""
        sql = f"SELECT serial_number FROM data_point_tbl WHERE device_name = '{self._name}' AND io_point_name = '温度'"
        sql1 = f"SELECT serial_number FROM data_point_tbl WHERE device_name = '{self._name}' AND io_point_name = '深度'"
        res = self._storage.execute_sql(sql)
        res1 = self._storage.execute_sql(sql1)
        serial_number_temperature = 'c' + str(res[0]['serial_number'])
        serial_number_deep = 'c' + str(res1[0]['serial_number'])
        data = self._storage.get_real_data([serial_number_temperature, serial_number_deep])
        a0, a1, a2, a3, a4, a5 = 0.0080, -0.1692, 25.3851, 14.0941, -7.0261, 2.7081
        b0, b1, b2, b3, b4, b5 = 0.0005, -0.0056, -0.0066, -0.0375, 0.0636, -0.0144
        c0, c1, c2, c3, c4 = 0.6766097, 2.00564e-2, 1.104259e-4, -6.9698e-7, 1.0031e-9
        d1, d2, d3, d4 = 3.426e-2, 4.464e-4, 4.215e-1, -3.107e-3
        e1, e2, e3 = 2.070e-5, -6.370e-10, 3.989e-15
        k = 0.0162
        R = conductivity / 42.914

        try:
            if data[serial_number_temperature] and data[serial_number_deep]:
                t = float(data[serial_number_temperature]) * 1.00024
                p = float(data[serial_number_deep])

                rt = c0 + c1 * t + c2 * t * t + c3 * t * t * t + c4 * t * t * t * t
                Rp = 1 + p * (e1 + e2 * p + e3 * p * p) / (1 + d1 * t + d2 * t * t + (d3 + d4 * t) * R)
                Rt = R / (Rp * rt)
                # print("\nRt=",Rt,'\n')
                S = (t - 15) * (b0 + b1 * math.sqrt(Rt) + b2 * Rt + b3 * Rt * math.sqrt(Rt) + b4 * Rt * Rt + b5 * Rt * Rt * math.sqrt(Rt)) / (1 + k * (t - 15))
                Salinity = a0 + a1 * math.sqrt(Rt) + a2 * Rt + a3 * Rt * math.sqrt(Rt) + a4 * Rt * Rt + a5 * Rt * Rt * math.sqrt(Rt) + S
                return Salinity
            else:
                return None
        except Exception as e:
            logger.error(f"{self._name}:{repr(e)}")
            return 0.00


def addr_parsing(addr_smash):
    """
    :param addr_smash: [X15.1] or [D15]
    :return: X, [15, 1] or D [15]
    """
    # addr_smash_1 = 'D2'
    # addr_smash = list(addr_smash)  # 拆分字符串为list ['X', '6', '.', '2']
    addr_type = addr_smash[0]  # 地址类型 : 'D' 或 'X'
    addr_smash = addr_smash[1:]  # 地址部分: '10' 或 '10.5'
    addr_list = list(map(int, addr_smash.split(".")))  # 用“.”分割字符串转换为整型存入列表
    return addr_type, addr_list