import copy import os from sqlalchemy.engine import Engine # type: ignore from typing import Any, Dict, Optional, Set from bemani.common import GameConstants, RegionConstants from bemani.data.types import ArcadeID class Database: def __init__(self, parent_config: "Config") -> None: self.__config = parent_config @property def address(self) -> str: return str(self.__config.get('database', {}).get('address', 'localhost')) @property def database(self) -> str: return str(self.__config.get('database', {}).get('database', 'bemani')) @property def user(self) -> str: return str(self.__config.get('database', {}).get('user', 'bemani')) @property def password(self) -> str: return str(self.__config.get('database', {}).get('password', 'bemani')) @property def engine(self) -> Engine: engine = self.__config.get('database', {}).get('engine') if engine is None: raise Exception("Config object is not instantiated properly, no SQLAlchemy engine present!") if not isinstance(engine, Engine): raise Exception("Config object is not instantiated properly, engine property is not a SQLAlchemy Engine!") return engine @property def read_only(self) -> bool: return bool(self.__config.get('database', {}).get('read_only', False)) class Server: def __init__(self, parent_config: "Config") -> None: self.__config = parent_config @property def address(self) -> str: return str(self.__config.get('server', {}).get('address', '127.0.0.1')) @property def keepalive(self) -> str: return str(self.__config.get('server', {}).get('keepalive', self.address)) @property def port(self) -> int: return int(self.__config.get('server', {}).get('port', 80)) @property def https(self) -> bool: return bool(self.__config.get('server', {}).get('https', False)) @property def uri(self) -> Optional[str]: uri = self.__config.get('server', {}).get('uri') return str(uri) if uri else None @property def redirect(self) -> Optional[str]: redirect = self.__config.get('server', {}).get('redirect') return str(redirect) if redirect else None @property def enforce_pcbid(self) -> bool: return bool(self.__config.get('server', {}).get('enforce_pcbid', False)) @property def pcbid_self_grant_limit(self) -> int: return int(self.__config.get('server', {}).get('pcbid_self_grant_limit', 0)) @property def region(self) -> int: region = int(self.__config.get('server', {}).get('region', RegionConstants.USA)) if region in {RegionConstants.EUROPE, RegionConstants.NO_MAPPING}: # Bogus values we support. return region if region < RegionConstants.MIN or region > RegionConstants.MAX: # Pick the original default for the network (USA). return RegionConstants.USA # Region was fine. return region class Client: def __init__(self, parent_config: "Config") -> None: self.__config = parent_config @property def address(self) -> str: address = self.__config.get('client', {}).get('address') if address is None: raise Exception("Config object is not instantiated properly, no client address present!") return str(address) class Machine: def __init__(self, parent_config: "Config") -> None: self.__config = parent_config @property def pcbid(self) -> str: pcbid = self.__config.get('machine', {}).get('pcbid') if pcbid is None: raise Exception("Config object is not instantiated properly, no machine pcbid present!") return str(pcbid) @property def arcade(self) -> Optional[ArcadeID]: return self.__config.get('machine', {}).get('arcade') class PASELI: def __init__(self, parent_config: "Config") -> None: self.__config = parent_config @property def enabled(self) -> bool: return bool(self.__config.get('paseli', {}).get('enabled', False)) @property def infinite(self) -> bool: return bool(self.__config.get('paseli', {}).get('infinite', False)) class WebHooks: def __init__(self, parent_config: "Config") -> None: self.discord = DiscordWebHooks(parent_config) class DiscordWebHooks: def __init__(self, parent_config: "Config") -> None: self.__config = parent_config def __getitem__(self, key: GameConstants) -> Optional[str]: uri = self.__config.get('webhooks', {}).get('discord', {}).get(key.value) return str(uri) if uri else None class Config(dict): def __init__(self, existing_contents: Dict[str, Any] = {}) -> None: super().__init__(existing_contents or {}) self.database = Database(self) self.server = Server(self) self.client = Client(self) self.paseli = PASELI(self) self.webhooks = WebHooks(self) self.machine = Machine(self) def clone(self) -> "Config": # Somehow its not possible to clone this object if an instantiated Engine is present, # so we do a little shenanigans here. engine = self.get('database', {}).get('engine') if engine is not None: self['database']['engine'] = None clone = Config(copy.deepcopy(self)) if engine is not None: self['database']['engine'] = engine clone['database']['engine'] = engine return clone @property def filename(self) -> str: filename = self.get('filename') if filename is None: raise Exception("Config object is not instantiated properly, no filename present!") return os.path.abspath(str(filename)) @property def support(self) -> Set[GameConstants]: support = self.get('support') if support is None: raise Exception("Config object is not instantiated properly, no support list present!") if not isinstance(support, set): raise Exception("Config object is not instantiated properly, support property is not a Set!") return support @property def secret_key(self) -> str: return str(self.get('secret_key', 'youdidntchangethisatalldidyou?')) @property def name(self) -> str: return str(self.get('name', 'e-AMUSEMENT Network')) @property def email(self) -> str: return str(self.get('email', 'nobody@nowhere.com')) @property def cache_dir(self) -> str: return os.path.abspath(str(self.get('cache_dir', '/tmp'))) @property def theme(self) -> str: return str(self.get('theme', 'default')) @property def event_log_duration(self) -> Optional[int]: duration = self.get('event_log_duration') return int(duration) if duration else None