#!/usr/bin/env python3 # -*- coding: utf-8 -*- from typing import Self, Any, Optional, Sequence import datetime from re import Match as REMatch from traceback import extract_tb as extract_traceback, format_stack as trace_format_stack from time import time as timestamp, sleep from Managers.I18NManager import I18NManager from Managers.SettingsManager import SettingsManager from Managers.PrintTypesManager import PrintTypesManager from Managers.TerminalManager import TerminalManager from Managers.ModelsManager import ModelsManager from Managers.UniqueKeysManager import UniqueKeysManager from Managers.QueusManager import QueuesManager from Managers.SessionsManager import SessionsManager from Managers.ControllersManager import ControllersManager from Managers.DispatchesManager import DispatchesManager from Managers.IndexesManager import IndexesManager from Managers.RoutesManager import RoutesManager from Managers.WebSocketServersManager import WebSocketServersManager from Managers.HTTPServersManager import HTTPServersManager from Managers.PseudoLoRAsManager import PseudoLoRAsManager from Managers.AIInterpretersManager import AIInterpretersManager from Utils.Common import Common from Utils.Patterns import RE class AnP: def __own_update(self:Self) -> None: self.__end_print_types = [ print_type.upper() for print_type in Common.get_keys(self.settings.get("end_print_types", None, self.__end_print_types)) ] self.__root_projects_paths = Common.get_texts(self.settings.get("root_projects_paths", None, self.__root_projects_paths)) self.__print_format = self.settings.get("print_format", None, self.__print_format) self.__exception_format = self.settings.get("exception_format", None, self.__exception_format) def __init__(self:Self, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None: self.__working:bool = True self.settings:SettingsManager = SettingsManager(self, inputs) self.i18n:I18NManager = I18NManager(self) self.print_types:PrintTypesManager = PrintTypesManager(self) self.__end_print_types:list[str] = ["UNKN", "EXCE", "ERRO"] self.__root_projects_paths:list[str] = [] self.__print_format:str = "[{type}] {yyyy}{mm}{dd} {hh}{ii}{ss} [{line}]{file}({method}): {message}" self.__exception_format:str = " '[{line}]{file}({method})'{lines}\n\n{exception_message}" self.__own_update() self.terminal:TerminalManager = TerminalManager(self) self.models:ModelsManager = ModelsManager(self) self.unique_keys:UniqueKeysManager = UniqueKeysManager(self) self.queues:QueuesManager = QueuesManager(self) self.sessions:SessionsManager = SessionsManager(self) self.controllers:ControllersManager = ControllersManager(self) self.dispatches:DispatchesManager = DispatchesManager(self) self.indexes:IndexesManager = IndexesManager(self) self.routes:RoutesManager = RoutesManager(self) self.web_socket_servers:WebSocketServersManager = WebSocketServersManager(self) self.http_servers:HTTPServersManager = HTTPServersManager(self) self.pseudoloras:PseudoLoRAsManager = PseudoLoRAsManager(self) self.ai_interpreters:AIInterpretersManager = AIInterpretersManager(self) def update(self:Self) -> None: self.settings.update() self.i18n.update() self.print_types.update() self.__own_update() self.terminal.update() self.models.update() self.unique_keys.update() self.queues.update() self.sessions.update() self.controllers.update() self.dispatches.update() self.indexes.update() self.routes.update() self.web_socket_servers.update() self.http_servers.update() self.pseudoloras.update() self.ai_interpreters.update() def reset(self:Self) -> None: self.settings.reset() self.i18n.reset() self.print_types.reset() self.terminal.reset() self.models.reset() self.unique_keys.reset() self.queues.reset() self.sessions.reset() self.controllers.reset() self.dispatches.reset() self.indexes.reset() self.routes.reset() self.web_socket_servers.reset() self.http_servers.reset() self.ai_interpreters.reset() def close(self:Self) -> None: self.__working = False self.queues.close() self.ai_interpreters.close() self.web_socket_servers.close() self.http_servers.close() def working(self:Self) -> bool: return self.__working def print(self:Self, _type:str, message:str|Sequence[str], inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None, i:int = 0 ) -> None: date:datetime.datetime = datetime.datetime.now() own:dict[str, Any] = { "raw_type" : _type, "type": self.print_types.get(_type), "message": self.i18n.get(message, inputs), **Common.get_dictionary(inputs), **Common.get_action_data(i + 1) } for key in ("year", "month", "day", "hour", "minute", "second"): k:str = "i" if key == "minute" else key[0] own[k] = own[key] = getattr(date, key) own[k + k] = ("00" + str(own[key]))[-2:] own["yyyy"] = ("0000" + str(own["year"]))[-4:] for root_path in self.__root_projects_paths: if own["file"].startswith(root_path): own["file"] = own["file"][len(root_path):] break if any(own["type"].startswith(end_type) for end_type in self.__end_print_types) and "end" in own: own["message"] += own["end"] print(Common.string_variables(self.__print_format, own)) def exception(self:Self, exception:Exception, message:str|Sequence[str], inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None, i:int = 0 ) -> None: lines:list[str] = extract_traceback(exception.__traceback__).format() matches:REMatch = RE.EXCEPTION.match(lines[-1]) data:dict[str, Any|None] = { **Common.get_dictionary(inputs), "lines" : "", "exception_message" : str(exception), "method" : matches.group(3), "line" : matches.group(2), "file" : matches.group(1) } block:str j:int for root_path in self.__root_projects_paths: if data["file"] and data["file"].startswith(root_path): data["file"] = data["file"][len(root_path):] for j, block in enumerate(trace_format_stack()[:-2] + lines): if block: data["lines"] += "\n " + str(j) + " - " + RE.NEW_LINE.split(block.strip())[0] data["end"] = Common.string_variables(self.__exception_format, data) message and self.print("exception", message, data, i + 2) def wait(self:Self, time:str|float) -> None: date:float = timestamp() while self.__working and timestamp() - date < float(time): sleep(.1)