add new converter 'dandian_converter' & fix some bug

alarm.py

@@ -2,7 +2,7 @@
 @Date  :2021/5/21/00219:10:57
 @Desc  :
 """
-from sanic.log import logger
+from logging_config import logger
 import threading
 import time
 from event_storage import EventStorage

config.json

@@ -3,7 +3,7 @@
     "ip": "127.0.0.1",
     "username": "zz",
     "password": "5BFkezzPY1C0/2/y2pMrag==\n",
-    "dataBaseName": "shucai_480"
+    "dataBaseName": "shucai"
   },
   "memoryDatabase": {
     "ip": "127.0.0.1",

configuration.py

@@ -1,5 +1,5 @@
 import json
-from sanic.log import error_logger
+from logging_config import logger
 import sys
 from AES_crypt import decrypt
 
@@ -11,7 +11,7 @@ class Configuration:
             with open(self.path) as json_file:
                 self.config = json.load(json_file)
         except FileNotFoundError as e:
-            error_logger.error(f"config file does not exist:{e}")
+            logger.error(f"无法找到配置文件:{e}")
             input("按任意键退出！")
             sys.exit()
 

connectors/shuizhi_tcp_connector.py

@@ -9,7 +9,7 @@ import struct
 import socket
 from connector import Connector
 from event_storage import EventStorage
-from sanic.log import logger
+from logging_config import logger
 from binascii import *
 from crcmod import *
 
@@ -137,9 +137,9 @@ class ShuizhiTcpConnector(Connector, threading.Thread):
             try:
                 for i in instruct_list:
                     # self.__sock.send(instruct_list[sendFlag])\
-                    time.sleep(.5)
+                    time.sleep(1)
                     self.__sock.send(i)
-                    time.sleep(.5)
+                    time.sleep(1)
                     self.__sock.send(i)
             except Exception as e:
                 self.__connected = False

connectors/tcp_connector.py

@@ -2,12 +2,15 @@
 @Date  :2021/5/21/00219:10:57
 @Desc  :
 """
+import json
 import time
 import threading
-import struct
 import socket
 import queue
-from sanic.log import logger
+
+from modbus_tk import utils
+
+from logging_config import logger
 from connector import Connector
 from event_storage import EventStorage
 
@@ -26,8 +29,8 @@ class TcpConnector(Connector, threading.Thread):
         self.__save_frequency = config['save_frequency']
         self.__command_queue = queue.Queue(50)
         self.setName(name)
-        self.__last_seve_time = 0
         self.__data_point_config = self.__storager.get_station_info(name)
+        self.__command = self.__storager.get_command_info(name)
 
     def open(self):
         self.__stopped = False
@@ -37,8 +40,15 @@ class TcpConnector(Connector, threading.Thread):
         self.__connect()
         self.__connected = True
         while True:
+            if isinstance(self.__command, list):
+                for i in self.__command:
+                    command_list = json.loads(i['command'])
+                    self.command_polling(command_list=command_list)
+                    time.sleep(1)
+            else:
+                self.command_polling()
             time.sleep(1)
-            self.command_polling()
+
             if self.__stopped:
                 break
 
@@ -98,21 +108,39 @@ class TcpConnector(Connector, threading.Thread):
             except Exception as e:
                 logger.info(f'Send command to [{self.get_name()}]:[{self.__ip}]:[{self.__port}] error:{e}')
 
-    def command_polling(self):
-        if self.__connected:
+    def exec_command(self, command):
+        try:
+            com = bytes.fromhex(command['instruct'])
+            self.__sock.send(com)
+            recv_data = self.__sock.recv(self.__size)
+            return recv_data
+        except Exception as e:
+            logger.error(f"{e}")
+
+    def command_polling(self, command_list=None):
+        if command_list:
             try:
-                time.sleep(0.2)
-                data = self.__sock.recv(self.__size)
-                if data == b'':
-                    self.__reconnect()
-                data = self.__converter.convert(self.__data_point_config, data)
-                logger.info(f"format_data = {data}")
-                if data:
-                    if data != "error" and data != 'pass':
-                        self.__storager.real_time_data_storage(data)
+                for i in range(len(command_list)):
+                    command_item = command_list[i]
+                    recv_data = self.exec_command(command=command_item)
+                    format_data = self.__converter.convert(self.__data_point_config, recv_data)
+                    logger.info(f'{self.name}:{format_data}')
+                    if format_data and format_data != "error" and format_data != 'pass':
+                        self.__storager.real_time_data_storage(format_data)
             except Exception as e:
                 logger.error(f'Other error occur [{self.get_name()}]:[{self.__ip}]:[{self.__port}]:{e}')
                 time.sleep(5)
                 self.__reconnect()
         else:
