module_universal_io/tools/analog_in.py

from io_module import IO_Module
from modbus import Modbus
from log_reader import AnalogInputLogReader
import colorama
from colorama import Fore
from time import sleep
from serial import Serial
from mb_registers import AI_REGS, AiReg
import matplotlib.pyplot as plt 
import matplotlib.animation as animation 

ai_name = {'AIN_1': 0, 'AIN_2': 1, 'AIN_3': 2, 'AIN_4': 3, 'AIN_5': 4, 
           'AIN_6': 5, 'AIN_7': 6, 'AIN_8': 7, 'AIN_9': 8, 'AIN_10': 9, 
           'AIN_11': 10, 'AIN_12': 11, 'V_ISO_CL': 12, 'V_ISO': 13,
           'CRNT_LIM_U_BFR_R': 14, 'CRNT_LIM_U_ABFR_R': 15,
           'AIN_FIL_1': 0, 'AIN_FIL_2': 1, 'AIN_FIL_3': 2, 'AIN_FIL_4': 3, 
           'AIN_FIL_5': 4, 'AIN_FIL_6': 5, 'AIN_FIL_7': 6, 'AIN_FIL_8': 7, 
           'AIN_FIL_9': 8, 'AIN_FIL_10': 9, 'AIN_FIL_11': 10, 'AIN_FIL_12': 11
           }

