module_universal_io/tools/log_reader.py

from modbus import Modbus
from mb_registers import reg_table
from colorama import Fore
from random import randint
from time import sleep
import colorama
import struct

ARCHIVE_ENTRY = 0x06
LOG_ENTRY = 0x07


class LogReader():
    def __init__(self, modbus: Modbus):
        self.modbus = modbus
        self.log_capacity = 0
        self.log_entries_number = 0
        self.archive_capacity = 0
        self.archive_entries_number = 0    
        self.archive_period = 0
        colorama.init(autoreset=True)

    def get_archive(self):
        print("LogReader")

    def get_log_info(self):
        data = self.modbus.read_holding_registers(reg_table['log_info'], 5)
        self.log_capacity = data[0]
        self.log_entries_number = data[1]
        self.archive_capacity = data[2]
        self.archive_entries_number = data[3]
        self.archive_period = data[4]

        print(Fore.CYAN + "Log and archive params:\n")
        print('Log capacity           :', Fore.CYAN + str(self.log_capacity))
        print('Log entries number     :', Fore.CYAN + str(self.log_entries_number))
        print('Archive capacity       :', Fore.CYAN + str(self.archive_capacity))
        print('Archive entries number :', Fore.CYAN + str(self.archive_entries_number))
        print('Archive period         :', Fore.CYAN + str(self.archive_period), end='\n')

    def set_archive_period(self, value):
        self.modbus.write_holding_register(reg_table['archive_per'], value)

class DigitalLogReader(LogReader):
    def __init__(self, modbus: Modbus):
        super().__init__(modbus)

    def get_archive(self):
        print("DigitalLogReader")

    def get_random_archive_entry(self):
        data = self.modbus.read_file_record(ARCHIVE_ENTRY, randint(1, self.archive_entries_number), 1)
        entry = struct.unpack('<QBB', data[5:15])
        print(Fore.CYAN + str(entry))

    def get_random_log_entry(self):
        data = self.modbus.read_file_record(LOG_ENTRY, randint(1, self.log_entries_number), 1)
        entry = struct.unpack('<QBBBfB', data[5:21])
        print(Fore.CYAN + str(entry))

    def get_random_entries(self):
        '''Читаем лог и архив'''
        print(Fore.CYAN + "\nEntries:\n")
        while (1):
            self.get_random_archive_entry()
            sleep(0.1)
            self.get_random_log_entry()
            sleep(0.1)



class AnalogInputLogReader(LogReader):
    def __init__(self):
        super().__init__()

    def get_archive(self):
        print("AnalogInputLogReader")



def main():

    module = DigitalLogReader('COM24', 115200, 15)

    module.get_log_info()
    # module.get_archive()
    # module.set_archive_period(10)

    return

    # for i in range(500):
    while (1):
        module.get_random_archive_entry()
        sleep(0.1)
        module.get_random_log_entry()
        sleep(0.1)



if __name__ == '__main__':
    main()