-            self.__reconnect()
+            try:
+                recv_data = self.__sock.recv(self.__size)
+                format_data = self.__converter.convert(self.__data_point_config, recv_data)
+                # logger.info(f'{self.name}: {format_data}')
+                if format_data and format_data != "error" and format_data != 'pass':
+                    self.__storager.real_time_data_storage(format_data)
+            except socket.timeout as e:
+                logger.error(f"{e}")
+            except Exception as e:
+                logger.error(f"{e}")
+                time.sleep(5)
+                self.__reconnect()

converters/__init__.py

@@ -4,4 +4,5 @@ from . import td266_converter
 from . import shuizhi_converter
 from . import adcp_converter
 from . import adcp_converter_480
+from . import dandian_converter
 

converters/adcp_converter.py

@@ -2,7 +2,7 @@
 @Date  :2021/5/21/00219:10:57
 @Desc  :
 """
-from sanic.log import logger
+from logging_config import logger
 from converter import Converter
 
 
@@ -12,12 +12,12 @@ from converter import Converter
 class AdcpConverter(Converter):
     def convert(self, config, data):
         if data:
-            logger.debug(config)
-            logger.debug(data)
+            # logger.debug(config)
+            logger.info(f"(ADCP)原始接收数据：{data}")
             dict = {}
             try:
                 raw_data = data.decode().split("\r\n")
-                # logger.debug(len(raw_data))
+                logger.debug(f"(ADCP)解码分割：{raw_data}")
                 if len(raw_data) == 32:
                     # logger.debug(raw_data)
                     raw_data = raw_data[1:-1]
@@ -30,9 +30,8 @@ class AdcpConverter(Converter):
                         flow_rate_data.append(t1[1] / 1000)  # 流速值除以1000
                         flow_direction.append(t1[2] / 10)  # 流向值除以10
                     format_data = flow_rate_data + flow_direction
-                    logger.debug(format_data)
+                    logger.debug(f"(ADCP)解析后数据: {len(format_data)}, {format_data}")
                     j = 0
-                    logger.debug(len(config), len(format_data))
                     for index in config:
                         name = 'c' + str(index['serial_number'])
                         if index['divisor'] is not None:
@@ -47,5 +46,5 @@ class AdcpConverter(Converter):
                 else:
                     return "error"
             except Exception as e:
-                logger.debug(e)
+                logger.error(e)
                 return "error"

converters/adcp_converter_480.py

@@ -1,8 +1,8 @@
 """
 @Date  :2022/8/10/00219:10:57
