diff --git a/README.md b/README.md index b893b03..215894d 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,14 @@ +## Python Version +Python 3.12.3 + ## Docker deployment ### Installation - Clone the repository to your target server host - Install docker if not already present -- Set the slack tokens and DB Web UI Credentials as environment variables using either method below: - - Linux - - `export SLACK_APP_TOKEN=` - - `export SLACK_BOT_TOKEN=` - - `export ME_CONFIG_BASICAUTH_USERNAME=` - - `export ME_CONFIG_BASICAUTH_PASSWORD=` - - .env File +- Set the environment variables (minimum of the slack tokens) using either method below. See [template.env](template.env) for all possible environment variables. + - `docker-compose` File - **Highest Precedence** + - Set the variables in your docker-compose file + - `.env` File - Create a file called `.env` alongside the docker-compose.yml file (see `template.env` in the repo) - Issue one of the following commands: - Local Build: `docker compose up -d --build` @@ -17,8 +17,52 @@ ### Configuration + +#### Network Ports The host ports mapped for the slack server and webserver should be configured in the docker compose file, however it is also possible to override the ports in the server configs directly if you are not using docker. +#### External Config Files +Current files: +- `logging.json` (located at [smib/logging.json](smib/logging.json) in the repo) +- `.env` + +This is mapped to `/app/config` in the container + +> [!IMPORTANT] +> If you map `/app/config` to a host directory, then you *MUST* add the 2 external files to this location. + +You can make this location accessible by Mapping the internal directory to a volume or bind mount in the docker compose file. + +Linux: +```yaml +volumes: + - /etc/smib/:/app/config/ +``` + +Windows: +```yaml +volumes: + - C:/smib/config:/app/config/ +``` + +Local Development: +- Set the `_EXTERNAL_CONFIG_LOCATION` environment variable to the directory containing the External Config Files + +#### Logging +Map the internal /app/logs directory to a volume or bind mount in the docker compose to store the logs outside the containers + +Linux: +```yaml +volumes: + - /var/log/smib/slack/:/app/logs/ +``` + +Windows: +```yaml +volumes: + - C:/smib/slack/logs:/app/logs/ +``` + ## SMIBHID [SMIBHID](smibhid/README.md) is the So Make It Bot Human Interface Device and definitely not a mispronunciation of any insults from a popular 90s documentary detailing the activites of the Jupiter Mining Core. @@ -37,4 +81,8 @@ I think some form of backwards compatibility or similar functionality would be g An [issue](https://github.com/somakeit/S.M.I.B./issues/83) has been created to track the progress and gather ideas. +## Version +When bumping the poetry version (in pyproject.toml), the `HID` class (part of SMIBHID) `version` attribute also needs manually updating. +> [!IMPORTANT] +> This version needs to match the release when it goes into the `master` branch. diff --git a/docker-compose-develop.yml b/docker-compose-develop.yml index 3a2e13f..e9d14dc 100644 --- a/docker-compose-develop.yml +++ b/docker-compose-develop.yml @@ -13,12 +13,10 @@ services: - WEBSOCKET_ALLOWED_HOSTS=smib-webserver,smib-webserver.smib-bridge-network - MONGO_DB_HOST=smib-db - # Passed in from HOST - - SLACK_APP_TOKEN - - SLACK_BOT_TOKEN networks: - smib-bridge-network command: "python -m smib.slack" + restart: unless-stopped smib-webserver: container_name: smib-webserver @@ -32,19 +30,17 @@ services: environment: - WEBSOCKET_HOST=smib-slack - # Passed in from HOST - - SLACK_APP_TOKEN - - SLACK_BOT_TOKEN networks: - smib-bridge-network command: "python -m smib.webserver" + restart: unless-stopped smib-db: # Specific version - latest that works on a pi image: mongo:4.4.18 container_name: smib-db - restart: always + restart: unless-stopped ports: - 27017:27017 networks: @@ -55,12 +51,13 @@ services: container_name: smib-db-ui depends_on: - smib-db - restart: always + restart: unless-stopped ports: - 8082:8081 environment: ME_CONFIG_MONGODB_URL: mongodb://smib-db:27017/ ME_CONFIG_BASICAUTH: true + ME_CONFIG_OPTIONS_READONLY: true networks: - smib-bridge-network diff --git a/docker-compose-master.yml b/docker-compose-master.yml index a65973e..1786f8a 100644 --- a/docker-compose-master.yml +++ b/docker-compose-master.yml @@ -13,12 +13,10 @@ services: - WEBSOCKET_ALLOWED_HOSTS=smib-webserver,smib-webserver.smib-bridge-network - MONGO_DB_HOST=smib-db - # Passed in from HOST - - SLACK_APP_TOKEN - - SLACK_BOT_TOKEN networks: - smib-bridge-network command: "python -m smib.slack" + restart: unless-stopped smib-webserver: container_name: smib-webserver @@ -32,19 +30,17 @@ services: environment: - WEBSOCKET_HOST=smib-slack - # Passed in from HOST - - SLACK_APP_TOKEN - - SLACK_BOT_TOKEN networks: - smib-bridge-network command: "python -m smib.webserver" + restart: unless-stopped smib-db: # Specific version - latest that works on a pi image: mongo:4.4.18 container_name: smib-db - restart: always + restart: unless-stopped ports: - 27017:27017 networks: @@ -55,12 +51,13 @@ services: container_name: smib-db-ui depends_on: - smib-db - restart: always + restart: unless-stopped ports: - 8082:8081 environment: ME_CONFIG_MONGODB_URL: mongodb://smib-db:27017/ ME_CONFIG_BASICAUTH: true + ME_CONFIG_OPTIONS_READONLY: true networks: - smib-bridge-network diff --git a/docker-compose.yml b/docker-compose.yml index 1f9ba4b..191a8fa 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -13,12 +13,10 @@ services: - WEBSOCKET_ALLOWED_HOSTS=smib-webserver,smib-webserver.smib-bridge-network - MONGO_DB_HOST=smib-db - # Passed in from HOST - - SLACK_APP_TOKEN - - SLACK_BOT_TOKEN networks: - smib-bridge-network command: "python -m smib.slack" + restart: unless-stopped smib-webserver: container_name: smib-webserver @@ -35,13 +33,14 @@ services: networks: - smib-bridge-network command: "python -m smib.webserver" + restart: unless-stopped smib-db: # Specific version - latest that works on a pi image: mongo:4.4.18 container_name: smib-db - restart: always + restart: unless-stopped ports: - 27017:27017 networks: @@ -52,7 +51,7 @@ services: container_name: smib-db-ui depends_on: - smib-db - restart: always + restart: unless-stopped ports: - 8082:8081 environment: @@ -60,9 +59,6 @@ services: ME_CONFIG_BASICAUTH: true ME_CONFIG_OPTIONS_READONLY: true - ME_CONFIG_BASICAUTH_USERNAME: - ME_CONFIG_BASICAUTH_PASSWORD: - networks: - smib-bridge-network diff --git a/poetry.lock b/poetry.lock index 13ed674..50cd756 100644 --- a/poetry.lock +++ b/poetry.lock @@ -258,14 +258,14 @@ files = [ [[package]] name = "dataclasses-json" -version = "0.6.5" +version = "0.6.6" description = "Easily serialize dataclasses to and from JSON." category = "main" optional = false python-versions = "<4.0,>=3.7" files = [ - {file = "dataclasses_json-0.6.5-py3-none-any.whl", hash = "sha256:f49c77aa3a85cac5bf5b7f65f4790ca0d2be8ef4d92c75e91ba0103072788a39"}, - {file = "dataclasses_json-0.6.5.tar.gz", hash = "sha256:1c287594d9fcea72dc42d6d3836cf14848c2dc5ce88f65ed61b36b57f515fe26"}, + {file = "dataclasses_json-0.6.6-py3-none-any.whl", hash = "sha256:e54c5c87497741ad454070ba0ed411523d46beb5da102e221efb873801b0ba85"}, + {file = "dataclasses_json-0.6.6.tar.gz", hash = "sha256:0c09827d26fffda27f1be2fed7a7a01a29c5ddcd2eb6393ad5ebf9d77e9deae8"}, ] [package.dependencies] @@ -1318,7 +1318,6 @@ files = [ {file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"}, {file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"}, {file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"}, - {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a08c6f0fe150303c1c6b71ebcd7213c2858041a7e01975da3a99aed1e7a378ef"}, {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"}, {file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"}, {file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"}, diff --git a/pyproject.toml b/pyproject.toml index 2bb8842..787d938 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "smib" -version = "1.0.0" +version = "1.1.0" description = "" authors = ["sam57719 "] readme = "README.md" @@ -17,12 +17,12 @@ websockets = "^12.0" tornado = "^6.4" injectable = "^3.4.7" apscheduler = "^3.10.4" -fastapi = "^0.111.0" uvicorn = "^0.29.0" dataclasses-json = "^0.6.4" aiohttp = "^3.9.5" pymongo = "^4.7.0" mogo = "^0.6.0" +fastapi = "^0.111.0" [build-system] diff --git a/smib-fast.Dockerfile b/smib-fast.Dockerfile index b340bc0..35189c9 100644 --- a/smib-fast.Dockerfile +++ b/smib-fast.Dockerfile @@ -1,5 +1,5 @@ # Use an official Python 3.11 runtime as a base image -FROM python:3.11-buster as builder +FROM python:3.12.3-bullseye as builder RUN pip install poetry==1.4.2 @@ -24,10 +24,10 @@ WORKDIR /app COPY smib ./smib COPY pyproject.toml poetry.lock README.md ./ -RUN poetry install --without dev --no-root && rm -rf $POETRY_CACHE_DIR +RUN poetry install --without dev && rm -rf $POETRY_CACHE_DIR # The runtime image, used to just run the code provided its virtual environment -FROM python:3.11-slim-buster as runtime +FROM python:3.12.3-slim-bullseye as runtime ENV VIRTUAL_ENV=/app/.venv \ PATH="/app/.venv/bin:$PATH" @@ -40,3 +40,12 @@ COPY --from=builder /etc/localtime /etc/localtime WORKDIR /app COPY smib ./smib + +# Remove logging.json from container +RUN rm ./smib/logging.json + +# Copy logging.json into correct container location +COPY smib/logging.json /app/config/logging.json + +# Copy .env if it exists +COPY .env* /app/config/ diff --git a/smib/common/config.py b/smib/common/config.py index 1b54117..5f457e9 100644 --- a/smib/common/config.py +++ b/smib/common/config.py @@ -1,17 +1,19 @@ -import json -import logging.config -from urllib.parse import urlparse -from decouple import config, Csv import warnings -import os from pathlib import Path -import smib +from urllib.parse import urlparse + +from decouple import config, AutoConfig +import smib from smib.common.utils import to_path warnings.filterwarnings("ignore", category=RuntimeWarning) ROOT_DIRECTORY = Path(smib.__file__).parent +EXTERNAL_CONFIG_LOCATION = config('_EXTERNAL_CONFIG_LOCATION', default='/app/config/', cast=Path) + + +config = AutoConfig(search_path=EXTERNAL_CONFIG_LOCATION) APPLICATION_NAME = config('APPLICATION_NAME', default='S.M.I.B.') @@ -19,4 +21,4 @@ WEBSOCKET_HOST = config('WEBSOCKET_HOST', default='localhost') WEBSOCKET_PORT = config('WEBSOCKET_PORT', default=4123, cast=int) WEBSOCKET_PATH = config('WEBSOCKET_PATH', default='ws', cast=to_path) -WEBSOCKET_URL = urlparse(f"{WEBSOCKET_SCHEME}://{WEBSOCKET_HOST}:{WEBSOCKET_PORT}/{WEBSOCKET_PATH}") +WEBSOCKET_URL = urlparse(f"{WEBSOCKET_SCHEME}://{WEBSOCKET_HOST}:{WEBSOCKET_PORT}/{WEBSOCKET_PATH}") \ No newline at end of file diff --git a/smib/common/logging_/setup.py b/smib/common/logging_/setup.py index beba5ae..204a69d 100644 --- a/smib/common/logging_/setup.py +++ b/smib/common/logging_/setup.py @@ -1,20 +1,35 @@ -import inspect import json import logging +import logging.config +from pathlib import Path -from smib.common.config import ROOT_DIRECTORY +from smib.common.config import EXTERNAL_CONFIG_LOCATION from smib.common.utils import get_module_name from injectable import injectable_factory, load_injection_container, inject -def read_logging_json(path=ROOT_DIRECTORY / 'logging.json'): +def read_logging_json(path=EXTERNAL_CONFIG_LOCATION / 'logging.json'): + path = Path(path) + + logging.basicConfig() + logging.info(f'Resolving logging.json to {path}') + + if not (path.exists() and path.is_file()): + logging.warning(f'No logging json file found at {path}') + return None + with open(path, 'rt') as file: config_file = json.load(file) return config_file -def setup_logging(path=ROOT_DIRECTORY / 'logging.json'): - logging.config.dictConfig(read_logging_json(path)) +def setup_logging(path=EXTERNAL_CONFIG_LOCATION / 'logging.json'): + try: + logging.config.dictConfig(read_logging_json(path)) + except Exception as e: + logging.basicConfig() + logger = logging.getLogger('setup_logging') + logger.warning(e) @injectable_factory(logging.Logger, qualifier="plugin_logger") diff --git a/smib/common/utils.py b/smib/common/utils.py index aef441e..53d0acd 100644 --- a/smib/common/utils.py +++ b/smib/common/utils.py @@ -1,3 +1,4 @@ +import importlib import inspect import logging import pickle @@ -8,7 +9,7 @@ from injectable import inject from slack_bolt.response import BoltResponse - +from importlib.metadata import version def is_pickleable(obj): try: @@ -88,3 +89,19 @@ def get_module_file(stack_num: int = 4) -> Path: frame = stack[stack_num] file = inspect.getfile(frame[0]) return Path(file) + + +def get_version() -> str: + from smib.common.config import ROOT_DIRECTORY + + try: + package_name = __package__.split('.')[0] + except AttributeError: + package_name = ROOT_DIRECTORY.parts[-1] + + return version(package_name) + + +if __name__ == '__main__': + print(f"{get_version() = }") + diff --git a/smib/logging.json b/smib/logging.json index f0d18fb..c46312c 100644 --- a/smib/logging.json +++ b/smib/logging.json @@ -19,7 +19,7 @@ "file_handler": { "class": "smib.common.logging_.handlers.EnsureDirectoryTimedRotatingFileHandler", "formatter": "detailed", - "filename": "logs/smib.log", + "filename": "/app/logs/smib.log", "when": "midnight", "interval": 1, "backupCount": 7 @@ -51,6 +51,12 @@ }, "pymongo": { "level": "WARNING" + }, + "uvicorn": { + "level": "WARNING" + }, + "asyncio": { + "level": "WARNING" } } } \ No newline at end of file diff --git a/smib/slack/__main__.py b/smib/slack/__main__.py index 521881b..2f74871 100644 --- a/smib/slack/__main__.py +++ b/smib/slack/__main__.py @@ -38,7 +38,7 @@ def create_slack_bolt_app(): logger.info(f"Created SlackApp: {APPLICATION_NAME}") app.error(handle_errors) app.middleware(inject_logger_to_slack_context) - logger.info(f"Registered SlackApp error handler: {handle_errors}") + logger.debug(f"Registered SlackApp error handler: {handle_errors}") return app diff --git a/smib/slack/config.py b/smib/slack/config.py index b676a93..6860856 100644 --- a/smib/slack/config.py +++ b/smib/slack/config.py @@ -1,3 +1,5 @@ +from decouple import Csv + from smib.common.config import * SLACK_APP_TOKEN = config('SLACK_APP_TOKEN') @@ -15,4 +17,4 @@ PLUGINS_DIRECTORY = config('PLUGINS_DIRECTORY', default=ROOT_DIRECTORY / 'slack' / 'plugins', cast=Path) -SPACE_OPEN_ANNOUNCE_CHANNEL_ID = config('SPACE_OPEN_ANNOUNCE_CHANNEL_ID', default='C06UDPLQRP1') \ No newline at end of file +SPACE_OPEN_ANNOUNCE_CHANNEL_ID = config('SPACE_OPEN_ANNOUNCE_CHANNEL_ID', default='space-open-announce') diff --git a/smib/slack/plugin/loaders/abstract_plugin_loader.py b/smib/slack/plugin/loaders/abstract_plugin_loader.py index f91ee42..c58cc54 100644 --- a/smib/slack/plugin/loaders/abstract_plugin_loader.py +++ b/smib/slack/plugin/loaders/abstract_plugin_loader.py @@ -50,10 +50,16 @@ def load_all(self) -> list[Plugin]: # If the plugin ID already exists, give it a new one if plugin.id in loaded_plugin_ids: + logger.debug(f"Plugin {plugin.id} already exists, giving it new id") plugin.id = f"{plugin.id}_{id(plugin)}" + logger.debug(f"New plugin id: {plugin.id}") plugins.append(plugin) + logger.debug(f"Plugin {plugin.id} loaded. Enabled: {plugin.enabled}") + + logger.info(f"Loaded {len(plugins)} {self.type} plugins") + return plugins def plugin_path_to_id(self, plugin_path: Path) -> str: @@ -76,6 +82,7 @@ def load_plugin(self, plugin_path: Path) -> Plugin: returned_plugin = self.register_plugin(plugin) if not plugin.enabled: + logger.debug(f"Plugin {plugin.id} is not enabled... Unloading...") self.unload_plugin(plugin) if plugin.error: @@ -90,13 +97,18 @@ def unload_plugin(self, plugin: Plugin) -> None: self._remove_middlewares(plugin) def _remove_listeners(self, plugin: Plugin) -> None: + logger = inject('logger') + logger.info(f"Removing listeners for plugin {plugin.id}") listeners = self.app._listeners[::] for listener in listeners: listener_path = inspect.getfile(inspect.unwrap(listener.ack_function)) if Path(listener_path).is_relative_to(plugin.directory): + logger.debug(f"Listener {listener.__name__} from {Path(listener_path).relative_to(plugin.directory).as_posix()} removed") self.app._listeners.remove(listener) def _remove_scheduled_jobs(self, plugin: Plugin) -> None: + logger = inject('logger') + logger.info(f"Removing scheduled jobs for plugin {plugin.id}") listeners = self.app._listeners[::] for listener in listeners: raw_listener_ack = inspect.unwrap(listener.ack_function) @@ -106,6 +118,7 @@ def _remove_scheduled_jobs(self, plugin: Plugin) -> None: continue if job := self._find_job_from_plugin_function(raw_listener_ack): + logger.debug(f"Scheduled job {job.id} from {Path(listener_path).relative_to(plugin.directory).as_posix()} removed") self.scheduler.remove_job(job.id) def _find_job_from_plugin_function(self, plugin_function: callable) -> Job: @@ -115,6 +128,8 @@ def _find_job_from_plugin_function(self, plugin_function: callable) -> Job: )), None) def _remove_middlewares(self, plugin: Plugin) -> None: + logger = inject('logger') + logger.info(f"Removing middlewares for plugin {plugin.id}") middlewares = self.app._middleware_list[::] for middleware in middlewares: func = getattr(middleware, 'func', None) @@ -123,11 +138,12 @@ def _remove_middlewares(self, plugin: Plugin) -> None: middleware_path = inspect.getfile(inspect.unwrap(func)) if Path(middleware_path).is_relative_to(plugin.directory): + logger.debug(f"Middleware {middleware.__name__} from {Path(middleware_path).relative_to(plugin.directory).as_posix()} removed") self.app._middleware_list.remove(middleware) def reload_plugin(self, plugin: Plugin) -> Plugin: logger: logging.Logger = inject("logger") - logger.debug(f"Reloading: {plugin}") + logger.debug(f"Reloading plugin {plugin.id}") self.unload_plugin(plugin) reloaded_plugin = self.load_plugin(plugin.directory) return reloaded_plugin diff --git a/smib/slack/plugin/manager.py b/smib/slack/plugin/manager.py index 6024c3b..780b185 100644 --- a/smib/slack/plugin/manager.py +++ b/smib/slack/plugin/manager.py @@ -2,6 +2,11 @@ from pathlib import Path from typing import Iterator +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from smib.slack.plugin import PluginType + from smib.slack.plugin.loaders.abstract_plugin_loader import AbstractPluginLoader from smib.slack.plugin.plugin import Plugin from injectable import inject, inject_multiple, injectable, autowired, Autowired @@ -29,8 +34,13 @@ def reload_all_plugins(self): self.plugins.append(reloaded_plugin) def disable_plugin(self, plugin: Plugin): - loader = next((x for x in self.plugin_loaders if x.type == plugin.type), None) + logger = inject("logger") + + logger.info(f"Disabling plugin {plugin.id}") + + loader = self._get_plugin_loader_from_type(plugin.type) if not loader: + logger.warning(f"Unable to disable plugin {plugin.id} - no loader found") return open(plugin.directory / '.disable', "wb") @@ -39,22 +49,37 @@ def disable_plugin(self, plugin: Plugin): reloaded_plugin = loader.reload_plugin(plugin) self.plugins.append(reloaded_plugin) + logger.info(f"Plugin {plugin.id} disabled") + def enable_plugin(self, plugin: Plugin): - loader = next((x for x in self.plugin_loaders if x.type == plugin.type), None) + logger = inject("logger") + + logger.info(f"Enabling plugin {plugin.id}") + + loader = self._get_plugin_loader_from_type(plugin.type) if not loader: + logger.warning(f"Unable to enable plugin {plugin.id} - no loader found") return if (disable_file := plugin.directory / '.disable').exists(): os.remove(disable_file) + else: + logger.debug(f"Plugin {plugin.id} already enabled! Reloading anyway...") self.plugins.remove(plugin) reloaded_plugin = loader.reload_plugin(plugin) self.plugins.append(reloaded_plugin) + logger.info(f"Plugin {plugin.id} enabled") + def get_plugin_from_file(self, file: Path) -> Plugin: return next((plugin for plugin in self.plugins if file.is_relative_to(plugin.directory)), None) def __iter__(self) -> Iterator[Plugin]: yield from self.plugins + def _get_plugin_loader_from_type(self, plugin_type: "PluginType") -> "AbstractPluginLoader": + loader = next((x for x in self.plugin_loaders if x.type == plugin_type), None) + return loader + diff --git a/smib/slack/plugins/space/openclose/__init__.py b/smib/slack/plugins/space/openclose/__init__.py index 62a7caa..322d864 100644 --- a/smib/slack/plugins/space/openclose/__init__.py +++ b/smib/slack/plugins/space/openclose/__init__.py @@ -5,6 +5,7 @@ import re from injectable import inject +from slack_sdk.errors import SlackApiError from smib.common.utils import http_bolt_response, is_json_encodable from smib.slack.custom_app import CustomApp @@ -34,7 +35,12 @@ def app_home_opened(client: WebClient, event: dict): def space_open(say, context, ack, client): ack() context['logger'].debug("Space Open!") - say(text='Space Open!', channel=SPACE_OPEN_ANNOUNCE_CHANNEL_ID) + + try: + say(text='Space Open!', channel=SPACE_OPEN_ANNOUNCE_CHANNEL_ID) + except SlackApiError as e: + context['logger'].debug(f"{SPACE_OPEN_ANNOUNCE_CHANNEL_ID = }") + context['logger'].warning(e) space: Space = Space.single() space.set_open() @@ -51,8 +57,11 @@ def space_open(say, context, ack, client): def space_closed(say, context, ack, client): ack() context['logger'].debug("Space Closed!") - - say(text='Space Closed!', channel=SPACE_OPEN_ANNOUNCE_CHANNEL_ID) + try: + say(text='Space Closed!', channel=SPACE_OPEN_ANNOUNCE_CHANNEL_ID) + except SlackApiError as e: + context['logger'].debug(f"{SPACE_OPEN_ANNOUNCE_CHANNEL_ID = }") + context['logger'].warning(e) space: Space = Space.single() space.set_closed() diff --git a/smib/slack/signal_handlers.py b/smib/slack/signal_handlers.py index 426c3a8..4a08a5b 100644 --- a/smib/slack/signal_handlers.py +++ b/smib/slack/signal_handlers.py @@ -6,6 +6,7 @@ def sigterm_handler(signum, frame): logger: logging.Logger = inject("logger") + logger.info("Received shutdown signal") logger.debug(f'Signal handler called with signal, {signum}') raise SystemExit('Exiting...') diff --git a/smib/slack/websocket/server.py b/smib/slack/websocket/server.py index b51b4b4..49ebcc1 100644 --- a/smib/slack/websocket/server.py +++ b/smib/slack/websocket/server.py @@ -29,17 +29,17 @@ def handle(self): slack_app: App event_type: str = bolt_request.body.get('event').get('type') - logger.debug(f"Received event: {event_type}") + logger.info(f"Received http event: {event_type}") bolt_response: BoltResponse = slack_app.dispatch(bolt_request) self.send_message(pickle.dumps(bolt_response)) http_status: HTTPStatus = HTTPStatus(bolt_response.status) - logger.debug(f"Sent status: {bolt_response.status} - {http_status.name}: {http_status.description}") + logger.debug(f"Sent response status: {bolt_response.status} - {http_status.name}: {http_status.description}") def connected(self): logger: logging.Logger = inject("logger") - logger.info(f"{self.address} connected") + logger.info(f"New websocket connection from {self.address}") hostname = None hostname_ip = None @@ -56,10 +56,12 @@ def connected(self): else: logger.debug(f'Address {address} resolved to hostname {hostname}') + logger.info(f"{hostname} connected") + if {address, hostname, hostname_ip}.intersection(set(WEBSOCKET_ALLOWED_HOSTS)): return - logger.warning(f"Connection from {self.address} is {NOT_AUTHORIZED}") + logger.warning(f"Connection from {self.address} is {NOT_AUTHORIZED}. Closing connection.") self.close(reason=NOT_AUTHORIZED) def handle_close(self): diff --git a/smib/webserver/README.md b/smib/webserver/README.md new file mode 100644 index 0000000..92ff8f7 --- /dev/null +++ b/smib/webserver/README.md @@ -0,0 +1,14 @@ +# S.M.I.B. Webserver + +Welcome to the S.M.I.B Webserver Documentation! + +## Endpoints + +- `/smib/event/{event}` + - Methods: + - `PUT` + - `POST` + - `GET` + +### Swagger Docs +Accessible at the `/docs` endpoint diff --git a/smib/webserver/__main__.py b/smib/webserver/__main__.py index 71dd872..5a25b99 100644 --- a/smib/webserver/__main__.py +++ b/smib/webserver/__main__.py @@ -1,4 +1,6 @@ import logging +import re +from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI, Request, APIRouter @@ -12,9 +14,9 @@ from smib.webserver.config import ( WEBSERVER_HOST, WEBSERVER_PORT, WEBSERVER_PATH_PREFIX, WEBSERVER_STATIC_DIRECTORY, WEBSERVER_TEMPLATES_DIRECTORY, - ROOT_DIRECTORY + ROOT_DIRECTORY, APPLICATION_NAME ) -from smib.common.utils import is_pickleable +from smib.common.utils import is_pickleable, get_version from smib.webserver.websocket_handler import WebSocketHandler from smib.common.logging_.setup import setup_logging, read_logging_json @@ -56,29 +58,76 @@ async def generate_bolt_request(fastapi_request: Request): def create_directories(): + logger = inject("logger") + logger.debug(f"Resolved Webserver Template Directory to: {WEBSERVER_TEMPLATES_DIRECTORY}") if not WEBSERVER_TEMPLATES_DIRECTORY.exists(): + logger.info(f"Creating webserver templates directory: {WEBSERVER_TEMPLATES_DIRECTORY}") WEBSERVER_TEMPLATES_DIRECTORY.mkdir() + logger.debug(f"Resolved Webserver Static Directory to: {WEBSERVER_STATIC_DIRECTORY}") if not WEBSERVER_STATIC_DIRECTORY.exists(): + logger.info(f"Creating webserver static directory: {WEBSERVER_STATIC_DIRECTORY}") WEBSERVER_STATIC_DIRECTORY.mkdir() +@asynccontextmanager +async def lifespan(app: FastAPI): + # Load the ML model + logger = inject("logger") + logger.info(f"Webserver started") + + yield + + # Clean up the ML models and release the resources + logger.info(f"Webserver Stopping") + + +def get_readme() -> str: + description = None + description_path = Path(__file__).parent / "README.md" + if description_path.exists() and description_path.is_file(): + with open(description_path) as readme: + description = readme.read() + + return description + + +def get_readme_without_title() -> str: + description = get_readme() + description = re.sub(r"#.*\n", "", description, 1) + return description + + +def get_title() -> str: + title = None + description = get_readme() + + return re.findall(r"#(.*)", description)[0] + + +event_responses = { + 404: {"description": "Not Processed"}, + 418: {"description": "Unhandled Exception"} +} + ws_handler = WebSocketHandler() -app = FastAPI() -router = APIRouter(prefix=WEBSERVER_PATH_PREFIX) + +app = FastAPI(lifespan=lifespan, title=get_title(), version=get_version(), description=get_readme_without_title(), redoc_url=None) +smib_router = APIRouter(prefix=WEBSERVER_PATH_PREFIX) +event_router = APIRouter(prefix='/event', tags=['S.M.I.B. Events'], responses=event_responses) create_directories() -router.mount("/static", StaticFiles(directory=WEBSERVER_STATIC_DIRECTORY), name="static") +app.mount("/static", StaticFiles(directory=WEBSERVER_STATIC_DIRECTORY), name="static") templates = Jinja2Templates(directory=str(WEBSERVER_TEMPLATES_DIRECTORY)) -@router.get('/event/{event}', tags=['SMIB Events']) -@router.post('/event/{event}', tags=['SMIB Events']) -@router.put('/event/{event}', tags=['SMIB Events']) +@event_router.get('/{event}', name="S.M.I.B. GET Event") +@event_router.post('/{event}', name="S.M.I.B. POST Event") +@event_router.put('/{event}', name="S.M.I.B. PUT Event") async def smib_event_handler(request: Request, event: str): logger = inject("logger") - logger.debug(f"Received event {event}") + logger.info(f"Received event {event}") ws_handler.check_and_reconnect_websocket_conn() bolt_request: BoltRequest = await generate_bolt_request(request) logger.debug(f"Request: {request} -> Bolt Request: {bolt_request}") @@ -87,28 +136,29 @@ async def smib_event_handler(request: Request, event: str): return to_starlette_response(bolt_response) -@router.get('/', response_class=HTMLResponse) -async def smib_home(request: Request): - return templates.TemplateResponse("dashboard.html", {"request": request}) - - @app.exception_handler(404) async def custom_404_handler(request, __): return templates.TemplateResponse("404.html", {"request": request}, status_code=404) -app.include_router(router) +smib_router.include_router(event_router) +app.include_router(smib_router) def main(app: FastAPI, ws_handler: WebSocketHandler): logger = inject("logger") try: import uvicorn - logger.info(f"Starting WebServer") - uvicorn.run(app, host=WEBSERVER_HOST, port=WEBSERVER_PORT, log_config=read_logging_json()) + logger.info(f"Starting WebServer v{get_version()}") + uvicorn.run(app, host=WEBSERVER_HOST, port=WEBSERVER_PORT, log_config=read_logging_json(), headers=[("server", APPLICATION_NAME)]) + except KeyboardInterrupt: + ... finally: + logger.info(f"Stopping WebsocketHandler") ws_handler.close_conn() + logger.info(f"Webserver Stopped") + if __name__ == '__main__': main(app, ws_handler) diff --git a/smib/webserver/config.py b/smib/webserver/config.py index 1b5c69e..5bca144 100644 --- a/smib/webserver/config.py +++ b/smib/webserver/config.py @@ -1,3 +1,5 @@ +import os + from smib.common.config import * WEBSERVER_SCHEME = config('WEBSERVER_SCHEME', default='http') diff --git a/smibhid/README.md b/smibhid/README.md index d745282..4392fea 100644 --- a/smibhid/README.md +++ b/smibhid/README.md @@ -14,6 +14,8 @@ Press the space_open or space_closed buttons to call the smib server endpoint ap - Confirms the space state after change by calling space_state - Regularly polls for space state (polling period configurable in config.py) and updates the SMIBHID status appropriately to sync with other space state controls - Flashes both space state LEDs at 2Hz if space state cannot be determined +- 2x16 character LCD display support +- Error information shown on connected displays where configured in modules using ErrorHandler class ## Circuit diagram ### Pico W Connections @@ -25,10 +27,23 @@ Press the space_open or space_closed buttons to call the smib server endpoint ap ### Example breadboard build ![Breadboard photo](images/breadboard.jpg) +### Example prototype build + + +## Hardware +Below is a list of hardware ad links for my specific build: +- [Raspberry Pi Pico W](https://thepihut.com/products/raspberry-pi-pico-w?variant=41952994754755) +- [Prototype board](https://thepihut.com/products/pico-proto-pcb?variant=41359085568195) +- [LED push button switch - Red](https://thepihut.com/products/rugged-metal-pushbutton-with-red-led-ring?variant=27740444561) +- [LED push button switch - Green](https://thepihut.com/products/rugged-metal-pushbutton-with-green-led-ring?variant=27740444625) +- [JST connectors](https://www.amazon.co.uk/dp/B07449V33P) +- [2x16 Character I2C display](https://thepihut.com/products/lcd1602-i2c-module?variant=42422810083523) + ## Deployment Copy the files from the smibhib folder into the root of a Pico W running Micropython and update values in config.py as necessary ### Configuration - Ensure the pins for the space open/closed LEDs and buttons are correctly specified for your wiring +- Configure I2C pins for the display if using, display will detect automatically or disable if not found - Populate Wifi SSID and password - Configure the webserver hostname/IP and port as per your smib.webserver configuration - Set the space state poll frequency in seconds (>= 5), set to 0 to disable the state poll @@ -41,17 +56,44 @@ The LED on the Pico W board is used to give feedback around network connectivity * No LED output: normal operation ## Developers -SMIB uses a class abstracted approach running an async loop using the builtin uasyncio, a static copy of the uaiohttpclient for making async requests and my custom logging module. +SMIB uses a class abstracted approach running an async loop using the built in asyncio library, a static copy of the uaiohttpclient for making async requests and my custom logging module. ### Logging -Set the loglevel argument for the HID object in \_\_main\_\_.py for global log level output where: 0 = Disabled, 1 = Critical, 2 = Error, 3 = Warning, 4 = Info + +#### Log level +Set the LOG_LEVEL value in config.py for global log level output configuration where: 0 = Disabled, 1 = Critical, 2 = Error, 3 = Warning, 4 = Info + +Example: `LOG_LEVEL = 2` + +#### Log Handlers +Populate the LOG_HANDLERS list in config.py with zero or more of the following log output handlers (case sensitive): "Console", "File" + +Example: `LOG_HANDLERS = ["Console", "File"]` + +#### Log file max size +Set the LOG_FILE_MAX_SIZE value in config.py to set the maximum size of the log file in bytes before rotating. The log rotater will create a maximum of 2 files at this size, so configure appropiately for anticpated flash free space. + +Example: `LOG_FILE_MAX_SIZE = 10240` + +### Error handling +Create a new instance of the ErrorHandling class in a module to register a list of possible errors for that module and enabled or disable them for display on connected screens using class methods. See the space state module for an example of implementation. ### Adding functionality Refer to the [S.M.I.B. contribution guidelines](https://github.com/somakeit/S.M.I.B./contribute) for more info on contributing. -Use existing space state buttons, lights, slack APi wrapper and watchers as an example for how to implement: +Use existing space state buttons, lights, slack API wrapper and watchers as an example for how to implement: - Create or use an existing (such as button) appropriate module and class with coroutine to watch for input or other appropriate event - In the HID class - Instantiate the object instance, passing an asyncio event to the watcher and add the watcher coroutine to the loop - Configure another coroutine to watch for the event and take appropriate action on event firing - - Add new API endpoint methods as needed as the API is upgraded to support them \ No newline at end of file + - Add new API endpoint methods as needed as the API is upgraded to support them +- Display drivers can be added by creating a new display driver module + - Ensure the driver registers itself with the driver registry, use LCD1602 as an example + - Import the new driver module in display.py + - Update the config.py file to cinlude the option for your new driver + +## Version +Ensure that the `HID` class version attribute is updated to match the version in `pyproject.toml` + +> [!IMPORTANT] +> This version needs to match the release when it goes into the `master` branch. diff --git a/smibhid/config.py b/smibhid/config.py index 366b8eb..c91ee78 100644 --- a/smibhid/config.py +++ b/smibhid/config.py @@ -1,3 +1,11 @@ +## Logging +# Level 0-4: 0 = Disabled, 1 = Critical, 2 = Error, 3 = Warning, 4 = Info +LOG_LEVEL = 2 +# Handlers: Populate list with zero or more of the following log output handlers (case sensitive): "Console", "File" +LOG_HANDLERS = ["Console", "File"] +# Max log file size in bytes, there will be a maximum of 2 files at this size created +LOG_FILE_MAX_SIZE = 10240 + ## IO SPACE_OPEN_BUTTON = 12 SPACE_CLOSED_BUTTON = 13 @@ -18,4 +26,12 @@ ## Space state # Set the space state poll frequency in seconds (>= 5), set to 0 to disable the state poll -space_state_poll_frequency_s = 5 \ No newline at end of file +space_state_poll_frequency_s = 5 + +## I2C +SDA_PIN = 8 +SCL_PIN = 9 +I2C_ID = 0 + +## Displays - Populate driver list with connected displays from this supported list: ["LCD1602"] +DISPLAY_DRIVERS = ["LCD1602"] \ No newline at end of file diff --git a/smibhid/images/SMIBHID circuit diagram.drawio b/smibhid/images/SMIBHID circuit diagram.drawio deleted file mode 100644 index c048a26..0000000 --- a/smibhid/images/SMIBHID circuit diagram.drawio +++ /dev/null @@ -1,94 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/smibhid/images/SMIBHID circuit diagram.drawio.png b/smibhid/images/SMIBHID circuit diagram.drawio.png index b827bce..784c1b3 100644 Binary files a/smibhid/images/SMIBHID circuit diagram.drawio.png and b/smibhid/images/SMIBHID circuit diagram.drawio.png differ diff --git a/smibhid/images/breadboard.jpg b/smibhid/images/breadboard.jpg index f6024c4..a1fd0bf 100644 Binary files a/smibhid/images/breadboard.jpg and b/smibhid/images/breadboard.jpg differ diff --git a/smibhid/lib/LCD1602.py b/smibhid/lib/LCD1602.py new file mode 100644 index 0000000..576fefc --- /dev/null +++ b/smibhid/lib/LCD1602.py @@ -0,0 +1,175 @@ +## Originally copied from https://files.waveshare.com/upload/d/db/LCD1602_I2C_Module_code.zip + +# -*- coding: utf-8 -*- +from time import sleep +from machine import I2C +from ulogging import uLogger +from display import driver_registry +from config import SDA_PIN, SCL_PIN, I2C_ID +from asyncio import sleep as async_sleep, create_task + +#Device I2C address +LCD_ADDRESS = (0x7c>>1) + +LCD_CLEARDISPLAY = 0x01 +LCD_RETURNHOME = 0x02 +LCD_ENTRYMODESET = 0x04 +LCD_DISPLAYCONTROL = 0x08 +LCD_CURSORSHIFT = 0x10 +LCD_FUNCTIONSET = 0x20 +LCD_SETCGRAMADDR = 0x40 +LCD_SETDDRAMADDR = 0x80 + +#flags for display entry mode +LCD_ENTRYRIGHT = 0x00 +LCD_ENTRYLEFT = 0x02 +LCD_ENTRYSHIFTINCREMENT = 0x01 +LCD_ENTRYSHIFTDECREMENT = 0x00 + +#flags for display on/off control +LCD_DISPLAYON = 0x04 +LCD_DISPLAYOFF = 0x00 +LCD_CURSORON = 0x02 +LCD_CURSOROFF = 0x00 +LCD_BLINKON = 0x01 +LCD_BLINKOFF = 0x00 + +#flags for display/cursor shift +LCD_DISPLAYMOVE = 0x08 +LCD_CURSORMOVE = 0x00 +LCD_MOVERIGHT = 0x04 +LCD_MOVELEFT = 0x00 + +#flags for function set +LCD_8BITMODE = 0x10 +LCD_4BITMODE = 0x00 +LCD_2LINE = 0x08 +LCD_1LINE = 0x00 +LCD_5x8DOTS = 0x00 + +class LCD1602: + """Driver for the LCD1602 16x2 character LED display""" + + def __init__(self) -> None: + """Configure and connect to display via I2C, throw error on connection issue.""" + self.log = uLogger("LCD1602") + self.log.info("Init LCD1602 display driver") + self._row = 16 + self._col = 2 + self.error_loop_task = None + + try: + self.LCD1602_I2C = I2C(I2C_ID, sda = SDA_PIN, scl = SCL_PIN, freq = 400000) + self._showfunction = LCD_4BITMODE | LCD_1LINE | LCD_5x8DOTS + self._begin(self._row) + except BaseException: + self.log.error("Error connecting to LCD display on I2C bus. Check I2C pins and ID and that correct module (I2C address) is connected.") + raise + + def _command(self, cmd: int) -> None: + """Execute a command against the display driver. Refer to command constants.""" + self.LCD1602_I2C.writeto_mem(LCD_ADDRESS, 0x80, chr(cmd)) + + def _write(self, data: int) -> None: + self.LCD1602_I2C.writeto_mem(LCD_ADDRESS, 0x40, chr(data)) + + def setCursor(self, col: int, row: int) -> None: + """Position the cursor ahead of writing a character or string.""" + if(row == 0): + col|=0x80 + else: + col|=0xc0 + self.LCD1602_I2C.writeto(LCD_ADDRESS, bytearray([0x80, col])) + + def clear(self) -> None: + """Clear the entire screen.""" + self._command(LCD_CLEARDISPLAY) + sleep(0.002) + + def print_startup(self, version: str) -> None: + """Render startup information on screen.""" + self.print_on_line(0, "S.M.I.B.H.I.D.") + self.print_on_line(1, f"Loading: v{version}") + + def printout(self, arg: str) -> None: + """Print a string to the cursor position.""" + if(isinstance(arg, int)): + arg=str(arg) + + for x in bytearray(arg, 'utf-8'): + self._write(x) + + def _text_to_line(self, text: str) -> str: + """Internal function to ensure line fits the screen and no previous line text is present for short strings.""" + text = text[:16] + text = "{:<16}".format(text) + return text + + def print_on_line(self, line: int, text: str) -> None: + """Print up to 16 characters on line 0 or 1.""" + self.setCursor(0, line) + self.printout(self._text_to_line(text)) + + def _display(self) -> None: + """Turn on display.""" + self._showcontrol |= LCD_DISPLAYON + self._command(LCD_DISPLAYCONTROL | self._showcontrol) + + def update_status(self, status: dict) -> None: + """Render state and error information on LCD display.""" + self.log.info("Updating display status on LCD1602") + self.errors = status["errors"] + state_line = 0 + self.log.info(f"Length of errors dict: {len(self.errors)}") + if len(self.errors) == 0: + self.log.info("No errors in status update") + self.print_on_line(0, "S.M.I.B.H.I.D.") + state_line = 1 + + self.print_on_line(state_line, f"State: {status["state"]}") + + if self.error_loop_task == None or self.error_loop_task.done(): + self.error_loop_task = create_task(self.async_error_printing_loop()) + + async def async_error_printing_loop(self) -> None: + while True: + for error in self.errors: + self.log.info(f"Printing error: {error}") + self.print_on_line(1, f"Err: {error}") + await async_sleep(2) + await async_sleep(0.1) + + def _begin(self, lines: int) -> None: + """Configure and set initial display output.""" + if (lines > 1): + self._showfunction |= LCD_2LINE + + self._numlines = lines + self._currline = 0 + + sleep(0.05) + + # Send function set command sequence + self._command(LCD_FUNCTIONSET | self._showfunction) + #delayMicroseconds(4500); # wait more than 4.1ms + sleep(0.005) + # second try + self._command(LCD_FUNCTIONSET | self._showfunction) + #delayMicroseconds(150); + sleep(0.005) + # third go + self._command(LCD_FUNCTIONSET | self._showfunction) + # finally, set # lines, font size, etc. + self._command(LCD_FUNCTIONSET | self._showfunction) + # turn the display on with no cursor or blinking default + self._showcontrol = LCD_DISPLAYON | LCD_CURSOROFF | LCD_BLINKOFF + self._display() + # clear it off + self.clear() + # Initialize to default text direction (for romance languages) + self._showmode = LCD_ENTRYLEFT | LCD_ENTRYSHIFTDECREMENT + # set the entry mode + self._command(LCD_ENTRYMODESET | self._showmode) + # backlight init + +driver_registry.register_driver("LCD1602", LCD1602) \ No newline at end of file diff --git a/smibhid/lib/button.py b/smibhid/lib/button.py index 1840495..0eabe27 100644 --- a/smibhid/lib/button.py +++ b/smibhid/lib/button.py @@ -4,9 +4,8 @@ class Button: - def __init__(self, log_level: int, GPIO_pin: int, button_name: str, button_pressed_event: Event) -> None: - self.log_level = log_level - self.logger = uLogger(f"Button {GPIO_pin}", log_level) + def __init__(self, GPIO_pin: int, button_name: str, button_pressed_event: Event) -> None: + self.logger = uLogger(f"Button {GPIO_pin}") self.gpio = GPIO_pin self.pin = Pin(GPIO_pin, Pin.IN, Pin.PULL_UP) self.name = button_name diff --git a/smibhid/lib/display.py b/smibhid/lib/display.py new file mode 100644 index 0000000..924f1c2 --- /dev/null +++ b/smibhid/lib/display.py @@ -0,0 +1,71 @@ +from ulogging import uLogger +from registry import driver_registry +from LCD1602 import LCD1602 +from config import DISPLAY_DRIVERS + +class Display: + """ + Abstracted display capabilities for supported physical displays. + Display drivers must be provided as modules and included in this module to be made available for loading in config.py + All abstracted functions should be defined in this module and will be passed to each configured display (if supported) for the driver to interpret. + + Example: + If an LCD1602 driver is configured to load, then issuing the command Display.print_startup() will render startup information appropriately on the 2x16 display if connected. + """ + def __init__(self) -> None: + self.log = uLogger("Display") + self.drivers = DISPLAY_DRIVERS + self.log.info("Init display") + self.enabled = False + self.screens = [] + self._load_configured_drivers() + self.state = "Unknown" + self.errors = {} + + def _load_configured_drivers(self) -> None: + for driver in self.drivers: + try: + driver_class = driver_registry.get_driver_class(driver) + + if driver_class is None: + raise ValueError(f"Display driver class '{driver}' not registered.") + + self.screens.append(driver_class()) + + except Exception as e: + print(f"An error occurred while confguring display driver '{driver}': {e}") + + if len(self.screens) > 0: + self.log.info(f"Display functionality enabled: {len(self.screens)} screens configured.") + else: + self.log.info("No screens configured successfully; Display functionality disabled.") + self.enabled = False + + def _execute_command(self, command: str, *args) -> None: + self.log.info(f"Executing command on screen drivers: {command}, with arguments: {args}") + for screen in self.screens: + if hasattr(screen, command): + method = getattr(screen, command) + if callable(method): + method(*args) + + def clear(self) -> None: + """Clear all screens.""" + self._execute_command("clear") + + def print_startup(self, version: str) -> None: + """Display startup information on all screens.""" + self._execute_command("print_startup", version) + + def _update_status(self) -> None: + """Update state and error information on all screens.""" + self.log.info("Updating status on all screens") + self._execute_command("update_status", {"state": self.state, "errors": self.errors}) + + def update_state(self, state: str) -> None: + self.state = state + self._update_status() + + def update_errors(self, errors: list) -> None: + self.errors = errors + self._update_status() diff --git a/smibhid/lib/error_handling.py b/smibhid/lib/error_handling.py new file mode 100644 index 0000000..ee348dd --- /dev/null +++ b/smibhid/lib/error_handling.py @@ -0,0 +1,85 @@ +from display import Display +from ulogging import uLogger + +class ErrorHandler: # TODO add pytests for this class + """ + Register a module for error handling and provide methods for registering, enabling, disabling, and getting error messages. + If a display is available, ensure your display handling module registers the display instance with the error handler using configure_display(). + The error handler will then ensure the display update status method is called when errors are enabled or disabled, passing in all enabled errors. + """ + + error_handler_registry = {} + + @classmethod + def register_error_handler(cls, error_handler_name: str, error_handler_instance) -> None: + cls.error_handler_registry[error_handler_name] = error_handler_instance + + @classmethod + def get_error_handler_class(cls, error_handler_name: str) -> None: + return cls.error_handler_registry.get(error_handler_name) + + @classmethod + def configure_display(cls, display: Display) -> None: + cls.display = display + + @classmethod + def update_errors_on_display(cls) -> None: + errors = [] + for error_handler in cls.error_handler_registry: + errors.extend(cls.error_handler_registry[error_handler].get_all_errors()) + cls.display.update_errors(errors) + + def __init__(self, module_name: str) -> None: + """Creates a new error handler instance for a module and registers it with the error handler registry.""" + self.log = uLogger(f"ErrorHandling - {module_name}") + self.errors = {} + self.register_error_handler(module_name, self) + + def register_error(self, key: str, message: str): + """Register a new error with its key, message, and enabled status.""" + if key not in self.errors: + self.errors[key] = {'message': message, 'enabled': False} + self.log.info(f"Registered error '{key}' with message '{message}'") + else: + raise ValueError(f"Error key '{key}' already registered.") + + def enable_error(self, key: str): + """Enable an error.""" + if key in self.errors: + self.errors[key]['enabled'] = True + self.log.info(f"Enabled error '{key}'") + self.update_errors_on_display() + else: + raise ValueError(f"Error key '{key}' not registered.") + + def disable_error(self, key: str): + """Disable an error.""" + if key in self.errors: + self.errors[key]['enabled'] = False + self.log.info(f"Disabled error '{key}'") + self.update_errors_on_display() + else: + raise ValueError(f"Error key '{key}' not registered.") + + def get_error_message(self, key: str) -> str: + """Get the error message for a given key.""" + if key in self.errors: + return self.errors[key]['message'] + else: + raise ValueError(f"Error key '{key}' not registered.") + + def is_error_enabled(self, key: str) -> bool: + """Check if an error is enabled.""" + if key in self.errors: + return self.errors[key]['enabled'] + else: + raise ValueError(f"Error key '{key}' not registered.") + + def get_all_errors(self) -> list: + """Return a list of all enabled errors.""" + errors = [] + for error in self.errors: + if self.errors[error]['enabled']: + errors.append(self.errors[error]['message']) + return errors + \ No newline at end of file diff --git a/smibhid/lib/hid.py b/smibhid/lib/hid.py index 2c2e424..9525d27 100644 --- a/smibhid/lib/hid.py +++ b/smibhid/lib/hid.py @@ -1,201 +1,37 @@ from ulogging import uLogger -import config -from button import Button -from asyncio import Event, create_task, get_event_loop, sleep, wait_for -from utils import StatusLED +from asyncio import get_event_loop, Event, create_task from slack_api import Wrapper -from lib.networking import WirelessNetwork -from constants import OPEN, CLOSED +from display import Display +from space_state import SpaceState +from error_handling import ErrorHandler class HID: - def __init__(self, loglevel: int) -> None: + def __init__(self) -> None: """ Human Interface Device for event spaces providing buttons and status LEDs for space open state. Create HID instance and then run startup() to start services for button monitoring and LED output. """ - self.log = uLogger("HID", loglevel) - self.space_open_button_event = Event() - self.space_closed_button_event = Event() - self.open_button = Button(loglevel, config.SPACE_OPEN_BUTTON, "Space_open", self.space_open_button_event) - self.closed_button = Button(loglevel, config.SPACE_CLOSED_BUTTON, "Space_closed", self.space_closed_button_event) - self.space_open_led = StatusLED(loglevel, config.SPACE_OPEN_LED) - self.space_closed_led = StatusLED(loglevel, config.SPACE_CLOSED_LED) - self.space_open_led.off() - self.space_closed_led.off() - self.wifi = WirelessNetwork(log_level=loglevel) - self.wifi.configure_wifi() - self.slack_api = Wrapper(loglevel, self.wifi) + self.log = uLogger("HID") + self.version = "1.1.0" + self.slack_api = Wrapper() self.loop_running = False - self.space_state = None - self.space_state_check_in_error_state = False - self.checking_space_state = False - self.checking_space_state_timeout_s = 30 + self.display = Display() + self.spaceState = SpaceState(self.display) + self.errorHandler = ErrorHandler("HID") + self.errorHandler.configure_display(self.display) - self.space_state_poll_frequency = config.space_state_poll_frequency_s - if self.space_state_poll_frequency != 0 and self.space_state_poll_frequency < 5: - self.space_state_poll_frequency = 5 - def startup(self) -> None: """ Initialise all aysnc services for the HID. """ - self.log.info("Starting HID") - self.log.info(f"Starting {self.open_button.get_name()} button watcher") - create_task(self.open_button.wait_for_press()) - self.log.info(f"Starting {self.closed_button.get_name()} button watcher") - create_task(self.closed_button.wait_for_press()) - self.log.info(f"Starting {self.open_button.get_name()} button pressed event catcher") - create_task(self.async_space_opened_watcher()) - self.log.info(f"Starting {self.closed_button.get_name()} button pressed event catcher") - create_task(self.async_space_closed_watcher()) - self.log.info("Starting network monitor") - create_task(self.wifi.network_monitor()) - if self.space_state_poll_frequency != 0: - self.log.info(f"Starting space state poller with frequency of {self.space_state_poll_frequency} seconds") - create_task(self.async_space_state_watcher()) - else: - self.log.info("Space state poller disabled by config") + self.log.info("--------Starting SMIBHID--------") + self.log.info(f"SMIBHID firmware version: {self.version}") + self.display.clear() + self.display.print_startup(self.version) + self.spaceState.startup() self.log.info("Entering main loop") self.loop_running = True loop = get_event_loop() - loop.run_forever() - - def set_output_space_open(self) -> None: - """Set LED's display etc to show the space as open""" - self.space_state = True - self.space_open_led.on() - self.space_closed_led.off() - self.log.info("Space state is open.") - - def set_output_space_closed(self) -> None: - """Set LED's display etc to show the space as closed""" - self.space_state = False - self.space_open_led.off() - self.space_closed_led.on() - self.log.info("Space state is closed.") - - def set_output_space_none(self) -> None: - """Set LED's display etc to show the space as none""" - self.space_state = None - self.space_open_led.off() - self.space_closed_led.off() - self.log.info("Space state is none.") - - def _set_space_state_check_to_error(self) -> None: - """Activities relating to space_state check moving to error state""" - self.log.info("Space state check has errored.") - self.space_state_check_in_error_state = True - self.state_check_error_open_led_flash_task = create_task(self.space_open_led.async_constant_flash(2)) - self.state_check_error_closed_led_flash_task = create_task(self.space_closed_led.async_constant_flash(2)) - - def _set_space_state_check_to_ok(self) -> None: - """Activities relating to space_state check moving to ok state""" - self.log.info("Space state check status error has cleared") - self.space_state_check_in_error_state = False - self.state_check_error_open_led_flash_task.cancel() - self.state_check_error_closed_led_flash_task.cancel() - self.space_open_led.off() - self.space_closed_led.off() - - def _free_to_check_space_state(self) -> bool: - """Check that we're not already checking for space state""" - self.log.info("Checking space state check state") - if self.checking_space_state: - self.log.warn("Already checking space state") - return False - else: - self.log.info("Free to check space state") - self.checking_space_state = True - return True - - def _set_space_output(self, new_space_state: bool | None) -> None: - """Call appropriate space output configuration method for new space state.""" - if new_space_state is OPEN: - self.set_output_space_open() - elif new_space_state is CLOSED: - self.set_output_space_closed() - elif new_space_state is None: - self.set_output_space_none() - else: - raise ValueError("Space state is not an expected value") - - async def async_update_space_state_output(self) -> None: - """ - Checks space state from server and sets SMIDHID output to reflect current space state, including errors if space state not available. - """ - self.log.info("Checking space state") - if not self._free_to_check_space_state(): - return - else: - try: - self.log.info("Checking space status from server") - new_space_state = await wait_for(self.slack_api.async_get_space_state(), self.checking_space_state_timeout_s) - self.log.info(f"Space state is: {new_space_state}") - if new_space_state != self.space_state: - self.log.info("Space state changed") - self._set_space_output(new_space_state) - - if self.space_state_check_in_error_state: - self.log.info("Space state unchanged") - self._set_space_state_check_to_ok() - - except Exception as e: - self.log.error(f"Error encountered updating space state: {e}") - if not self.space_state_check_in_error_state: - self._set_space_state_check_to_error() - raise - - finally: - self.log.info("Setting checking_space_state to False") - self.checking_space_state = False - - async def async_space_opened_watcher(self) -> None: - """ - Coroutine to be added to the async loop for watching for the space open button press event and taking appropriate actions. - """ - while True: - await self.space_open_button_event.wait() - self.space_open_button_event.clear() - flash_task = create_task(self.space_open_led.async_constant_flash(4)) - try: - await self.slack_api.async_space_open() - flash_task.cancel() - self.set_output_space_open() - create_task(self.async_update_space_state_output()) - except Exception as e: - self.log.error(f"An exception was encountered trying to set SMIB space state: {e}") - flash_task.cancel() - self.space_open_led.off() - - async def async_space_closed_watcher(self) -> None: - """ - Coroutine to be added to the async loop for watching for the space close button press event and taking appropriate actions. - """ - while True: - await self.space_closed_button_event.wait() - self.space_closed_button_event.clear() - flash_task = create_task(self.space_closed_led.async_constant_flash(4)) - try: - await self.slack_api.async_space_closed() - flash_task.cancel() - self.set_output_space_closed() - create_task(self.async_update_space_state_output()) - except Exception as e: - self.log.error(f"An exception was encountered trying to set SMIB space state: {e}") - flash_task.cancel() - self.space_closed_led.off() - - async def async_space_state_watcher(self) -> None: - """ - Coroutine to frequently poll the space state from the slack server and update SMIBHID output if the state has changed. - """ - while True: - self.log.info("Polling space state") - try: - create_task(self.async_update_space_state_output()) - except Exception as e: - self.log.error(f"State poller encountered an error updating space state: {e}") - finally: - await sleep(self.space_state_poll_frequency) + loop.run_forever() \ No newline at end of file diff --git a/smibhid/lib/networking.py b/smibhid/lib/networking.py index 73dd546..8289f54 100644 --- a/smibhid/lib/networking.py +++ b/smibhid/lib/networking.py @@ -6,13 +6,13 @@ import config from lib.ulogging import uLogger from lib.utils import StatusLED -import uasyncio +from asyncio import sleep, Event class WirelessNetwork: - def __init__(self, log_level: int) -> None: - self.logger = uLogger("WIFI", log_level) - self.status_led = StatusLED(log_level) + def __init__(self) -> None: + self.logger = uLogger("WIFI") + self.status_led = StatusLED() self.wifi_ssid = config.WIFI_SSID self.wifi_password = config.WIFI_PASSWORD self.wifi_country = config.WIFI_COUNTRY @@ -58,7 +58,7 @@ def dump_status(self): async def wait_status(self, expected_status, *, timeout=config.WIFI_CONNECT_TIMEOUT_SECONDS, tick_sleep=0.5) -> bool: for unused in range(ceil(timeout / tick_sleep)): - await uasyncio.sleep(tick_sleep) + await sleep(tick_sleep) status = self.dump_status() if status == expected_status: return True @@ -143,7 +143,7 @@ async def check_network_access(self) -> bool: async def network_monitor(self) -> None: while True: await self.check_network_access() - await uasyncio.sleep(5) + await sleep(5) def get_mac(self) -> str: return self.mac diff --git a/smibhid/lib/registry.py b/smibhid/lib/registry.py new file mode 100644 index 0000000..65e637e --- /dev/null +++ b/smibhid/lib/registry.py @@ -0,0 +1,18 @@ +class DriverRegistry: + """ + Object for driver modules to register, so code can look up a driver class from the driver name. + Driver modules should import driver_registry and call register_driver to map the driver class object to the driver name. + Example: driver_registry.register_driver("LCD1602", LCD1602). + Calling code imports driver_registry and uses get_register_class to obtain the class to create a new driver object. + Example: driver_class = driver_registry.get_driver_class("LCD1602") + """ + def __init__(self) -> None: + self._registry = {} + + def register_driver(self, driver_name: str, driver_class) -> None: + self._registry[driver_name] = driver_class + + def get_driver_class(self, driver_name: str) -> None: + return self._registry.get(driver_name) + +driver_registry = DriverRegistry() \ No newline at end of file diff --git a/smibhid/lib/slack_api.py b/smibhid/lib/slack_api.py index 5a68f61..163b7ec 100644 --- a/smibhid/lib/slack_api.py +++ b/smibhid/lib/slack_api.py @@ -9,9 +9,9 @@ class Wrapper: """ API wrapper for the REST API accepting comands to pass to the local slack server socket. """ - def __init__(self, log_level: int, wifi: WirelessNetwork) -> None: - self.log = uLogger("Slack API", debug_level=log_level) - self.wifi = wifi + def __init__(self) -> None: + self.log = uLogger("Slack API") + self.wifi = WirelessNetwork() self.event_api_base_url = "http://" + WEBSERVER_HOST + ":" + WEBSERVER_PORT + "/smib/event/" async def async_space_open(self) -> None: diff --git a/smibhid/lib/space_state.py b/smibhid/lib/space_state.py new file mode 100644 index 0000000..f9be64b --- /dev/null +++ b/smibhid/lib/space_state.py @@ -0,0 +1,218 @@ +from ulogging import uLogger +import config +from utils import StatusLED +from button import Button +from asyncio import Event, create_task, sleep, wait_for +from constants import OPEN, CLOSED +from display import Display +from slack_api import Wrapper +from error_handling import ErrorHandler + +class SpaceState: + def __init__(self, display: Display) -> None: + """ + Pass an asyncio event object to error_event and use a coroutine to + monitor for event triggers to handle errors in space state checking + by querying the is_in_error_state attribute. + """ + self.log = uLogger("SpaceState") + self.display = display + self.slack_api = Wrapper() + self.space_open_button_event = Event() + self.space_closed_button_event = Event() + self.open_button = Button(config.SPACE_OPEN_BUTTON, "Space_open", self.space_open_button_event) + self.closed_button = Button(config.SPACE_CLOSED_BUTTON, "Space_closed", self.space_closed_button_event) + self.space_open_led = StatusLED(config.SPACE_OPEN_LED) + self.space_closed_led = StatusLED(config.SPACE_CLOSED_LED) + self.space_open_led.off() + self.space_closed_led.off() + self.space_state = None + self.checking_space_state = False + self.checking_space_state_timeout_s = 30 + self.space_state_poll_frequency = config.space_state_poll_frequency_s + if self.space_state_poll_frequency != 0 and self.space_state_poll_frequency < 5: + self.space_state_poll_frequency = 5 + self.configure_error_handling() + + def configure_error_handling(self) -> None: + self.error_handler = ErrorHandler("SpaceState") + self.errors = { + "API": "Slow API", + "CHK": "State check" + # "API": "The space state API is taking too long to respond.", needs scrolling feature on lcd1602 + # "CHK": "An error occurred while checking the space state." + } + + for error_key, error_message in self.errors.items(): + self.error_handler.register_error(error_key, error_message) + + def startup(self) -> None: + self.log.info(f"Starting {self.open_button.get_name()} button watcher") + create_task(self.open_button.wait_for_press()) + self.log.info(f"Starting {self.closed_button.get_name()} button watcher") + create_task(self.closed_button.wait_for_press()) + self.log.info(f"Starting {self.open_button.get_name()} button pressed event catcher") + create_task(self.async_space_opened_watcher()) + self.log.info(f"Starting {self.closed_button.get_name()} button pressed event catcher") + create_task(self.async_space_closed_watcher()) + + if self.space_state_poll_frequency != 0: + self.log.info(f"Starting space state poller with frequency of {self.space_state_poll_frequency} seconds") + create_task(self.async_space_state_watcher()) + else: + self.log.info("Space state poller disabled by config") + + def set_output_space_open(self) -> None: + """Set LED's display etc to show the space as open""" + self.space_state = True + self.space_open_led.on() + self.space_closed_led.off() + self.display.update_state("Open") + self.log.info("Space state is open.") + + def set_output_space_closed(self) -> None: + """Set LED's display etc to show the space as closed""" + self.space_state = False + self.space_open_led.off() + self.space_closed_led.on() + self.display.update_state("Closed") + self.log.info("Space state is closed.") + + def set_output_space_none(self) -> None: + """Set LED's display etc to show the space as none""" + self.space_state = None + self.space_open_led.off() + self.space_closed_led.off() + self.display.update_state("None") + self.log.info("Space state is none.") + + def _set_space_state_check_to_error(self) -> None: + """Activities relating to space_state check moving to error state""" + self.log.info("Space state check has errored.") + if not self.error_handler.is_error_enabled("CHK"): + self.error_handler.enable_error("CHK") + self.state_check_error_open_led_flash_task = create_task(self.space_open_led.async_constant_flash(2)) + self.state_check_error_closed_led_flash_task = create_task(self.space_closed_led.async_constant_flash(2)) + + def _set_space_state_check_to_ok(self) -> None: + """Activities relating to space_state check moving to ok state""" + self.log.info("Space state check status error has cleared") + if self.error_handler.is_error_enabled("CHK"): + self.error_handler.disable_error("CHK") + self.state_check_error_open_led_flash_task.cancel() + self.state_check_error_closed_led_flash_task.cancel() + self.space_open_led.off() + self.space_closed_led.off() + self._set_space_output(self.space_state) + + def _free_to_check_space_state(self) -> bool: + """Check that we're not already checking for space state""" + self.log.info("Checking space state check state") + if self.checking_space_state: + self.log.warn("Already checking space state") + if not self.error_handler.is_error_enabled("API"): + self.error_handler.enable_error("API") + return False + else: + self.log.info("Free to check space state") + self.checking_space_state = True + if self.error_handler.is_error_enabled("API"): + self.error_handler.disable_error("API") + return True + + def _set_space_output(self, new_space_state: bool | None) -> None: + """Call appropriate space output configuration method for new space state.""" + if new_space_state is OPEN: + self.set_output_space_open() + elif new_space_state is CLOSED: + self.set_output_space_closed() + elif new_space_state is None: + self.set_output_space_none() + else: + raise ValueError("Space state is not an expected value") + + async def async_update_space_state_output(self) -> None: + """ + Checks space state from server and sets SMIDHID output to reflect current space state, including errors if space state not available. + """ + self.log.info("Checking space state") + if not self._free_to_check_space_state(): + return + else: + try: + self.log.info("Checking space status from server") + new_space_state = await wait_for(self.slack_api.async_get_space_state(), self.checking_space_state_timeout_s) + self.log.info(f"Space state is: {new_space_state}, was: {self.space_state}") + self._set_space_output(new_space_state) + self._set_space_state_check_to_ok() + + except Exception as e: + self.log.error(f"Error encountered updating space state: {e}") + self._set_space_state_check_to_error() + raise + + finally: + self.log.info("Setting checking_space_state to False") + self.checking_space_state = False + + async def async_space_opened_watcher(self) -> None: + """ + Coroutine to be added to the async loop for watching for the space open button press event and taking appropriate actions. + """ + while True: + await self.space_open_button_event.wait() + self.space_open_button_event.clear() + flash_task = create_task(self.space_open_led.async_constant_flash(4)) + try: + await self.slack_api.async_space_open() + flash_task.cancel() + self.set_output_space_open() + create_task(self.async_update_space_state_output()) + except Exception as e: + self.log.error(f"An exception was encountered trying to set SMIB space state: {e}") + flash_task.cancel() + self.space_open_led.off() + + async def async_space_closed_watcher(self) -> None: + """ + Coroutine to be added to the async loop for watching for the space close button press event and taking appropriate actions. + """ + while True: + await self.space_closed_button_event.wait() + self.space_closed_button_event.clear() + flash_task = create_task(self.space_closed_led.async_constant_flash(4)) + try: + await self.slack_api.async_space_closed() + flash_task.cancel() + self.set_output_space_closed() + create_task(self.async_update_space_state_output()) + except Exception as e: + self.log.error(f"An exception was encountered trying to set SMIB space state: {e}") + flash_task.cancel() + self.space_closed_led.off() + + async def async_space_state_watcher(self) -> None: + """ + Coroutine to frequently poll the space state from the slack server and update SMIBHID output if the state has changed. + """ + + async def task_wrapper_for_error_handling(): + try: + await self.async_update_space_state_output() + except Exception as e: + self.log.error(f"State poller task encountered an error updating space state: {e}") + + while True: + self.log.info("Polling space state") + try: + create_task(task_wrapper_for_error_handling()) + except Exception as e: + self.log.error(f"State poller encountered an error creating task: {e}") + finally: + await sleep(self.space_state_poll_frequency) + + def set_error_id(self, error_id: int) -> None: + self.error_id = error_id + + def get_error_id(self) -> int: + return self.error_id \ No newline at end of file diff --git a/smibhid/lib/ulogging.py b/smibhid/lib/ulogging.py index fc2cd1d..9e0a16d 100644 --- a/smibhid/lib/ulogging.py +++ b/smibhid/lib/ulogging.py @@ -1,29 +1,118 @@ -import gc +from gc import mem_free +from os import stat, remove, rename class uLogger: - def __init__(self, module_name: str, debug_level: int) -> None: + def __init__(self, module_name: str, log_level: int = 0, handlers: list = []) -> None: """ - Init with module name to log and session debug level + Init with module name to log and session debug level, that defaults to 2 and can be overidden globally using log_level=x in config.py Raise a debug message using the appropriate function for the severity Debug level 0-3: Each level adds more verbosity 0 = Disabled, 1 = Critical, 2 = Error, 3 = Warning, 4 = Info """ self.module_name = module_name - self.debug_level = debug_level + self.configure_log_level(log_level) + self.configure_handlers(handlers) - def info(self, message) -> None: - if self.debug_level > 3: - print(f"[Mem: {round(gc.mem_free() / 1024)}kB free][Info][{self.module_name}]: {message}") + def configure_log_level(self, log_level: int) -> None: + self.log_level = 0 + + if log_level > 0: + self.log_level = log_level + else: + try: + from config import LOG_LEVEL as config_log_level + self.log_level = config_log_level + except ImportError: + print("LOG_LEVEL not found in config.py not found. Using default log level.") + except Exception as e: + print(f"An unexpected error occurred: {e}. Using default log level.") - def warn(self, message) -> None: - if self.debug_level > 2: - print(f"[Mem: {round(gc.mem_free() / 1024)}kB free][Warning][{self.module_name}]: {message}") + def configure_handlers(self, handlers: list) -> None: + self.handlers = [] + self.handler_objects = [] + + if len(handlers) > 0: + self.handlers = handlers + else: + try: + from config import LOG_HANDLERS as config_log_handlers + self.handlers = config_log_handlers + except ImportError: + print("LOG_HANDLERS not found in config.py not found. Using default output handler.") + except Exception as e: + print(f"An unexpected error occurred: {e}. Using default output handler.") + + for handler in self.handlers: + try: + handler_class = globals().get(handler) + + if handler_class is None: + raise ValueError(f"Handler class '{handler}' not found.") + + handler = handler_class() + self.handler_objects.append(handler) + except Exception as e: + print(f"An error occurred while confguring handler '{handler}': {e}") + raise - def error(self, message) -> None: - if self.debug_level > 1: - print(f"[Mem: {round(gc.mem_free() / 1024)}kB free][*Error*][{self.module_name}]: {message}") + def decorate_message(self, message: str, level: str) -> str: + decorated_message = f"[Mem: {round(mem_free() / 1024)}kB free][{level}][{self.module_name}]: {message}" + return decorated_message + + def process_handlers(self, message: str) -> None: + for handler in self.handler_objects: + try: + handler.emit(message) + except Exception as e: + print(f"An error occurred while processing handler '{handler}': {e}") + raise + + def info(self, message: str) -> None: + if self.log_level > 3: + self.process_handlers(self.decorate_message(message, "Info")) + + def warn(self, message: str) -> None: + if self.log_level > 2: + self.process_handlers(self.decorate_message(message, "Warning")) + + def error(self, message: str) -> None: + if self.log_level > 1: + self.process_handlers(self.decorate_message(message, "Error")) + + def critical(self, message: str) -> None: + if self.log_level > 0: + self.process_handlers(self.decorate_message(message, "Critical")) + +class Console: + def __init__(self) -> None: + pass + + def emit(self, message) -> None: + print(message) + +class File: + def __init__(self) -> None: + self.log_file = "log.txt" + self.second_log_file = "log2.txt" + from config import LOG_FILE_MAX_SIZE + self.LOG_FILE_MAX_SIZE = LOG_FILE_MAX_SIZE + + def emit(self, message) -> None: + with open(self.log_file, "a") as log_file: + log_file.write(message + "\n") + self.check_for_rotate() + + def check_for_rotate(self) -> None: + log_file_size = stat(self.log_file)[6] + if log_file_size > self.LOG_FILE_MAX_SIZE: + self.rotate_file() - def critical(self, message) -> None: - if self.debug_level > 0: - print(f"[Mem: {round(gc.mem_free() / 1024)}kB free][!Critical!][{self.module_name}]: {message}") \ No newline at end of file + def rotate_file(self) -> None: + try: + remove(self.second_log_file) + except OSError: + print(f"{self.second_log_file} did not exist to be deleted.") + + rename(self.log_file, self.second_log_file) + \ No newline at end of file diff --git a/smibhid/lib/utils.py b/smibhid/lib/utils.py index 17cef49..cc51f3a 100644 --- a/smibhid/lib/utils.py +++ b/smibhid/lib/utils.py @@ -2,6 +2,7 @@ from machine import Pin import uasyncio from lib.ulogging import uLogger +from os import stat, statvfs class StatusLED: """ @@ -9,8 +10,8 @@ class StatusLED: Info log level output of state changes. Supports sync and async flash functions taking count and frequency arguments. """ - def __init__(self, log_level: int, gpio_pin: int = -1) -> None: - self.logger = uLogger("Status_LED", log_level) + def __init__(self, gpio_pin: int = -1) -> None: + self.logger = uLogger("Status_LED") if gpio_pin > -1: self.status_led = Pin(gpio_pin, Pin.OUT) self.pin_id = gpio_pin @@ -59,4 +60,4 @@ def flash(self, count: int, hz: float) -> None: sleep(sleep_duration) self.on() sleep(sleep_duration) - self.off() \ No newline at end of file + self.off() diff --git a/smibhid/main.py b/smibhid/main.py index b98f4b3..c2f958c 100644 --- a/smibhid/main.py +++ b/smibhid/main.py @@ -2,5 +2,5 @@ from lib.hid import HID -hid = HID(loglevel=2) +hid = HID() hid.startup() \ No newline at end of file diff --git a/template.env b/template.env index 5dda0e5..9aeb43c 100644 --- a/template.env +++ b/template.env @@ -1,7 +1,36 @@ +### Common Configuration Values +#APPLICATION_NAME= + +# Websocket Config +#WEBSOCKET_SCHEME= +#WEBSOCKET_HOST= +#WEBSOCKET_PORT= +#WEBSOCKET_PATH= + +### Slack APP Configuration Values + # Slack Credentials SLACK_BOT_TOKEN= SLACK_APP_TOKEN= -# Mongo DB UI (Mongo Express) Auth -ME_CONFIG_BASICAUTH_USERNAME= -ME_CONFIG_BASICAUTH_PASSWORD= \ No newline at end of file +#WEBSOCKET_ALLOWED_HOSTS= + +# MONGO DB Credentials +#MONGO_DB_HOST= +#MONGO_DB_PORT= +#MONGO_DB_DEFAULT_DB= +#MONGO_DB_CONNECT_TIMEOUT_SECONDS= + +# Plugins Config +#PLUGINS_DIRECTORY= +#SPACE_OPEN_ANNOUNCE_CHANNEL_ID= + +## Webserver Config +#WEBSERVER_SCHEME= +#WEBSERVER_HOST= +#WEBSERVER_PORT= +#WEBSERVER_PATH= +#WEBSERVER_SECRET_KEY= +#WEBSERVER_PATH_PREFIX= +#WEBSERVER_TEMPLATES_DIRECTORY= +#WEBSERVER_STATIC_DIRECTORY= \ No newline at end of file