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