-@Desc  : 增加温度解析
+@Desc  : 增加温度深度解析
 """
-from sanic.log import logger
+from logging_config import logger
 from converter import Converter
 
 
@@ -16,15 +16,19 @@ class AdcpConverter(Converter):
             # logger.debug(data)
             dict = {}
             try:
+                logger.debug(f"len(data)= {len(data)}")
+                logger.debug(f"(data)= {data}")
+                if len(data) < 500:
+                    return
                 raw_data = data.decode().split("\r\n")
-                # logger.debug(f"len(raw_data): {len(raw_data)}")
+                logger.debug(f"len(raw_data): {len(raw_data)}")
                 if len(raw_data) == 32:
-                    # logger.debug(f"raw_data= {raw_data}")
+                    logger.debug(f"raw_data= {raw_data}")
                     title = raw_data[0].split(" ")
-                    tempe = [float(title[-3])]
-                    # logger.debug(f"len(title)= {len(title)}")
-                    # logger.debug(f"title= {title}")
-                    # logger.debug(f"tempe={tempe}")
+                    addon = [float(title[-3]), float(title[-4])]
+                    logger.debug(f"len(title)= {len(title)}")
+                    logger.debug(f"title= {title}")
+                    logger.debug(f"addon={addon}")
                     raw_data = raw_data[1:-1]
                     flow_rate_data = []
                     flow_direction = []
@@ -34,7 +38,7 @@ class AdcpConverter(Converter):
                         t1 = [int(x) for x in t]  # 字符串列表转int列表：[1, 211, 3075]
                         flow_rate_data.append(t1[1] / 1000)  # 流速值除以1000
                         flow_direction.append(t1[2] / 10)  # 流向值除以10
-                    format_data = flow_rate_data + flow_direction + tempe
+                    format_data = flow_rate_data + flow_direction + addon
                     # logger.debug(f"len(format_data): {len(format_data)}")
                     # logger.debug(f"format_data: {format_data}")
                     j = 0
@@ -49,10 +53,8 @@ class AdcpConverter(Converter):
                         dict[name] = format_data[j]
                         j += 1
                     return dict
-                elif len(raw_data) > 0:
-                    return "pass"
                 else:
                     return "error"
             except Exception as e:
-                logger.debug(e)
-                return "error"
+                logger.error(e)
+                return

converters/dandian_converter.py

@@ -0,0 +1,33 @@
+"""
+@File  : dandian_converter.py
+@Author: lee
+@Date  : 2022/8/30/0030 14:42:02
+@Desc  :
+"""
+import json
+import re
+from logging_config import logger
+from converter import Converter
+
+
+class DandianConverter(Converter):
+    def convert(self, config, data):
+        # 原始data: {'data': b'\x11\x11pval,22.650,-37.896,3.613,11.104,14.158,17.992,70.457,243.563,-8.010,-16.111,11.813,0\r\n'}
+        # 格式化数据：['\x11\x11pval', '22.658', '-36.617', '11.304', '10.291', '14.330', '17.643', '82.970', '253.655', '-4.965', '-16.930', '11.813', '0\r\n']
+        #                               温度，     pitch，    roll，    X流速，   Y流速，   流速，     方位，   流向，  南 - 北向流速，东 - 西向流速，电压，状态
+        try:
+            data = data.decode('utf-8').split(',')
+            logger.info(f"(新版单点流速仪)原始数据: {data}")
+            dict = {}
+            for index in config:
+                name = 'c' + str(index['serial_number'])
+                i = int(index['address'])
+                if index['divisor'] is None:
+                    dict[name] = float(data[i])
+                else:
+                    dict[name] = round((float(data[i]) / index['divisor']), 2)
+            logger.info(f"(新版单点流速仪)解析后数据：{data}")
+            return dict
+        except Exception as e:
+            logger.error(e)
+            return "error"

converters/nmea0183_converter.py

@@ -2,7 +2,7 @@
 @Date  :2021/5/21/00219:10:57
 @Desc  :
 """
-from sanic.log import logger
+from logging_config import logger
 import binascii
 
 from converter import Converter
@@ -11,11 +11,15 @@ from converter import Converter
 class NEMA0183Converter(Converter):
 
     def convert(self, config, data):
-        data = data.decode()
-        if self.checksum(data):
-            res = self.check_type(config, data)
-            return res
-        return "error"
+        data = data.decode().split("\r\n")
+        logger.info(f"(波高传感器)原始接收数据：len:{len(data)}, data: {data}")
+        for i in data:
+            if self.checksum(i):
+                res = self.check_type(config, i)
+                return res
+            else:
+                logger.info(f"checksum校验失败：{i}")
+                return "error"
 
     def make_checksum(self, data):
         '''
@@ -59,16 +63,16 @@ class NEMA0183Converter(Converter):
             return False
 
     def check_type(self, config, data):
-        logger.info(f"原始数据(波高传感器)：len:{len(data)}, data: {data}")
         dict = {}
         data = data.split('*')
         # Splits up the NMEA data by comma
         data = data[0].split(',')
+        logger.info(f"进一步格式化数据(波高传感器)：len:{len(data)}, data: {data}")
         if data[0] == '$PMIRWM':
             for index in config:
                 name = 'c' + str(index['serial_number'])
-                i = int(index['address'])
-                dict[name] = data[i]
+                address = int(index['address'])
+                dict[name] = format_value(index, data[address])
             logger.info(f"解析后数据(波高传感器)：len:{len(dict)}, dict: {dict}")
             return dict
         # return data[0]
@@ -82,8 +86,27 @@ class NEMA0183Converter(Converter):
         return date + 'T' + time + 'Z'
 
 
+def format_value(index, value):
+    if value:
+        value = float(value)
+        divisor = index['divisor']
+        offset = index['offset']
+        low_limit = index['low_limit']
+        up_limit = index['up_limit']
+        if divisor:
+            value /= divisor
+        if offset:
+            value -= offset
+        if low_limit <= value <= up_limit:
+            return value
+        else:
+            return ''
+    else:
+        return ''
+
+
 '''        
