#!/usr/bin/env python3 # -*- coding: utf-8 -*- from typing import Any, Callable, Self, Optional from Interfaces.Application.AnPInterface import AnPInterface from Abstracts.AIInterpretersAbstract import AIInterpretersAbstract from Models.AIResponseModel import AIResponseModel from Utils.Common import Common class AIInterpretersManager: def __init__(self:Self, anp:AnPInterface) -> None: self.anp:AnPInterface = anp self.__interpreters:dict[str, AIInterpretersAbstract] = {} self.update() def update(self:Self) -> None: key:str for key in ("default_ai_interpreters_files", "ai_interpreters_files", "default_ai_interpreters", "ai_interpreters"): self.add(self.anp.settings.get(key), True) def reset(self:Self) -> None: self.__interpreters = {} self.update() def close(self:Self) -> None: key:str for key in list(self.__interpreters.keys()): self.remove(key) def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None: subinputs:dict[str, Any|None] for subinputs in self.anp.files.load_json(inputs, True): key:str value:Any|None for key, value in subinputs.items(): if overwrite or key not in self.__interpreters: interpreter:AIInterpretersAbstract|None = None if isinstance(value, AIInterpretersAbstract): interpreter = value elif isinstance(value, dict) and "type" in value: Model:AIInterpretersAbstract|None = self.anp.models.get(AIInterpretersAbstract, value["type"]) if Model is not None: interpreter = Model(self.anp, key, value) if interpreter is not None: if key in self.__interpreters: self.remove(key) self.__interpreters[key] = interpreter def remove(self:Self, keys:Optional[str|list[str]] = None) -> None: key:str for key in Common.get_keys(keys) if keys else list(self.__interpreters.keys()): if key in self.__interpreters: self.__interpreters[key].close() del self.__interpreters[key] def request(self:Self, key:str, session:int|None, message:str, orders:str|list[str] = [], callback:Optional[Callable[[int, AIResponseModel], None]] = None, custom_context:list[int] = [] ) -> tuple[int|None, AIResponseModel|None]: response:AIResponseModel|None = None if key in self.__interpreters: session, response = self.__interpreters[key].request(session, message, orders, callback, custom_context) return session, response