tmp

exer/time_1.py

@@ -55,9 +55,10 @@ def test_5():
 
 
 
+print(time.ctime(time.time()))
 
 # test_1()
 # test_2()
 # test_3()
 # test_4()
-test_5()
+# test_5()

hw_libs/io_module.py

@@ -1,5 +1,11 @@
 from modbus import Modbus, MBError, NoResponseError
+import colorama
+from colorama import Fore
 import time
+import os
+
+reg_table = {'in_bits': 0x0100, 'uptime': 0x0800, 'rtc_unix': 0x0802, 'rtc_sinhro': 0x0804}
+
 
 class IO_Module(Modbus):
     def __init__(self, tty: str, brate: int, address: int):
@@ -23,19 +29,84 @@ class IO_Module(Modbus):
         if response[:3] != bytes((self.address, 0x41, 0x01)):
             raise MBError('Incorrect response')
 
-    def update(self):
+    def write_fw_part(self, data: bytes):
+        """Write piece of FW data in IAP mode"""
+        header = bytes((self.address, 0x41, 0x02))
+        request = b''.join((header, self.update_segment_number.to_bytes(2, 'big'), data))
+        response = self.raw_communicate(request + self._crc(request), 5)
+        # self.print_hex(response)
+        if len(response) != 5:
+            raise MBError('Incorrect response length')
+        if (response[:3]) != header:
+            raise MBError('Incorrect response')
+        self.update_segment_number += 1
+
+    def iap_finish(self):
+        """Complete FW transmission and check response"""
+        header = request = bytes((self.address, 0x41, 0x03))
+        response = self.raw_communicate(request + self._crc(request), 5)
+        if len(response) != 5:
+            raise MBError('Incorrect response length')
+        if response[:3] != header:
+            raise MBError('Incorrect response')
+
+    def update(self, path):
+        self.MB_TIMEOUT = 3
+        size = os.path.getsize('fw.bin')
+        print('Switch to IAP mode')
         self.iap_start()
         time.sleep(4)
-        self.write_fw_start(123)
+        print(f'Start writing {size} bytes of FW')
+        self.write_fw_start(size)
+        time.sleep(2)
+        print(f'Open FW file "{path}"...')
+        with open(path, 'rb') as f:
+            done = progress_cur = progress_pre = 0
+            while True:
+                buf = f.read(128)
+                if len(buf):
+                    self.write_fw_part(buf)
+                    progress_cur = done / size
+                else:
+                    break
+        print('End of transmission')
+        self.iap_finish()
 
+    def get_input_bit(self) -> str:
+        data = self.read_holding_registers(reg_table['in_bits'], 1)
+        return format(data[0], '08b')
 
+    def get_uptime(self):
+        return self.read_uint32_holding(reg_table['uptime'])
 
+    def get_rtc(self):
+        return self.read_uint32_holding(reg_table['rtc_unix'])
+
+    def set_rtc(self, utc):
+        self.write_uint32(reg_table['rtc_sinhro'], utc)
+
+    
 
 def main():
+    colorama.init(autoreset=True)
     dev = IO_Module('COM22', 115200, 1)
+    dev.MB_DEBUG = True
+    # dev.update('fw.bin')
 
-    dev.update()
+    # Запрос системных параметров, установка времени
+    
+    # print('Device uptime:', dev.get_uptime())
+    # unix_time = dev.get_rtc()
+    # print(f'RTC: {time.ctime(unix_time)}. Unix time stamp:  {unix_time}')
+    # print('Set time:', int(time.time()))
+    # dev.set_rtc(int(time.time()))
+    # time.sleep(1)
+    # unix_time = dev.get_rtc()
+    # print(f'RTC: {time.ctime(unix_time)}. Unix time stamp:  {unix_time}')
     
+    while True:
+        print(Fore.GREEN + 'Inputs values [bit field]:', Fore.YELLOW + dev.get_input_bit())
+        time.sleep(1)
     
 
 if __name__ == '__main__':

