106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Self, Any
|
|
from threading import Thread
|
|
from re import Match as REMatch
|
|
from Interfaces.Application.CXCVInterface import CXCVInterface
|
|
from Models.CommandsModel import CommandsModel
|
|
from Utils.Patterns import RE
|
|
from Utils.Utils import Utils
|
|
|
|
class TerminalManager:
|
|
|
|
def __init__(self:Self, cxcv:CXCVInterface) -> None:
|
|
|
|
key:str
|
|
|
|
self.cxcv:CXCVInterface = cxcv
|
|
self.__started:bool = False
|
|
self.__commands:tuple[CommandsModel, ...] = ()
|
|
self.__thread:Thread = Thread(target=self.__execute)
|
|
|
|
def start(self:Self) -> None:
|
|
if self.__started:
|
|
return
|
|
self.__started = True
|
|
|
|
for key in ("default_commands_files", "default_commands", "commands_files", "commands"):
|
|
self.add(self.cxcv.settings.get(key), True)
|
|
|
|
self.__thread.start()
|
|
|
|
def __execute(self:Self) -> None:
|
|
while self.cxcv.working:
|
|
|
|
order:str = input().strip()
|
|
|
|
if not order:
|
|
continue
|
|
|
|
matches:REMatch = RE.COMMAND.match(order)
|
|
|
|
if not matches:
|
|
continue
|
|
|
|
name:str = matches.group(1)
|
|
arguments:list[Any|None] = []
|
|
parameters:dict[str, Any|None] = {}
|
|
arguments_block:str|None = (matches.group(2) or "").strip()
|
|
done:bool = False
|
|
|
|
if arguments_block:
|
|
|
|
block:tuple[str|None, ...]
|
|
|
|
for block in RE.get_all(RE.COMMAND_ARGUMENT, arguments_block):
|
|
|
|
key:str = Utils.get_from(block, (1, 5, 6, 7))
|
|
value:str|None = Utils.get_from(block, (2, 3, 4))
|
|
|
|
if value is not None:
|
|
parameters[key] = value.strip()
|
|
else:
|
|
arguments += [key,]
|
|
|
|
for command in self.__commands:
|
|
if (done := command.execute(name, parameters, *arguments)):
|
|
break
|
|
|
|
done or print(f"Command '{name}' not found.")
|
|
|
|
def get_i_command(self, names:str|list[str]|tuple[str, ...]) -> int|None:
|
|
for i, command in enumerate(self.__commands):
|
|
if command.check(names):
|
|
return i
|
|
return None
|
|
|
|
def add(self:Self, inputs:Any|None, overwrite:bool = False) -> None:
|
|
if isinstance(inputs, CommandsModel):
|
|
|
|
i:int|None = self.get_i_command(inputs)
|
|
|
|
if i is not None:
|
|
overwrite and self.__commands[i].update(inputs)
|
|
else:
|
|
self.__commands += (inputs,)
|
|
|
|
elif isinstance(inputs, dict):
|
|
if "names" in inputs and isinstance(inputs["names"], (str, list, tuple)):
|
|
|
|
names:str|list[str]|tuple[str, ...] = inputs["names"]
|
|
action:callable[[dict[str, Any|None], list[Any|None]], None]|None = inputs["action"] if "action" in inputs else None
|
|
i:int|None = self.get_i_command(names)
|
|
|
|
if i is not None:
|
|
overwrite and self.__commands[i].update(names, action)
|
|
else:
|
|
self.__commands += (CommandsModel(self.cxcv, names, action),)
|
|
|
|
else:
|
|
for command in self.cxcv.files.load_json(inputs, True):
|
|
if isinstance(command, (CommandsModel, dict)):
|
|
self.add(command, overwrite)
|
|
|
|
def close(self:Self, parameters:dict[str, Any|None], *arguments:list[Any|None]) -> None:
|
|
self.cxcv.close() |