codewars/hw_libs/io_module.py

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):
        super().__init__(tty, brate, address)
        self.update_segment_number = 0

    def iap_start(self):
        """Reboot device in IAP mode"""
        request = bytes((self.address, 0x41, 0x01))
        response = self.raw_communicate(request + self._crc(request))

    def write_fw_start(self, size: int):
        """Ask device to start update"""
        self.update_segment_number = 0
        request = bytes((self.address, 0x41, 0x01, 0xEF, 0xBE, 0xAD, 0xDE)) + size.to_bytes(4, 'big')
        response = self.raw_communicate(request + self._crc(request), 5)
        if len(response) == 0:
            raise NoResponseError('No response on WRITE_START command')
        if len(response) != 5:
            raise MBError('Incorrect response length')
        if response[:3] != bytes((self.address, 0x41, 0x01)):
            raise MBError('Incorrect response')

    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)
        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')

    # Запрос системных параметров, установка времени
    
    # 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__':
    main()