|
|
|
|
import os
|
|
|
|
|
import asyncio
|
|
|
|
|
import importlib
|
|
|
|
|
from typing import Dict, Optional, List, Union
|
|
|
|
|
|
|
|
|
|
from logging.config import dictConfig
|
|
|
|
|
from logging import getLogger, Logger
|
|
|
|
|
|
|
|
|
|
from ..variables.loader import Datavars
|
|
|
|
|
from ..commands.commands import Command, CommandRunner
|
|
|
|
|
from .utils.responses import ResponseStructure
|
|
|
|
|
# from .utils.workers import Worker
|
|
|
|
|
|
|
|
|
|
from .schemas.config import ConfigSchema
|
|
|
|
|
|
|
|
|
|
from calculate.utils.tools import Singleton
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Получаем конфигурацию сервера.
|
|
|
|
|
TESTING = bool(os.environ.get("TESTING", False))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if not TESTING:
|
|
|
|
|
from .config import config
|
|
|
|
|
server_config = ConfigSchema(**config)
|
|
|
|
|
else:
|
|
|
|
|
from tests.server.config import config
|
|
|
|
|
server_config = ConfigSchema(**config)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class ServerData(metaclass=Singleton):
|
|
|
|
|
def __init__(self, config: ConfigSchema = server_config):
|
|
|
|
|
self.event_loop = asyncio.get_event_loop()
|
|
|
|
|
|
|
|
|
|
self._variables_path = config.variables_path
|
|
|
|
|
|
|
|
|
|
# Конфигурируем логгирование.
|
|
|
|
|
dictConfig(config.logger_config)
|
|
|
|
|
self.logger: Logger = getLogger("main")
|
|
|
|
|
self.log_message = {'DEBUG': self.logger.debug,
|
|
|
|
|
'INFO': self.logger.info,
|
|
|
|
|
'WARNING': self.logger.warning,
|
|
|
|
|
'ERROR': self.logger.error,
|
|
|
|
|
'CRITICAL': self.logger.critical}
|
|
|
|
|
|
|
|
|
|
self._datavars: Optional[Datavars] = None
|
|
|
|
|
|
|
|
|
|
# Словарь описаний команд.
|
|
|
|
|
self.commands: Dict[str, Command] = self._get_commands_descriptions(
|
|
|
|
|
config.commands_path)
|
|
|
|
|
|
|
|
|
|
# Словарь CID и экземпляров команд, передаваемых воркерам.
|
|
|
|
|
self.commands_runners: Dict[str, CommandRunner] = {}
|
|
|
|
|
|
|
|
|
|
# Словарь WID и экземпляров процессов-воркеров, передаваемых воркерам.
|
|
|
|
|
# TODO добавить менеджера воркеров.
|
|
|
|
|
# self.workers: Dict[int, Worker] = {}
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def datavars(self):
|
|
|
|
|
if self._datavars is not None:
|
|
|
|
|
return self._datavars
|
|
|
|
|
else:
|
|
|
|
|
self._datavars = self._load_datavars(self._variables_path,
|
|
|
|
|
self.logger)
|
|
|
|
|
return self._datavars
|
|
|
|
|
|
|
|
|
|
def _load_datavars(vars_path: str, logger: Logger) -> Datavars:
|
|
|
|
|
return Datavars(variables_path=vars_path,
|
|
|
|
|
logger=logger)
|
|
|
|
|
|
|
|
|
|
def _get_commands_descriptions(self, commands_path: str
|
|
|
|
|
) -> Dict[str, Command]:
|
|
|
|
|
'''Метод для получения совокупности описаний команд.'''
|
|
|
|
|
output = {}
|
|
|
|
|
package = ".".join(commands_path.split("/"))
|
|
|
|
|
|
|
|
|
|
for entry in os.scandir(commands_path):
|
|
|
|
|
if (not entry.name.endswith('.py')
|
|
|
|
|
or entry.name in {"commands.py", "__init__.py"}):
|
|
|
|
|
continue
|
|
|
|
|
module_name = entry.name[:-3]
|
|
|
|
|
try:
|
|
|
|
|
module = importlib.import_module("{}.{}".format(package,
|
|
|
|
|
module_name))
|
|
|
|
|
for obj_name in dir(module):
|
|
|
|
|
obj = module.__getattribute__(obj_name)
|
|
|
|
|
if isinstance(obj, Command):
|
|
|
|
|
output[obj.id] = obj
|
|
|
|
|
except Exception:
|
|
|
|
|
continue
|
|
|
|
|
return output
|
|
|
|
|
|
|
|
|
|
def make_command(self, command_id: str, ) -> int:
|
|
|
|
|
'''Метод для создания команды по ее описанию.'''
|
|
|
|
|
# command_description = self.commands[command_id]
|
|
|
|
|
|
|
|
|
|
# def _get_worker_object(self, wid: Optional[int] = None) -> Worker:
|
|
|
|
|
# '''Метод для получения воркера для команды.'''
|
|
|
|
|
# if wid is not None:
|
|
|
|
|
# worker = Worker(wid, self._event_loop, self._datavars)
|
|
|
|
|
# self._workers[wid] = worker
|
|
|
|
|
# elif not self._workers:
|
|
|
|
|
# worker = Worker(0, self._event_loop, self._datavars)
|
|
|
|
|
# self._workers[0] = worker
|
|
|
|
|
# else:
|
|
|
|
|
# wid = max(self._workers.keys()) + 1
|
|
|
|
|
# worker = Worker(wid, self._event_loop, self._datavars)
|
|
|
|
|
# self._workers[wid] = worker
|
|
|
|
|
# return worker
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class Commands:
|
|
|
|
|
'''Предварительная реализация контейнера описаний команд.'''
|
|
|
|
|
def __init__(self, *commands: List[Command]):
|
|
|
|
|
self._by_id: dict = dict()
|
|
|
|
|
self._by_command: dict = dict()
|
|
|
|
|
for command in commands:
|
|
|
|
|
self._by_id[command.id] = command
|
|
|
|
|
self._by_command[command.command] = command
|
|
|
|
|
|
|
|
|
|
def get_commands(self, base_url: str) -> Dict[str, dict]:
|
|
|
|
|
response = dict()
|
|
|
|
|
|
|
|
|
|
for command in self._by_id.values():
|
|
|
|
|
data = ResponseStructure(base_url)
|
|
|
|
|
data.add_data(id=command.id,
|
|
|
|
|
title=command.title,
|
|
|
|
|
category=command.category,
|
|
|
|
|
command=command.command)
|
|
|
|
|
data.add_link("self", f"/commands/{command.id}?by_id=true")
|
|
|
|
|
data.add_link("parameters", f"/commands/{command.id}/parameters")
|
|
|
|
|
data.add_link("configure", f"/configs/{command.id}")
|
|
|
|
|
response[command.id] = data.get_dict()
|
|
|
|
|
|
|
|
|
|
return response
|
|
|
|
|
|
|
|
|
|
def get_by_id(self, command_id: str, base_url: str) -> Dict[str, Dict]:
|
|
|
|
|
if command_id in self._by_id:
|
|
|
|
|
command = self._by_id[command_id]
|
|
|
|
|
data = ResponseStructure(base_url)
|
|
|
|
|
data.add_data(id=command.id,
|
|
|
|
|
title=command.title,
|
|
|
|
|
category=command.category,
|
|
|
|
|
command=command.command)
|
|
|
|
|
data.add_link("self", f"/commands/{command.id}?by_id=1")
|
|
|
|
|
data.add_link("parameters", f"/commands/{command.id}/parameters")
|
|
|
|
|
data.add_link("configure", f"/configs/{command.id}")
|
|
|
|
|
parameters_data = self._get_parameters_data(command.parameters,
|
|
|
|
|
command.id, base_url)
|
|
|
|
|
data.embed("parameters", parameters_data)
|
|
|
|
|
return data.get_dict()
|
|
|
|
|
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
def find_command(self, console_command: str,
|
|
|
|
|
base_url: str) -> Union[dict, None]:
|
|
|
|
|
if console_command in self._by_command:
|
|
|
|
|
command = self._by_command[console_command]
|
|
|
|
|
|
|
|
|
|
data = ResponseStructure(base_url)
|
|
|
|
|
data.add_data(id=command.id,
|
|
|
|
|
title=command.title,
|
|
|
|
|
category=command.category,
|
|
|
|
|
command=command.command)
|
|
|
|
|
data.add_link("self", f"/commands/{command.id}?by_id=1")
|
|
|
|
|
data.add_link("parameters", f"/commands/{command.id}/parameters")
|
|
|
|
|
data.add_link("configure", f"/configs/{command.id}")
|
|
|
|
|
parameters_data = self._get_parameters_data(command.parameters,
|
|
|
|
|
command.id, base_url)
|
|
|
|
|
data.embed("parameters", parameters_data)
|
|
|
|
|
|
|
|
|
|
return data.get_dict()
|
|
|
|
|
else:
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
def get_parameters(self, command_id: str, base_url: str
|
|
|
|
|
) -> Union[dict, None]:
|
|
|
|
|
if command_id in self._by_id:
|
|
|
|
|
command = self._by_id[command_id]
|
|
|
|
|
parameters_data = self._get_parameters_data(command.parameters,
|
|
|
|
|
command.id, base_url)
|
|
|
|
|
return parameters_data
|
|
|
|
|
else:
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
def get_command_object(self, command_id: str) -> Command:
|
|
|
|
|
if command_id in self._by_id:
|
|
|
|
|
return self._by_id[command_id]
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
def _get_parameters_data(self, parameters: list, command_id: str,
|
|
|
|
|
base_url: str) -> List[dict]:
|
|
|
|
|
data = ResponseStructure(base_url)
|
|
|
|
|
data.add_data(parameters)
|
|
|
|
|
data.add_link("self", f"/commands/{command_id}/parameters")
|
|
|
|
|
return data.get_dict()
|