You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

198 lines
7.9 KiB

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()