bemaniutils/bemani/data/config.py

import copy
import os
from sqlalchemy.engine import Engine  # type: ignore
from typing import Any, Dict, Optional, Set

from bemani.common.constants import GameConstants
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))


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 event_log_duration(self) -> Optional[int]:
        duration = self.get('event_log_duration')
        return int(duration) if duration else None