#!/usr/bin/env python3 # -*- coding: utf-8 -*- from typing import Self, Any, TypeVar from Interfaces.Application.CXCVInterface import CXCVInterface from Abstracts.ProcedureAbstract import ProcedureAbstract from Utils.Utils import Utils T = TypeVar("T") class ProceduresManager: def __init__(self:Self, cxcv:CXCVInterface) -> None: key:str self.cxcv:CXCVInterface = cxcv self.__procedures:dict[str, dict[str, ProcedureAbstract]] = {} for key in ("default_procedures_models", "procedures_models"): self.cxcv.models.add(ProcedureAbstract, self.cxcv.settings.get(key), True) for key in ( "default_procedures_files", "default_procedures", "procedures_files", "procedures" ): self.add(self.cxcv.settings.get(key), True) def get(self:Self, Type:type[T], procedure:tuple[str, str]|T) -> T|None: if isinstance(procedure, tuple): key:str via:str key, via = procedure return self.__procedures[key][via] if ( key in self.__procedures and via in self.__procedures[key] and isinstance(self.__procedures[key][via], Type) ) else None return procedure if isinstance(procedure, T) else None def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None: subinputs:dict[str, dict[str, ProcedureAbstract]] for subinputs in self.cxcv.files.load_json(inputs): via:str vias:dict[str, ProcedureAbstract|str] for key, vias in subinputs.items(): key:str subinputs:dict[str, Any|None] if key not in self.__procedures: self.__procedures[key] = {} for via, subinputs in vias.items(): if overwrite or via not in self.__procedures[key]: ProcedureModel:type[ProcedureAbstract] = self.cxcv.models.get( ProcedureAbstract, Utils.get_value("model", subinputs, key) ) if ProcedureModel is not None: self.__procedures[key][via] = ProcedureModel(self.cxcv, key, via, subinputs)