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, 'in_cnt': 0x0102, 'in_mode': 0x0120, 'in_norm': 0x0122, 'in_deb_start': 0x124,
             'out_cur': 0x0200, 'out_mode': 0x0202, 'out_mode_save': 0x0203, 'pwm_duty': 0x0210,
             'pwm_duty_save': 0x0220, 'pwm_per': 0x0230, 'pwm_per_save': 0x0240, 
             'rtc_unix': 0x0802, 'rtc_sinhro': 0x0804, 'uptime': 0x0800,}


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

    # 0x0100 - текущее состояние входов
    def get_inputs_bit(self) -> str:
        data = self.read_holding_registers(reg_table['in_bits'], 1)
        return format(data[0], '08b')

    # 0x0101 - 0x0110 Счетчики импульсов
    def get_inputs_counters(self):
        data = []
        for i in range(reg_table['in_cnt'], reg_table['in_cnt'] + 16, 2):
            data.append(self.read_uint32_holding(i))
        return data                              

    # 0x0120 - режим работы входов
    def get_inputs_mode(self):
        data = self.read_holding_registers(reg_table['in_mode'], 1)
        return format(data[0], '08b')

    def set_inputs_mode(self, val):
        self.write_holding_register(reg_table['in_mode'], val)

    #        
    def set_input_mode(self, input, val):
        ret = self.read_holding_registers(reg_table['in_mode'], 1)
        if val == 1:
            data = ret[0] | (0b1 << (input - 1))
        else:
            data = ret[0] & ~(0b1 << (input - 1))
        self.set_inputs_mode(data)

    # 0x0122 - нормальное состояние входов
    def get_inputs_norm_state(self):
        data = self.read_holding_registers(reg_table['in_norm'], 1)
        return format(data[0], '08b')

    def set_inputs_norm_state(self, val):
        self.write_holding_register(reg_table['in_norm'], val)

    # 0x0124 - время антидребезга (ms)
    def get_debounce_channel(self, input):
        data = self.read_holding_registers(reg_table['in_deb_start'] + input - 1, 1)
        return data[0]

    def get_debounce_channels(self):
        return self.read_holding_registers(reg_table['in_deb_start'], 8)

    def set_debounce_channel(self, input, val):
        self.write_holding_register(reg_table['in_deb_start'] + input - 1, val)

    # 0x0200 - текущее состояние выходов в обычно режиме
    def get_outputs(self):
        data = self.read_holding_registers(reg_table['out_cur'], 1)
        return format(data[0], '08b')

    # 0x0200 - текущее состояние выходов в обычно режиме
    def set_outputs(self, val):
        self.write_holding_register(reg_table['out_cur'], val)

    def set_output(self, output, val):
        ret = self.read_holding_registers(reg_table['out_cur'], 1)
        if val == 1:
            data = ret[0] | (0b1 << (output - 1))
        else:
            data = ret[0] & ~(0b1 << (output - 1))
        self.set_outputs(data)

    # 0x0202 - режим работы выходов; 0 - обычный, 1 - ШИМ
    def get_outputs_mode(self):
        data = self.read_holding_registers(reg_table['out_mode'], 1)
        return format(data[0], '08b')
    
    def set_outputs_mode(self, val):
        self.write_holding_register(reg_table['out_mode'], val)

    def set_output_mode(self, output, val):
        ret = self.read_holding_registers(reg_table['out_mode'], 1)
        if val == 1:
            data = ret[0] | (0b1 << (output - 1))
        else:
            data = ret[0] & ~(0b1 << (output - 1))
        self.set_outputs_mode(data)

    # 0x0203 - состояние выходов (режим обычного выхода) в безопасном режиме работы
    def get_outputs_mode_save(self):
        data = self.read_holding_registers(reg_table['out_mode_save'], 1)
        return format(data[0], '08b')

    def set_outputs_mode_save(self, val):
        self.write_holding_register(reg_table['out_mode_save'], val)

    def set_output_mode_save(self, output, val):
        ret = self.read_holding_registers(reg_table['out_mode_save'], 1)
        if val == 1:
            data = ret[0] | (0b1 << (output - 1))
        else:
            data = ret[0] & ~(0b1 << (output - 1))
        self.set_outputs_mode_save(data)

    # 0x0210 - заполнение PWM (%)
    def get_pwm_duty(self, output):
        data = self.read_holding_registers(reg_table['pwm_duty'] + output - 1, 1)
        return data[0]
    
    def get_pwm_duty_all(self):
        return self.read_holding_registers(reg_table['pwm_duty'], 8)

    def set_pwm_duty(self, output, val):
        self.write_holding_register(reg_table['pwm_duty'] + output - 1, val)

    # 0x0220 - заполнение PWM (%)
    def get_pwm_duty_all_save(self):
        return self.read_holding_registers(reg_table['pwm_duty_save'], 8)

    # 0x0230 - период PWM (0.1 сек)
    def get_pwm_period_all(self):
        return self.read_holding_registers(reg_table['pwm_per'], 8)

    # 0x0240 - период PWM в безопасном режиме (0.1 сек)
    def get_pwm_period_all_save(self):
        return self.read_holding_registers(reg_table['pwm_per_save'], 8)

    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('COM24', 115200, 14)
    dev.MB_DEBUG = False
    
    trigger = 0
    # 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}')
    
    
    
    for i in range(1, 9):
        dev.set_input_mode(i, 1)
        # print('Inputs mode [bit field]   :', Fore.GREEN + dev.get_inputs_mode())

    for i in range(1, 9):
        dev.set_debounce_channel(i, 50 + i)
        dev.set_output(i, 0)
        
    # 0x0203 - состояние выходов (режим обычного выхода) в безопасном режиме работы
    for i in range(1, 9):
        dev.set_output_mode_save(i, 0)

    # 0x0210 - заполнение PWM (ms)
    for i in range(1, 9):
        dev.set_pwm_duty(i, 10 + i)    

    # Установить 1-ый выход в режим PWM
    # dev.set_output_mode(1, 0)

    # Установить нормальное состояние входов
    # dev.set_inputs_norm_state(0b00110101)

    # print(dev.get_pwm_duty_all())

    while True:
        print('~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~')
        # Значения входов (битовое поле)
        print('Inputs values [bit field] :', Fore.GREEN + dev.get_inputs_bit())

        # Режим работы входов (битовое поле)
        print('Inputs mode [bit field]   :', Fore.GREEN + dev.get_inputs_mode())

        # Нормальное состояние входов (битовое поле)
        print('Inputs norm [bit field]   :', Fore.GREEN + dev.get_inputs_norm_state())

        # Период антидребезга (ms)
        print('Debounce input (ms)       :', Fore.GREEN + ' | '.join(str(el) for el in dev.get_debounce_channels()))

        # Значение счетчиков
        data = dev.get_inputs_counters()
        print('Inputs counters           :', Fore.GREEN + ' | '.join(str(el) for el in data))

        # Текущее состояние выходов в обычном режиме
        print('Outputs norm [bit field]  :', Fore.GREEN + dev.get_outputs())

        # Состояние выходов в безопасном режиме
        print('Outputs save [bit field]  :', Fore.GREEN + dev.get_outputs_mode_save())

        # Режим работы выходов
        print('Outputs mode [bit field]  :', Fore.GREEN + dev.get_outputs_mode())

        # 0x0210 - заполнение PWM (ms)
        print('PWM duty cycle [%]        :', Fore.GREEN + ' | '.join(str(el) for el in dev.get_pwm_duty_all()))

        # 0x0220 - заполнение PWM в безопасном режиме (ms)
        print('PWM duty cycle (save) [%] :', Fore.GREEN + ' | '.join(str(el) for el in dev.get_pwm_duty_all_save()))

        # 0x0230 - период PWM (0.1 сек)
        print('PWM period [0.1 sec]      :', Fore.GREEN + ' | '.join(str(el) for el in dev.get_pwm_period_all()))
        
        # 0x0240 - период PWM в безопасном режиме (0.1 сек)
        print('PWM period save [0.1 sec] :', Fore.GREEN + ' | '.join(str(el) for el in dev.get_pwm_period_all_save()))

        # Дергаем одним выходом        
        # dev.set_output(2, trigger)
        # trigger = not trigger

        # # Для проверки выходов в обычном режиме
        # for i in range(1, 9):
        #     dev.set_output(i, 1)
        #     print('Outputs norm [bit field]  :', Fore.GREEN + dev.get_output())
        #     time.sleep(0.1)

        # for i in range(1, 9):
        #     dev.set_output(i, 0)
        #     print('Outputs norm [bit field]  :', Fore.GREEN + dev.get_output())
        #     time.sleep(0.1)

        # # Режим работы выходов
        # for i in range(1, 9):
        #     dev.set_output_mode(i, 1)
        #     print('Outputs mode [bit field]  :', Fore.GREEN + dev.get_outputs_mode())
        #     time.sleep(0.1)

        # for i in range(1, 9):
        #     dev.set_output_mode(i, 0)
        #     print('Outputs mode [bit field]  :', Fore.GREEN + dev.get_outputs_mode())
        #     time.sleep(0.1)

        # break
        time.sleep(1)



    # for i in range(1, 9):
    #     dev.set_input_mode(i, 1)
    #     print('Inputs mode [bit field]  :', Fore.GREEN + dev.get_inputs_mode())
    

    # for i in range(8, 0, -1):
    #     dev.set_input_mode(i, 0)
    #     print('Inputs mode [bit field]  :', Fore.GREEN + dev.get_inputs_mode())
        

if __name__ == '__main__':
    main()