module_universal_io/tools/digital_io.py

from io_module import IO_Module
from modbus import Modbus, MBError, NoResponseError
from log_reader import DigitalLogReader
import colorama
from colorama import Fore
from time import sleep
import time
from datetime import datetime, timedelta, timezone
from mb_registers import DIO_REGS
from serial import Serial


class IO_Digital(IO_Module):

    def __init__(self, modbus: Modbus):
        self.modbus = modbus
        super().__init__(self.modbus)
        self.log = DigitalLogReader(self.modbus)
 
    '''Запрос параметров'''
    # 0x0100 - текущее состояние входов
    def get_inputs_bit(self) -> str:
        data = self.modbus.read_holding_registers(DIO_REGS['in_bits'], 1)
        return format(data[0], '08b')

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

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

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

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

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

    # 0x0130
    def get_loads_bit(self):
        data = self.modbus.read_holding_registers(DIO_REGS['load_bits'], 1)
        return format(data[0], '08b')

    # 0x0131
    def get_cred_bit(self):
        data = self.modbus.read_holding_registers(DIO_REGS['cred_bits'], 1)
        return format(data[0], '08b')

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

    '''Установка параметров'''
    def set_inputs_mode(self, val):
        self.modbus.write_holding_register(DIO_REGS['in_mode'], val)

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

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

    # Установить значение на конкретном выходе [1..8]
    def set_output(self, output, val):
        ret = self.modbus.read_holding_registers(DIO_REGS['out_cur'], 1)
        if val == 1:
            data = ret[0] | (0b1 << (output - 1))
        else:
            data = ret[0] & ~(0b1 << (output - 1))
        self.set_outputs(data)


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

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

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

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

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

    def print_loads(self):
        # Значения датчиков нагрузки (битовое поле)
        print('Loads values [bit field] :', Fore.GREEN + self.get_loads_bit())
        # Слово достоверности датчиков нагрузки (битовое поле)
        print('Credibility loads [bit field] :', Fore.GREEN + self.get_cred_bit())


class IO_DigitalTester:

    def __init__(self, dev_tester: IO_Digital, dev_dut: IO_Digital):
        self.tester = dev_tester
        self.dut = dev_dut

    '''Управление DUT'''

    # Подать/снять питание на DUT
    def dut_switch(self, state: bool):
        self.tester.set_output(1, state)

    # Установить 1 на нечетных входах DUT
    def dut_set_odd_inputs(self):
        self.tester.set_output(2, False)

    # Установить 1 на четный входах DUT
    def dut_set_even_imputs(self):
        self.tester.set_output(2, True)

    '''Тест входов'''
    def test_inputs(self):
        self.dut_set_even_imputs()
        time.sleep(0.1)
        self.dut.print_inputs()
        time.sleep(1)
        self.dut_set_odd_inputs()
        time.sleep(0.1)
        self.dut.print_inputs()
        time.sleep(1)

    '''Тест выходов и датчика обрыва нагрузки'''
    def test_load(self):
        # Все выходы DUT разомкнуты
        print("Все выходы DUT разомкнуты...")
        self.dut.set_outputs(0)
        self.dut.print_loads()
        time.sleep(0.1)
        # Замкнуть все выходы DUT (лампочка должна гореть)
        print("Все выходы DUT замкнуты...")
        self.dut.set_outputs(0b11111111)
        time.sleep(0.1)
        
        for i in range(1, 9):
            # Разомкнуть i-ый выход DUT
            print(f"Разомкнуть выход {i}")
            self.dut.set_output(i, 0)
            time.sleep(1)
            self.dut.print_loads()
            self.dut.set_output(i, 1)
            time.sleep(1)

    def get_load(self):
        self.dut.set_outputs(0b11110111)
        # self.dut.set_outputs(0)
        time.sleep(1)
        self.dut.print_loads()


def main():
    colorama.init(autoreset=True)
    
    serial_port = Serial('COM22', 115200, timeout=0.05, parity='N', xonxoff=False)
    
    modbus_tester = Modbus(serial_port, 1)
    # dev_tester = IO_Digital(modbus_tester)
    dio = IO_Digital(modbus_tester)


    '''Тесты отдельного модуля DIO'''
    dio.sys.get_system_vars()

    # dio.sys.set_info(b'00010000000000000000000000000000')    
    # dio.sys.set_info('This is super DIO!')

    # dio.print_inputs()
    # dio.get_inputs_counters()

    # modbus_dut = Modbus(serial_port, 2)
    # dev_dut = IO_Digital(modbus_dut)

    # tester = IO_DigitalTester(dev_tester, dev_dut)

    '''Включить DUT'''
    # tester.dut_switch(True)

    '''Запросить системные настройки DUT'''
    # dev_dut.sys.get_system_vars()

    '''Тест входов. Переключение значений на входах DUT'''
    # for i in range(10):
    #     tester.test_inputs()

    '''Тестирование выходов и датчиков обрыва нагрузки'''
    # tester.test_load()
    # for i in range(100):
    #     tester.get_load()




    '''Системные переменные и параметры'''
    # dev_tester.sys.get_system_vars()

    ''' Установить текущее время с учетом часового пояса'''
    # dev.sys.set_rtc()     
    

    '''Лог и архив. Настройки лога.'''
    # dev.log.get_log_info()
    # dev.log.get_random_entries()
    # for i in range(8):
    #     dev.log.set_archive_period(5 + i*2, i)
    
    # dev.log.get_log_info()

    # dev.log.get_all_archive()
    # dev.log.get_all_log()
    
    # dev.log.log_clear()
    # dev.log.archive_clear()

    '''Лог'''
    # dev.log.get_all_log()

    '''Архив'''
    # dev.log.get_all_archive()
    # print(dev.log.get_archive_entry(0, 1))

    '''Сохранение настроек'''
    # dev.sys.save_sattings()

    '''Настройки'''

    '''Регистры модуля'''
    # dev.print_inputs()

    '''Настройи модуля'''
    # for i in range(1, 9):
    #     dev.set_input_mode(i, 1)
        # print('Inputs mode [bit field]   :', Fore.GREEN + dev.get_inputs_mode())
    # dev.set_input_mode(4, 1)
    # dev.print_inputs()

    '''Обновление'''
    # dio.updater.update('fw.bin', 'MAI_12')
    
    
if __name__ == '__main__':
    main()