diff --git a/.vscode/settings.json b/.vscode/settings.json
deleted file mode 100644
index e555990..0000000
--- a/.vscode/settings.json
+++ /dev/null
@@ -1,19 +0,0 @@
-{
- "editor.codeActionsOnSave": {
- "source.fixAll.ruff": "always",
- "source.organizeImports.ruff": "always",
- "source.sort.json": "always"
- },
- "editor.formatOnSave": true,
- "python.languageServer": "None",
- "scss-to-css-compile.browsers": [
- "ie > 9",
- "iOS > 8",
- "Android >= 4.4",
- "ff > 38",
- "Chrome > 38"
- ],
- "scss-to-css-compile.compileOnSave": false,
- "scss-to-css-compile.outDir": "./",
- "scss-to-css-compile.output": "compressed"
-}
\ No newline at end of file
diff --git a/app.py b/app.py
index 50e8fd7..5cf36e8 100644
--- a/app.py
+++ b/app.py
@@ -1,32 +1,72 @@
import logging
import os
+import sys
+import threading
from threading import Thread
+from time import sleep
+from typing import final
from starfall.config import Config
from starfall.log import Log
-from starfall.web import App
+from starfall.types import SnapshotQueue
+from starfall.web import WebUI
+
+CURRENT_VERSION = "0.1.0-alpha.1"
+
+
+@final
+class Application:
+ app: WebUI
+ config: Config
+ log: Log
+ queue: SnapshotQueue
+ threads: dict[str, Thread]
+
+ def __init__(self):
+ self.log = Log()
+
+ logging.getLogger("app").debug(
+ "OS Type: %s\nWorking Directory: %s\nApplication version: %s",
+ os.name,
+ os.path.realpath(os.curdir),
+ CURRENT_VERSION,
+ )
+
+ self.config = Config("config.toml", "config-sample.toml")
+ if not self.config.get("app.run", False):
+ logging.getLogger("app").critical(
+ "Config file disallows execution.\n"
+ + "Please make the necessary changes to the configuration"
+ + " file and relaunch the program:\n\t%s",
+ os.path.realpath(self.config.file_name),
+ )
+ sys.exit()
+
+ self.threads = {}
+
+ self.queue = SnapshotQueue()
+
+ self.app = WebUI()
+
+ def start(self):
+ self.threads["flask"] = Thread(
+ target=self.app.run,
+ args=(
+ self.config,
+ self.queue,
+ ),
+ daemon=True,
+ )
+ self.threads["flask"].start()
+
+ while threading.active_count() > 0:
+ sleep(0.25)
+ self.monitor_queue()
+
+ def monitor_queue(self):
+ for entry in self.queue.get_all_for_receiver("app"):
+ logging.getLogger("app").info("Received queue message: %r", entry)
+
if __name__ == "__main__":
- Log.setup()
-
- try:
- Config.load()
- except FileNotFoundError:
- dst = Config.create()
- logging.getLogger().critical(
- msg="\n".join(
- [
- "*** Config file was not found ***",
- "",
- "Sample config.toml file has been created in the project's root directory:",
- "[b]" + os.path.realpath("./%s" % dst) + "[/b]",
- "Edit it and run the program again.",
- ]
- ),
- extra={"markup": True},
- )
- exit()
-
- App.__init__()
-
- threads = [Thread(App.run())]
+ Application().start()
diff --git a/config-sample.toml b/config-sample.toml
index 0c028b9..68c429c 100644
--- a/config-sample.toml
+++ b/config-sample.toml
@@ -1,5 +1,19 @@
# == StarfallBot configuration file ==
+[app]
+# Whether the application is allowed to start at all.
+run = false
+
+[discord]
+# Discord bot token.
+token = "YOUR BOT TOKEN HERE"
+
+[twitch]
+# Twitch OAuth token.
+oauth_token = "oauth:TWITCH OAUTH TOKEN"
+# Twitch channel.
+channel = "TWITCH CHANNEL NAME"
+
[web]
# What IP the web server is bound to.
# Use "localhost" until the bot is production ready.
@@ -12,10 +26,3 @@ port = 5000
database_url = "sqlite:///starfallbot.db"
# Secret key, used to sign the session cookie.
secret_key = "PRIVATE SECRET KEY CHANGEME"
-
-[discord]
-token = "YOUR BOT TOKEN HERE"
-
-[twitch]
-oauth_token = "oauth:TWITCH OAUTH TOKEN"
-channel = "TWITCH CHANNEL NAME"
\ No newline at end of file
diff --git a/starfall/config.py b/starfall/config.py
index 2f812c1..ac9f268 100644
--- a/starfall/config.py
+++ b/starfall/config.py
@@ -1,48 +1,66 @@
+import logging
import os
from shutil import copy
-from typing import Any
+from typing import Any, final
import tomllib
+@final
class Config:
- data: dict[str, Any] = dict()
- """Configuration imported from config.toml."""
+ """Provides a wrapper for toml config files."""
- @classmethod
- def create(cls) -> str:
- """Copy config-sample.toml file to config.toml.
+ def __init__(self, file_name: str, sample_file_name: str):
+ """Read in a config file and provide it as a :obj:`Config` object.
- Does not alert the user to the modification.
+ Args:
+ file_name (str): Path to a TOML config file.
+ sample_file_name (str): Path to a sample TOML config file.
+ Used if :param:`file_name` does not exist.
"""
- return copy(src="config-sample.toml", dst="config.toml")
- @classmethod
- def load(cls) -> None:
- """Loads the contents of the config.toml file into the class.
+ self.file_name = file_name
+ """str: Path to a TOML config file."""
- Raises:
- FileNotFoundError: If the file does not exist.
- """
- if not cls.file_exists():
- raise FileNotFoundError("config.toml does not exist")
- with open("config.toml", "rb") as f:
- cls.data = tomllib.load(f)
+ self.sample_file_name = sample_file_name
+ """str: Path to a sample TOML config file. Used if :attr:`file_name` does not exist."""
- @classmethod
- def file_exists(cls) -> bool:
- """Whether the config.toml file exists or needs to be created.
+ self._toml_data = {}
+ """dict: TOML data, parsed to a nested dict."""
+
+ try:
+ with open(str(self.file_name), "rb") as f:
+ self._toml_data = tomllib.load(f)
+ except OSError as error:
+ file_name = copy(str(self.sample_file_name), str(self.file_name))
+ logging.getLogger().warning(
+ "Could not read config file %r, reason: %r",
+ os.path.realpath(file_name),
+ error,
+ )
+
+ def get(self, key: str, default: Any = None):
+ """Recursively calls `get(key, default)` to search nested dictionary.
+
+ Examples:
+ Fetch a nested configuration value.
+
+ >>> config = Config("config.toml", "config-sample.toml")
+ >>> config.get("web.run", False)
+ True
Returns:
- bool: True if path exists and is a file, False otherwise.
+ Any: Value at key if found, provided default or None otherwise.
"""
- return os.path.exists("config.toml") and os.path.isfile("config.toml")
+ nested_key: list[str] = key.split(".")
+ current_item: Any = self._toml_data
- @classmethod
- def get(cls, key: str, default: Any):
- """Get config key, if it exists, or return a default.
+ # Traverse the dict for every depth of the key the user provided.
+ # If any part of the key does not exist, bail and return default.
+ for part in nested_key:
+ if part not in current_item:
+ return default
+ current_item = current_item[part]
- Returns:
- Any: Result, or default if key does not exist.
- """
- return cls.data.get(key, default)
+ # No parts left; `current_item` now holds the requested value.
+ return current_item
diff --git a/starfall/log.py b/starfall/log.py
index 02d4d5c..eab4278 100644
--- a/starfall/log.py
+++ b/starfall/log.py
@@ -5,12 +5,11 @@ from rich.logging import RichHandler
class Log:
- @classmethod
- def setup(cls) -> None:
+ def __init__(self) -> None:
handler = RichHandler(level=logging.DEBUG, rich_tracebacks=True)
logging.basicConfig(
level=logging.DEBUG,
- format="%(message)s",
+ format="%(message)s [%(name)s]",
datefmt="[%Y-%m-%d %H:%M:%S]",
handlers=[handler],
)
diff --git a/starfall/types.py b/starfall/types.py
new file mode 100644
index 0000000..2b4bb57
--- /dev/null
+++ b/starfall/types.py
@@ -0,0 +1,28 @@
+from queue import Queue
+from typing import Any, NamedTuple, final
+
+
+@final
+class QueueMessage(NamedTuple):
+ sender: str
+ receiver: str
+ message: Any
+ variable: str | None = None
+
+
+@final
+class SnapshotQueue(Queue[QueueMessage]):
+ def snapshot(self) -> list[QueueMessage]:
+ with self.mutex:
+ return list(self.queue)
+
+ def get_all_for_receiver(self, receiver: str) -> list[QueueMessage]:
+ res: list[QueueMessage] = []
+ with self.mutex:
+ entries = [entry for entry in self.queue if entry.receiver == receiver]
+ for entry in entries:
+ index = self.queue.index(entry)
+ self.queue.rotate(-index)
+ res.append(self.queue.popleft())
+ self.queue.rotate(index)
+ return res
diff --git a/starfall/web/__init__.py b/starfall/web/__init__.py
index 3621aba..00106bb 100644
--- a/starfall/web/__init__.py
+++ b/starfall/web/__init__.py
@@ -1,36 +1,59 @@
+import importlib
+import logging
import os
-from typing import Any
+from inspect import isclass
+from pathlib import Path
+from pkgutil import iter_modules
+from typing import final
-from flask import Flask
+from flask import Blueprint, Flask
from starfall.config import Config
from starfall.db import db
-from starfall.web.home import home_blueprint
+from starfall.types import SnapshotQueue
+from starfall.web.blueprints.base import BaseBlueprint
-class App:
- app: Flask = Flask(__name__)
- config: dict[str, Any] = dict() # pyright: ignore[reportExplicitAny]
+@final
+class WebUI:
+ def __init__(self):
+ self.config: Config | None = None
+ self.queue: SnapshotQueue | None = None
+ self.app: Flask | None = None
+ self.blueprint: Blueprint = Blueprint("main", __name__)
- @classmethod
- def __init__(cls):
- cls.load_config()
- cls.app.config.update( # pyright: ignore[reportUnknownMemberType]
- SECRET_KEY=cls.config.get("secret_key"),
- SQLALCHEMY_DATABASE_URI=cls.config.get("database_url"),
+ def run(self, config: Config, queue: SnapshotQueue):
+ self.config = config
+ self.queue = queue
+
+ logging.getLogger("web").debug("Hello from %r", type(self))
+
+ self.app = Flask(__name__)
+ self.app.config.update(
+ SECRET_KEY=str(self.config.get("web.secret_key")),
+ SQLALCHEMY_DATABASE_URI=str(self.config.get("web.database_url")),
)
- cls.app.root_path = os.path.realpath(".")
- cls.app.static_folder = os.path.realpath("./web/static")
- cls.app.template_folder = "web"
+ self.app.root_path = os.path.realpath(".")
+ self.app.static_folder = os.path.realpath("./web/static")
+ self.app.template_folder = "web"
- db.init_app(cls.app)
- cls.app.register_blueprint(home_blueprint)
+ db.init_app(self.app)
+ self.import_blueprints()
+ self.app.register_blueprint(self.blueprint, options={"queue": self.queue})
+ self.app.run(host=self.config.get("web.host"), port=self.config.get("web.port"))
- @classmethod
- def load_config(cls):
- cls.config = Config.get("web", {})
+ def import_blueprints(self):
+ path = os.path.realpath(os.path.dirname(__file__) + os.sep + "blueprints")
+ for _, module_name, _ in iter_modules([path], f"{__name__}.blueprints."):
+ if module_name.endswith("base"):
+ continue
- @classmethod
- def run(cls):
- cls.app.run(host=cls.config.get("host"), port=cls.config.get("port"))
+ module = importlib.import_module(module_name)
+
+ for attribute_name in dir(module):
+ attribute = getattr(module, attribute_name)
+
+ if isclass(attribute) and issubclass(attribute, BaseBlueprint):
+ globals()[attribute_name] = attribute
+ globals()[attribute_name](self.blueprint)
diff --git a/starfall/web/blueprints/base.py b/starfall/web/blueprints/base.py
new file mode 100644
index 0000000..843ee0d
--- /dev/null
+++ b/starfall/web/blueprints/base.py
@@ -0,0 +1,39 @@
+import inspect
+import logging
+from abc import ABC
+
+from flask import Blueprint
+from flask.blueprints import BlueprintSetupState
+
+from starfall.types import SnapshotQueue
+
+
+class BaseBlueprint(ABC):
+ def __init__(self, blueprint: Blueprint):
+ blueprint.record(self.on_blueprint_setup)
+
+ if type(self) != BaseBlueprint: # noqa: E721
+ self._debug("Attaching blueprint of type %r to web service", type(self))
+
+ def on_blueprint_setup(self, state: BlueprintSetupState):
+ self.queue: SnapshotQueue = state.options["options"]["queue"]
+
+ if type(self) != BaseBlueprint: # noqa: E721
+ self._info("Blueprint of type %r successfully set up", type(self))
+
+ def _log_access(self):
+ # There is always a previous frame - This function is called when a
+ # route is accessed by a user through a method unknown to this one.
+ self._info("Route access: %r", inspect.currentframe().f_back.f_code.co_name)
+
+ def _debug(self, msg: str, *args: object):
+ logging.getLogger("web").debug(msg, *args)
+
+ def _info(self, msg: str, *args: object):
+ logging.getLogger("web").info(msg, *args)
+
+ def _warn(self, msg: str, *args: object):
+ logging.getLogger("web").warning(msg, *args)
+
+ def _crit(self, msg: str, *args: object):
+ logging.getLogger("web").critical(msg, *args)
diff --git a/starfall/web/blueprints/home.py b/starfall/web/blueprints/home.py
new file mode 100644
index 0000000..0d65144
--- /dev/null
+++ b/starfall/web/blueprints/home.py
@@ -0,0 +1,13 @@
+from flask import Blueprint, render_template
+
+from starfall.web.blueprints.base import BaseBlueprint
+
+
+class HomeBlueprint(BaseBlueprint):
+ def __init__(self, blueprint: Blueprint) -> None:
+ super().__init__(blueprint)
+ blueprint.add_url_rule("/", view_func=self.index)
+
+ def index(self):
+ self._log_access()
+ return render_template("home.html")
diff --git a/starfall/web/home.py b/starfall/web/home.py
deleted file mode 100644
index c3a6ff5..0000000
--- a/starfall/web/home.py
+++ /dev/null
@@ -1,8 +0,0 @@
-from flask import Blueprint, render_template
-
-home_blueprint: Blueprint = Blueprint("main", __name__)
-
-
-@home_blueprint.route("/")
-def index():
- return render_template("home.html")
diff --git a/web/index.html b/web/index.html
index a5e4aa0..72ea173 100644
--- a/web/index.html
+++ b/web/index.html
@@ -8,9 +8,11 @@
-
+
+
+
@@ -22,6 +24,20 @@
{% block beforeMain %}
{% endblock %}
+
+
+
+