-data = "$PMIRR,20210325,033351.719,0.000,0.000,V,0.00*0F" 
+data = "$PMIRR,20210325,033351.719,0.000,0.000,V,0.00*0F"
 
 data2 = data.split('*')
 data2 = data2[0].split(',')

converters/td266_converter.py

@@ -1,15 +1,19 @@
 import json
 import re
-from sanic.log import logger
+from logging_config import logger
 
 from converter import Converter
 
 
 class TD266Converter(Converter):
     def convert(self, config, data):
+        # 原始data: data = b'4420\t1194\t29.823\t104.507\t-7.471\t28.872\t253.153\t9.369\t1.816\t91.491\t-59.593\t100\t9.542\t9.589\t0.015\r\n'
+        # 去除结尾\r\n: data = b'4420\t1194\t29.823\t104.507\t-7.471\t28.872\t253.153\t9.369\t1.816\t91.491\t-59.593\t100\t9.542\t9.589\t0.015'
+        # decode('utf-8'): data = 4420	1194	29.823	104.507	-7.471	28.872	253.153	9.369	1.816	91.491	-59.593	100	9.542	9.589	0.015
+        # split('\t'): data = ['4420', '1194', '29.823', '104.507', '-7.471', '28.872', '253.153', '9.369', '1.816', '91.491', '-59.593', '100', '9.542', '9.589', '0.015']
         try:
             data = data.decode('utf-8').split('\t')
-            logger.info(f"原始数据(单点流速仪): {data}")
+            logger.info(f"(单点流速仪)原始数据: {data}")
             dict = {}
             for index in config:
                 name = 'c' + str(index['serial_number'])
@@ -18,7 +22,7 @@ class TD266Converter(Converter):
                     dict[name] = float(data[i])
                 else:
                     dict[name] = round((float(data[i]) / index['divisor']), 2)
-            logger.info(f"解析后数据(单点流速仪)：{data}")
+            logger.info(f"(单点流速仪)解析后数据：{data}")
             return dict
         except Exception as e:
             logger.error(e)

converters/wxt536_converter.py

@@ -1,34 +1,29 @@
-from sanic.log import logger
+from logging_config import logger
 import re
 from converter import Converter
 
 
 class WXT536Converter(Converter):
-    '''
-            [{"name":"c1","addr":"1"},
-          {"name":"c2","addr":"2"},
-          {"name":"c3","addr":"3"},
-          {"name":"c4","addr":"4"},
-          {"name":"c5","addr":"5"},
-          {"name":"c6","addr":"6"}]
-    '''
+    """
+    data: b'0R0,Dm=267D,Sm=1.2M,Ta=-25.0C,Ua=87.1P,Pa=1001.9H,Rc=-0.00M,Th=28.3C,Vh=0.0#'
+    """
 
     def convert(self, config, data):
-        logger.info(f"data: {data}")
+        logger.info(f"(气象仪)原始接收数据: {data}")
         if data:
             dict = {}
             try:
                 list = data.decode().split(",")
-                logger.info(f"list: {list}, len: {len(list)}")
-                if len(list) == 9:
+                logger.info(f"(气象仪)解码分割后， 标准长度:9，实际长度:{len(list)}, 内容: {list}, ")
+                if list[0] == '0R0':
                     for index in config:
                         name = 'c' + str(index['serial_number'])
                         i = int(index['address'])
+                        value = None
                         if list[i][-1] != "#":
-                            dict[name] = re.findall(r"\d+\.?\d*", list[i])[0]
-                        else:
-                            dict[name] = "null"
-                    logger.info(f"解析后数据(气象传感器)：{dict}")
+                            value = re.findall(r"-*\d+\.?\d*", list[i])[0]
+                        dict[name] = format_value(index, value)
+                    logger.info(f"(气象仪)解析后数据：{dict}")
                     return dict
                 elif len(list) > 0:
                     return "pass"