class IO_AnalogInput(IO_Module):

    GRAPTH_LEN = 100

    def __init__(self, modbus: Modbus):
        self.modbus = modbus
        super().__init__(self.modbus)
        self.log = AnalogInputLogReader(self.modbus)

        self.fig = plt.figure(1)
        self.input = self.fig.add_subplot(1, 1, 1)
        self.input.set_title('input_1')

        self.x = [0]
        self.data = []

        self.graph_input = ai_name['AIN_1']
        self.graph_input_func = 0

    # Чтение параметров

    def get_inputs_state(self):
        'Значения состояний входов вкл./выкл. (битовое поле)'
        data = self.modbus.read_holding_registers(AiReg.IN_STATE.value, 1)
        return format(data[0], '012b')


    def get_inputs_mode(self):
        'Режим измерения входов (0 - напряжение или 1 - ток. (битовое поле))'
        data = self.modbus.read_holding_registers(AiReg.IN_MODE.value , 1)
        return format(data[0], '012b')

    
    def get_inputs_alarm(self):
        'Аварии аналоговых входов (битовое поле)'
        data = self.modbus.read_holding_registers(AiReg.IN_FAILURE.value, 1)
        return format(data[0], '012b')


    def get_ext_sens_power(self):
        'Состояние питания внешних датчиков (вкл./выкл.)'
        data = self.modbus.read_holding_registers(AiReg.EXT_SENS_POWER.value, 1)


    def get_input_gain(self, channel):
        'Коэффициент усиления канала [1..12] (внутренний коэф-т ADC)'
        data = self.modbus.read_holding_registers(AiReg.IN_GAINE_FACTOR.value + channel - 1, 1)
        return data[0]


    def get_k_factor(self, channel):
        'Коэффициент пересчета K канала [1..12]'
        return self.modbus.read_float_holding(AiReg.IN_K_FACTOR.value + 2*(channel - 1))
        

    def get_b_factor(self, channel):
        'Коэффициент пересчета B канала [1..12]'
        return self.modbus.read_float_holding(AiReg.IN_B_FACTOR.value + 2*(channel - 1))


    def get_raw_input(self, channel):
        'Сырые данные по отдельному каналу [1..12]'
        return self.modbus.read_holding_registers(AiReg.IN_RAW.value + channel - 1, 1)[0]


    def get_raw_inputs(self):
        'Сырые данные по всем каналам'
        data = self.modbus.read_holding_registers(AiReg.IN_RAW.value, 16)
        return data


    def get_fil_inputs(self):
        'Фильтрованные данные'
        data = []
        for i in range(12):
            data.append(self.modbus.read_float_holding(AiReg.IN_FILTER.value + i*2))
        return data


    # Установка параметров


    def set_inputs_state(self, val):
        'Значения состояний входов вкл./выкл. (битовое поле)'
        self.modbus.write_holding_register(AiReg.IN_STATE.value , val)


    def set_inputs_mode(self, val):
        'Режим измерения входов (0 - напряжение или 1 - ток. (битовое поле))'
        self.modbus.write_holding_register(AiReg.IN_STATE.value, val)


    def set_ext_sens_power(self, val):
        'Состояние питания внешних датчиков (вкл./выкл.)'
        self.modbus.write_holding_register(AiReg.EXT_SENS_POWER.value , val)


    def set_input_gain(self, input, value):
        '''
        Коэффициент усиления канала [1..12] (внутренний коэф-т ADC)
        Допустимые значения: 1, 2, 4, 8, 16, 32, 64, 128
        '''
        if value not in (1, 2, 4, 8, 16, 32, 64, 128):
            return None
        self.modbus.write_holding_register(AiReg.IN_GAINE_FACTOR.value + input - 1, value)


    def set_k_factor(self, channel, value):
        'Коэффициент пересчета K канала [1..12]'
        self.modbus.write_float(AiReg.IN_K_FACTOR.value + 2*(channel - 1), value)


    def set_b_factor(self, channel, value):
        'Коэффициент пересчета K канала [1..12]'
        self.modbus.write_float(AiReg.IN_B_FACTOR.value + 2*(channel - 1), value)


    '''Настройки входов'''
    def print_inputs(self):
        print(Fore.GREEN + '____________________________________________')
        print(Fore.GREEN + 'Analog inputs settings:')

        # Значения состояний входов вкл./выкл. (битовое поле)
        print('Inputs state [bit field] :', Fore.GREEN + self.get_inputs_state())

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

        # Коэффициенты усиления
        for i in range(1, 13):
            print(f'Gain factor channel {i}    : ', end='')
            print(Fore.GREEN + str(self.get_input_gain(i)))

    '''Вывод параметров. Сырые данные 16 входов'''
    def print_raw_inputs(self):
        data = self.get_raw_inputs()
        print(f"[ADC raw] IN_1: {data[0]}, IN_2: {data[1]}, IN_3: {data[2]}, IN_4: {data[3]}")
        print(f"[ADC raw] IN_5: {data[4]}, IN_6: {data[5]}, IN_7: {data[6]}, IN_8: {data[7]}")
        print(f"[ADC raw] IN_9: {data[8]}, IN_10: {data[9]}, IN_11: {data[10]}, IN_12: {data[11]}")
        print(f"[ADC raw] V_ISO_CL: {data[12]}, V_ISO: {data[13]}")
        print(f"[ADC raw] CRNT_LIM_U_BFR_R: {data[14]}, CRNT_LIM_U_ABFR_R: {data[15]}")

    '''Вывод параметров. Фильтрованные данные 12 входов'''
    def print_filtered_inputs(self):
        data = self.get_fil_inputs()
        print(f"[ADC fil] IN_1: {data[0]}, IN_2: {data[1]}, IN_3: {data[2]}, IN_4: {data[3]}")
        print(f"[ADC fil] IN_5: {data[4]}, IN_6: {data[5]}, IN_7: {data[6]}, IN_8: {data[7]}")
        print(f"[ADC fil] IN_9: {data[8]}, IN_10: {data[9]}, IN_11: {data[10]}, IN_12: {data[11]}")

    '''Вывод данных на график'''
    def show_graph(self, func, channel: str):
        self.graph_input_func = func
        self.graph_input = ai_name[channel]
        ani = animation.FuncAnimation(self.fig, self.draw, interval=50)
        plt.show()

    def show_graph_filtered(self, channel: str):
        self.graph_input = ai_name[channel]

    def draw(self, i):
        data = self.graph_input_func()

        self.data.append(data[self.graph_input]) 

        self.input.clear()
        self.input.plot(self.x, self.data)

        if len(self.data) == self.GRAPTH_LEN:
            self.data.pop(0)
            self.x.pop(0)

        self.x.append(self.x[-1] + 1)

        # print(self.in1)
        # print(self.x)

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

    '''Режим работы аналоговых входов'''
    # ai.print_inputs()

    '''Установка коэффициентов усиления. Канал, коэффициент'''
    '''
    for channel in range(1, 13):
        ai.set_input_gain(channel, 8)

    for channel in range(1, 13):
        print(ai.get_input_gain(channel))
    '''

    '''Коэффициенты K и B'''
    # for i in range(1, 13):
    #     ai.set_k_factor(i, 1.0)
    #     ai.set_b_factor(i, 0.0)

    '''Последовательное включение входов по одному.'''
    '''
    while True:
        for i in range(12):
            ai.set_inputs_state(1 << i)        
            print(ai.get_inputs_state())
            sleep(1)
    '''            
    '''Настройка режима измерения'''
    '''
    while True:
        for i in range(12):
            ai.set_inputs_mode(1 << i)
            print(ai.get_inputs_mode())
            sleep(1)
    '''


    # ai.sys.get_system_vars()     
    
    # print(ai.get_inputs_state())
    ai.set_inputs_state(0b1111_1111_1111)
    # ai.set_inputs_state(0b0000_0000_0001)
    # ai.set_inputs_state(0b1111_1111_1111)
    # print(ai.get_inputs_state())
    
    # sleep(1)
    # ai.set_inputs_state(0b0)

    # print(ai.get_inputs_mode())
    # ai.set_inputs_mode(0b0000_0000_0000)
    # print(ai.get_inputs_mode())

    '''Питание внешних датчиков'''
    # ai.set_ext_sens_power(0)

    '''Аварии аналоговых входов'''
    # for i in range(100):
    #     print(ai.get_inputs_alarm())
    #     sleep(1)

    '''Данные каналов'''
    # while True:
    #     ai.print_raw_inputs()
    #     ai.print_filtered_inputs()
    #     sleep(1)

    # ai.get_raw_inputs()
    # ai.print_raw_inputs()

    '''Сырые данные'''
    # while True:
    #     print(ai.get_raw_input(12))
    #     sleep(1)

    '''Вывод на график. Сырые данные'''
    # ai.show_graph(ai.get_raw_inputs,  'AIN_1')

    '''Вывод на график. Фильтрованные данные'''
    # ai.show_graph(ai.get_fil_inputs, 'AIN_FIL_12')

    '''Системные настройки'''
    # ai.sys.get_system_vars()

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

    '''Обновление прошивки'''
    # serial_port.timeout = 1
    # modbus_tester.MB_DEBUG = True
    # ai.updater.update('fw.bin', 'MAI_12')
    

if __name__ == '__main__':
    main()