hw_libs/modbus.py

@@ -1,6 +1,7 @@
 import time
 from serial import Serial
 import colorama
+import struct
 from colorama import Fore, Style 
 from typing import Sequence, Union
 from binascii import b2a_hex
@@ -50,7 +51,7 @@ class Modbus(ModbusMixin):
     # MB_CRC_TABLE: Sequence[int] = DEFAULT_MB_CRC_TABLE
     MB_TIMEOUT: float = 0.05
     MB_CRC_TABLE: Sequence[int] = DEFAULT_MB_CRC_TABLE
-    MB_DEBUG: bool = True
+    MB_DEBUG: bool = False
     MB_TRIES: int = 3
 
     def __init__(self, tty: str, brate: int, address: int):
@@ -140,6 +141,28 @@ class Modbus(ModbusMixin):
     def read_holding_registers(self, address: int, count: int) -> Sequence[int]:
         return self._read_registers(3, address, count)
 
+    def read_holding_registers_raw(self, address: int, count: int) -> bytes:
+        return self._read_registers(3, address, count, True)
+
+    def read_uint32_holding(self, address: int) -> int:
+        """Read 32-bit integer from holding registers"""
+        return struct.unpack('>I', self.read_holding_registers_raw(address, 2))[0]
+
+    # 0x10
+    def write_holding_registers_raw(self, address:int, values: bytes):
+        """Write 16-bit integers to holding registers on device"""
+        request = bytearray(address.to_bytes(2, 'big'))
+        request += (len(values) // 2).to_bytes(2, 'big')
+        request += len(values).to_bytes(1, 'big')
+        request += values
+        response = self.mb_func(16, bytes(request), 4)
+        if request[:4] != response:
+            raise MBError('Incorrect response payload')
+    
+    def write_uint32(self, address: int, value: int):
+        """Write 32-bit integer to holding register"""
+        self.write_holding_registers_raw(address, struct.pack('>I', value))
+
     def test_send(self, data: bytes):
         while True:
             self.serial.write(data)
@@ -156,42 +179,17 @@ class Modbus(ModbusMixin):
         #     time.sleep(1)
 
 def main():
+    ADDR = 1
     colorama.init()
-    # st = 'Hello world'
-    # st.encode('utf-8')
-    
-    # print(st.encode('utf-8'))
-    dev = Modbus('COM22', 115200, 1)
-    # to_send = b'\x01\x03\x01\x00\x00\x01'
-    # dev.communicate(to_send, 7)
-
-    # Пример чтения одного регистра по адресу 0x0100
-    while True:
-        print(dev.read_holding_registers(0x0100, 1))
-        time.sleep(1)
-    # dev._read_registers(0x03, 0x0100, 1, False)
-    # print(dev._read_registers(0x03, 0x0100, 1, True))
+    dev = Modbus('COM22', 115200, ADDR)
     
-    
-    # dev.communicate()
-    # dev.test_send(bytes('hello world\r\n', encoding="utf_8"))
-    # tx_data = 'hello world\r\n'
-    # tx_data = 'abc'
-    # dev.send_recv(bytes(tx_data, encoding="utf_8"), len(tx_data))
-
-    # Modbus.test()
-
-    # print(i.to_bytes(2, 'big'))
-    # print(Modbus.crc(i.to_bytes(1, 'big')))
-    # print(Modbus.crc(b'\x00\x01'))
-    # crc = Modbus.crc(b'\x00\x01')
-    # dev.raw_communicate(b'\x00\x01')
-    # print(ModbusMixin.print_hex(crc))
-    # print(Modbus.crc(b'\xd0\x00'))
+
+    # print(dev.read_holding_registers(0x0100, 1))
+    # print(dev.read_uint32_holding(0x0100))
+    # time.sleep(1)
+
 
 if __name__ == '__main__':
     main()
-    # a = 11
-    # print(a.to_bytes(2, 'big'))