@@ -37,3 +32,22 @@ class WXT536Converter(Converter):
             except Exception as e:
                 logger.error(e)
                 return "error"
+
+
+def format_value(index, value):
+    if value:
+        value = float(value)
+        divisor = index['divisor']
+        offset = index['offset']
+        low_limit = index['low_limit']
+        up_limit = index['up_limit']
+        if divisor:
+            value /= divisor
+        if offset:
+            value -= offset
+        if low_limit <= value <= up_limit:
+            return value
+        else:
+            return ''
+    else:
+        return ''

create_data_tbl.py

@@ -0,0 +1,69 @@
+import time
+import traceback
+import pymysql
+
+
+class Mysql:
+    def __init__(self, host='', user='', passwd='', db='', port=3306, charset='utf8'):
+        self.host = host
+        self.user = user
+        self.passwd = passwd
+        self.db = db
+        self.port = port
+        self.charset = charset
+        self.conn = None
+        self.cursor = None
+        self._conn()
+
+    def _conn(self):
+        try:
+            self.conn = pymysql.connect(host=self.host, user=self.user, password=self.passwd, db=self.db, port=self.port)
+            self.cursor = self.conn.cursor(pymysql.cursors.DictCursor)
+            return True
+        except Exception as e:
+            print(e)
+            return False
+
+    def run(self):
+        self.create_data_table()
+
+    def get_stations_info(self):
+        sql = "SELECT * FROM station_info_tbl WHERE status = 1;"
+        self.cursor.execute(sql)
+        results = self.cursor.fetchall()
+        return results
+
+    def get_device_by_station_name(self, station_name):
+        sql = "select DISTINCT (device_name) from data_point_tbl where station_name = '%s'" % station_name
+        self.cursor.execute(sql)
+        results = self.cursor.fetchall()
+        return results
+
+    def get_point_by_device_name(self, device_name):
+        sql = "SELECT * FROM data_point_tbl WHERE device_name = '%s'" % device_name
+        self.cursor.execute(sql)
+        results = self.cursor.fetchall()
+        return results
+
+    def create_data_table(self):
+        stations_list = self.get_stations_info()
+        for each_station in stations_list:
+            devices_list = self.get_device_by_station_name(each_station['station_name'])
+            for each_device in devices_list:
+                points_list = self.get_point_by_device_name(each_device['device_name'])
+                print(points_list)
+                print("------------")
+                table_name = 'table_' + each_device['device_name']
+                sql_c = "CREATE TABLE IF NOT EXISTS %s (id bigint primary key , times datetime NOT NULL,INDEX (times)) ENGINE=InnoDB DEFAULT CHARSET=utf8;" % table_name
+                self.cursor.execute(sql_c)
+                for i in points_list:
+                    dataType = i['storage_type']
+                    columnName = "c" + str(i['serial_number'])
+                    sql_add = "ALTER TABLE %s ADD %s  %s comment '%s'" % (table_name, columnName, dataType, i['io_point_name'])
+                    print(f"sql_add: {sql_add}")
+                    self.cursor.execute(sql_add)
+                print(table_name, "done !")
+
+
+if __name__ == '__main__':
+    Mysql(host='127.0.0.1', user='root', passwd='zzZZ4144670..', db='shucai').run()

gateway.spec

