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 time
import colorama
import struct
from datetime import datetime, timedelta, timezone

ARCHIVE_ENTRY = 0x06
LOG_ENTRY = 0x07



class LogParser:

    events = {1: 'Включение питания/перезагрузка',
              2: 'Перевод времени',
              3: 'Обновление ПО'}

    utc_offset = 10800

    def __init__(self) -> None:
        t = datetime.now(timezone.utc).astimezone()
        self.utc_offset = t.utcoffset() // timedelta(seconds=1)
        

    @staticmethod
    def print_entry(entry: tuple, index: int):
        timestamp = time.ctime(entry[0]/1000 - LogParser.utc_offset)
        ans = f"[LOG] {index}: {LogParser.events[entry[1]]}, {timestamp}, state: {entry[2]}, channel: {entry[3]}, value: {entry[4]}"
        print(Fore.CYAN + ans)




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_archive_entry(self, index):
        data = self.modbus.read_file_record(ARCHIVE_ENTRY, index, 1)        
        return struct.unpack('<QBB', data[5:15])
        
    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_log_entry(self, index):
        data = self.modbus.read_file_record(LOG_ENTRY, index, 1)
        return struct.unpack('<QBBBfB', data[5:21])

    def print_log_entry(self, index):
        data = self.modbus.read_file_record(LOG_ENTRY, index, 1)
        LogParser.print_entry(struct.unpack('<QBBBfB', data[5:21]), index)

    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)

    def get_all_archive(self):
        self.get_log_info()
        
        for i in range(1, self.archive_entries_number + 1):
            print(self.get_archive_entry(i))

    def get_all_log(self):
        self.get_log_info()
        for i in range(1, self.log_entries_number + 1):
            self.print_log_entry(i)
            sleep(0.01)


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