all the opposite

This commit is contained in:
əlemi 2022-04-19 02:26:40 +02:00
parent 631ff0bc2c
commit 49fe4f185f
No known key found for this signature in database
GPG key ID: BBCBFE5D7244634E
5 changed files with 91 additions and 102 deletions

View file

@ -1,2 +1 @@
from .treepuncher import Treepuncher from .treepuncher import Treepuncher, ConfigObject, Addon, Notifier
from .addon import Addon, ConfigObject

View file

@ -13,8 +13,7 @@ from dataclasses import dataclass, MISSING, fields
from setproctitle import setproctitle from setproctitle import setproctitle
from .treepuncher import Treepuncher, MissingParameterError from .treepuncher import Treepuncher, MissingParameterError, Addon, ConfigObject
from .addon import Addon, ConfigObject
from .helpers import configure_logging from .helpers import configure_logging
def main(): def main():

View file

@ -1,69 +0,0 @@
import json
from typing import Any, Dict, List, Set, get_type_hints, get_args, get_origin
from dataclasses import dataclass, MISSING, fields
def parse_with_hint(val:str, hint:Any) -> Any:
if hint is bool:
if val.lower() in ['1', 'true', 't', 'on', 'enabled']:
return True
return False
if hint is list or get_origin(hint) is list:
if get_args(hint):
return [ parse_with_hint(x, get_args(hint)[0]) for x in val.split() ]
return val.split()
if hint is set or get_origin(hint) is set:
if get_args(hint):
return set( parse_with_hint(x, get_args(hint)[0]) for x in val.split() )
return set(val.split())
if hint is dict or get_origin(hint) is dict:
return json.loads(val)
return (get_origin(hint) or hint)(val) # try to instantiate directly
class ConfigObject:
def __getitem__(self, key: str) -> Any:
return getattr(self, key)
class Addon:
name: str
config: ConfigObject
_client: 'Treepuncher'
@dataclass(frozen=True)
class Options(ConfigObject):
pass
@property
def client(self) -> 'Treepuncher':
return self._client
def __init__(self, client: 'Treepuncher', *args, **kwargs):
self._client = client
self.name = type(self).__name__
cfg = self._client.config
opts: Dict[str, Any] = {}
cfg_clazz = get_type_hints(type(self))['config']
if cfg_clazz is not ConfigObject:
for field in fields(cfg_clazz):
default = field.default if field.default is not MISSING \
else field.default_factory() if field.default_factory is not MISSING \
else MISSING
if cfg.has_option(self.name, field.name):
opts[field.name] = parse_with_hint(self._client.config[self.name].get(field.name), field.type)
elif default is MISSING:
repr_type = field.type.__name__ if isinstance(field.type, type) else str(field.type) # TODO fix for 3.8 I think?
raise ValueError(
f"Missing required value '{field.name}' of type '{repr_type}' in section '{self.name}'"
)
else: # not really necessary since it's a dataclass but whatever
opts[field.name] = default
self.config = self.Options(**opts)
self.register()
def register(self):
pass
async def initialize(self):
pass
async def cleanup(self):
pass

View file

@ -1,26 +0,0 @@
from typing import Callable, List
from .addon import Addon
class Notifier(Addon): # TODO this should be an Addon too!
_report_functions : List[Callable]
def __init__(self):
self._report_functions = []
def add_reporter(self, fn:Callable):
self._report_functions.append(fn)
return fn
def report(self) -> str:
return '\n'.join(str(fn()).strip() for fn in self._report_functions)
def notify(self, text, log:bool = False, **kwargs):
print(text)
async def initialize(self):
pass
async def cleanup(self):
pass

View file

@ -5,7 +5,7 @@ import asyncio
import datetime import datetime
import uuid import uuid
from typing import List, Dict, Optional, Any, Type, get_args, get_origin, get_type_hints, Set from typing import List, Dict, Optional, Any, Type, get_args, get_origin, get_type_hints, Set, Callable
from time import time from time import time
from dataclasses import dataclass, MISSING, fields from dataclasses import dataclass, MISSING, fields
from configparser import ConfigParser from configparser import ConfigParser
@ -16,10 +16,96 @@ from aiocraft.mc.packet import Packet
from aiocraft.mc.auth import AuthInterface, AuthException, MojangAuthenticator, MicrosoftAuthenticator, OfflineAuthenticator from aiocraft.mc.auth import AuthInterface, AuthException, MojangAuthenticator, MicrosoftAuthenticator, OfflineAuthenticator
from .storage import Storage, SystemState from .storage import Storage, SystemState
from .notifier import Notifier
from .addon import Addon
from .game import GameState, GameChat, GameInventory, GameTablist, GameWorld from .game import GameState, GameChat, GameInventory, GameTablist, GameWorld
def parse_with_hint(val:str, hint:Any) -> Any:
if hint is bool:
if val.lower() in ['1', 'true', 't', 'on', 'enabled']:
return True
return False
if hint is list or get_origin(hint) is list:
if get_args(hint):
return [ parse_with_hint(x, get_args(hint)[0]) for x in val.split() ]
return val.split()
if hint is set or get_origin(hint) is set:
if get_args(hint):
return set( parse_with_hint(x, get_args(hint)[0]) for x in val.split() )
return set(val.split())
if hint is dict or get_origin(hint) is dict:
return json.loads(val)
return (get_origin(hint) or hint)(val) # try to instantiate directly
class ConfigObject:
def __getitem__(self, key: str) -> Any:
return getattr(self, key)
class Addon:
name: str
config: ConfigObject
_client: 'Treepuncher'
@dataclass(frozen=True)
class Options(ConfigObject):
pass
@property
def client(self) -> 'Treepuncher':
return self._client
def __init__(self, client: 'Treepuncher', *args, **kwargs):
self._client = client
self.name = type(self).__name__
cfg = self._client.config
opts: Dict[str, Any] = {}
cfg_clazz = get_type_hints(type(self))['config']
if cfg_clazz is not ConfigObject:
for field in fields(cfg_clazz):
default = field.default if field.default is not MISSING \
else field.default_factory() if field.default_factory is not MISSING \
else MISSING
if cfg.has_option(self.name, field.name):
opts[field.name] = parse_with_hint(self._client.config[self.name].get(field.name), field.type)
elif default is MISSING:
repr_type = field.type.__name__ if isinstance(field.type, type) else str(field.type) # TODO fix for 3.8 I think?
raise ValueError(
f"Missing required value '{field.name}' of type '{repr_type}' in section '{self.name}'"
)
else: # not really necessary since it's a dataclass but whatever
opts[field.name] = default
self.config = self.Options(**opts)
self.register()
def register(self):
pass
async def initialize(self):
pass
async def cleanup(self):
pass
class Notifier(Addon): # TODO this should be an Addon too!
_report_functions : List[Callable]
def __init__(self):
self._report_functions = []
def add_reporter(self, fn:Callable):
self._report_functions.append(fn)
return fn
def report(self) -> str:
return '\n'.join(str(fn()).strip() for fn in self._report_functions)
def notify(self, text, log:bool = False, **kwargs):
print(text)
async def initialize(self):
pass
async def cleanup(self):
pass
class MissingParameterError(Exception): class MissingParameterError(Exception):
pass pass