@@ -5,7 +5,7 @@ block_cipher = None
 
 
 a = Analysis(['gateway.py'],
-             pathex=['E:\\WorkSpace\\shuicai\\60Mship\\venv\\Lib\\site-packages'],
+             pathex=['E:\\WorkSpace\\DataCollectionDevice\\venv\\Lib\\site-packages'],
              binaries=[],
              datas=[],
              hiddenimports=[],

hard_disk_storage.py

@@ -1,5 +1,5 @@
 import datetime
-import logging
+from logging_config import logger
 import openpyxl
 import pymysql
 import traceback
@@ -16,7 +16,6 @@ class HardDiskStorage:
         self.port = port
         self.charset = charset
         self.conn = None
-        self.logger = logging.getLogger()
         if not self._conn():
             self._reConn()
 
@@ -25,7 +24,7 @@ class HardDiskStorage:
             self.conn = pymysql.connect(host=self.host, user=self.user, password=self.passwd, database=self.db, port=self.port, autocommit=True)
             return True
         except Exception as e:
-            self.logger.error(f'failed to connect to {self.host}:{self.port}:{self.db} by [{self.user}:{self.passwd}]:{e}')
+            logger.error(f'failed to connect to {self.host}:{self.port}:{self.db} by [{self.user}:{self.passwd}]:{e}')
             return False
 
     def _reConn(self, num=28800, stime=3):  # 重试连接总次数为1天,这里根据实际情况自己设置,如果服务器宕机1天都没发现就......

historical_data_storage.py

@@ -2,7 +2,7 @@ import json
 import threading
 from event_storage import EventStorage
 import time
-from sanic.log import logger
+from logging_config import logger
 
 
 class HistoricalDataStorage(threading.Thread):

logging_config.py

@@ -12,8 +12,8 @@ LOGGING_CONFIG = dict(
     disable_existing_loggers=False,
     loggers={
         "sanic.root": {
-            "level": "DEBUG",
-            "handlers": ["console", "custome_console"]
+            "level": "INFO",  # 默认DEBUG
+            "handlers": ["console"]
         },
         "sanic.error": {
             "level": "INFO",
@@ -22,11 +22,18 @@ LOGGING_CONFIG = dict(
             "qualname": "sanic.error",
         },
         "sanic.access": {
-            "level": "WARNING",
+            "level": "INFO",  # 默认DEBUG
             "handlers": ["access_console"],
             "propagate": True,
             "qualname": "sanic.access",
         },
+        # 新曾自定义日志，用于数据采集程序
+        "sanic.my": {
+            "level": "INFO",
+            "handlers": ["my_console", "file_console"],
+            "propagate": True,
+            "qualname": "my.debug",
+        },
     },
     handlers={
         "console": {
@@ -44,22 +51,35 @@ LOGGING_CONFIG = dict(
             "formatter": "access",
             "stream": sys.stdout,
         },
-        "custome_console": {
+        # 数据采集程序控制台输出handler
+        "my_console": {
+            "class": "logging.StreamHandler",
+            "formatter": "my",
+            "stream": sys.stdout,
+        },
+        # 数据采集程序文件输出handler
+        "file_console": {
             'class': 'logging.handlers.RotatingFileHandler',
             'filename': 'app.log',
-            'level': 'INFO',
-            'maxBytes': 100*1024,
+            'maxBytes': 200 * 1024,
             'delay': True,
-            "formatter": "generic",
-            "backupCount": 10
+            "formatter": "my",
+            "backupCount": 10,
+            "encoding": "utf-8"
         },
     },
     formatters={
         "generic": {
-            "format": "%(asctime)s [%(process)d] [%(filename)s:%(lineno)d] [%(levelname)s] %(message)s",
+            "format": "%(asctime)s [%(process)d] [%(levelname)s] %(message)s",
             "datefmt": "[%Y-%m-%d %H:%M:%S %z]",
             "class": "logging.Formatter",
         },
+        # 自定义文件格式化器
+        "my": {
+            "format": "%(asctime)s [%(filename)s:%(lineno)d] [%(levelname)s] %(message)s",
+            "datefmt": "[%Y-%m-%d %H:%M:%S]",
+            "class": "logging.Formatter",
+        },
         "access": {
             "format": "%(asctime)s - (%(name)s) [%(levelname)s][%(host)s]: "
                       + "%(request)s %(message)s %(status)d %(byte)d",
@@ -68,6 +88,4 @@ LOGGING_CONFIG = dict(
         },
     },
 )
-logger = logging.getLogger("sanic.root")
-error_logger = logging.getLogger("sanic.error")
-access_logger = logging.getLogger("sanic.access")
+logger = logging.getLogger("sanic.my")

memory_storage.py

@@ -15,7 +15,7 @@ class MemoryStorage:
         try:
             pipe = self.conn.pipeline(transaction=True)
             for key_name in data_dict.keys():
-                pipe.set(key_name, data_dict[key_name], ex=600)  # redis过期时间10mines
+                pipe.set(key_name, data_dict[key_name], ex=1800)  # redis过期时间30mines
             pipe.execute()
         except Exception as e:
             print(e)