trmnl/byos_fastapi
0
0
Fork 0
mirror of https://github.com/usetrmnl/byos_fastapi.git synced 2026-04-29 13:44:09 -07:00
Code Issues Packages Projects Releases Wiki Activity

initial import

Browse Source
This commit is contained in:
Rui Carmo
2025-12-13 10:19:52 +00:00
commit 38cf95d55f
67 changed files with 15438 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/copilot-instructions.md
+22
View File
@@ -0,0 +1,22 @@
# TRMNL Server
This project is an e-ink device server that allows a client to connect to it and generates images and data for display on TRMNL e-ink devices.
## Coding Style Guidelines
When contributing to this project, please adhere to the following coding style guidelines:
* DO NOT CREATE NEW CODE FILES unless explicitly told to.
* SPACES, not TABS, for indentation. Use 4 spaces per indentation level.
* Code in a functional style, with concise functions that do one thing only.
* NEVER duplicate code. Always re-use existing code or create new helper functions. If they are reusable, add them to `utils.py` or `models.py` as appropriate.
* When importing, prefer explicit imports (`from sys import stderr`) rather than just importing the module. A critical example is doing `from os.path import join, dirname, abspath` instead of `import os` and then using `os.path.join()`, etc. Never mind how many imports this creates; explicit imports are preferred for clarity.
* Inside a package, prefer package-relative imports (`from .utils import helper_function`) rather than absolute imports (`from trmnl_server.utils import helper_function`).
* When creating new functions, include type hints for all parameters and return values.
* Do not create one-liner wrappers around existing/internal module functions unless absolutely necessary. Use the public ones instead.
* When considering creating utility functions, try not to create one or two-liners. Inline the logic instead if they are that simple.
* Add utility functions to `utils.py` and constants to `config.py`, making sure to import them where needed and that any major configuration parameters are handled in a consistent way.
* NEVER add import statements inside functions or methods. Add any and all imports at the top of the file.
* Only perform database operations in `models.py`. Create or re-use new helpers there as needed.
* Before writing new helpers or adding inline logic for things that might be reusable, check if there are existing ones that can be re-used or adapted in `utils.py` or `models.py`.
.github/workflows/pylint.yml
+33
View File
@@ -0,0 +1,33 @@
name: Pylint
on: [push]
jobs:
build:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.9", "3.10"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v3
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pylint astroid
pip install -r requirements.txt
- name: Analysing the code with pylint
run: |
pylint_output=$(pylint $(git ls-files '*.py') --output-format=parseable | tee pylint.log)
echo "$pylint_output"
SCORE=$(echo "$pylint_output" | awk -F ' ' '/Your code has been rated at/ {print $7}' | cut -d'/' -f1)
if (( $(echo "$SCORE >= 9.0" | bc -l) )); then
echo "Pylint score ($SCORE) is acceptable."
exit 0
else
echo "Pylint score ($SCORE) is too low."
exit 1
fi
.gitignore
+15
View File
@@ -0,0 +1,15 @@
.DS_Store
trmnl_log.json
db/clientData.txt
logs/server.log
ssl/cert.pem
config.yaml
ssl/key.pem
db/clientLog.txt
trmnl.db
venv/*
__pycache__/*
web/*.bmp
web/*.png
*.pyc
var/
.vscode/settings.json
Vendored
+7
View File
@@ -0,0 +1,7 @@
{
"python.testing.pytestArgs": [
"tests"
],
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true
}
LICENSE
+23
View File
@@ -0,0 +1,23 @@
MIT License
Based on previous work by the original project owner: https://github.com/ohAnd/trmnlServer
Copyright (c) 2025 Rui Carmo
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
Makefile
+18
View File
@@ -0,0 +1,18 @@
PYTHON ?= python3
MODULE ?= trmnl_server
ARGS ?=
export SERVER_PORT ?= 4567
.PHONY: serve test clean
serve:
$(PYTHON) -m $(MODULE) $(ARGS)
test:
$(PYTHON) -m pytest -q tests
clean:
find . -name '__pycache__' -type d -prune -exec rm -rf {} +
find . -name '*.pyc' -delete
rm -rf .pytest_cache .coverage
rm -rf var/generated
rm -rf var/db
OFL.txt
+93
View File
@@ -0,0 +1,93 @@
Copyright 2020 The Space Grotesk Project Authors (https://github.com/floriankarsten/space-grotesk)
This Font Software is licensed under the SIL Open Font License, Version 1.1.
This license is copied below, and is also available with a FAQ at:
http://scripts.sil.org/OFL
-----------------------------------------------------------
SIL OPEN FONT LICENSE Version 1.1 - 26 February 2007
-----------------------------------------------------------
PREAMBLE
The goals of the Open Font License (OFL) are to stimulate worldwide
development of collaborative font projects, to support the font creation
efforts of academic and linguistic communities, and to provide a free and
open framework in which fonts may be shared and improved in partnership
with others.
The OFL allows the licensed fonts to be used, studied, modified and
redistributed freely as long as they are not sold by themselves. The
fonts, including any derivative works, can be bundled, embedded,
redistributed and/or sold with any software provided that any reserved
names are not used by derivative works. The fonts and derivatives,
however, cannot be released under any other type of license. The
requirement for fonts to remain under this license does not apply
to any document created using the fonts or their derivatives.
DEFINITIONS
"Font Software" refers to the set of files released by the Copyright
Holder(s) under this license and clearly marked as such. This may
include source files, build scripts and documentation.
"Reserved Font Name" refers to any names specified as such after the
copyright statement(s).
"Original Version" refers to the collection of Font Software components as
distributed by the Copyright Holder(s).
"Modified Version" refers to any derivative made by adding to, deleting,
or substituting -- in part or in whole -- any of the components of the
Original Version, by changing formats or by porting the Font Software to a
new environment.
"Author" refers to any designer, engineer, programmer, technical
writer or other person who contributed to the Font Software.
PERMISSION & CONDITIONS
Permission is hereby granted, free of charge, to any person obtaining
a copy of the Font Software, to use, study, copy, merge, embed, modify,
redistribute, and sell modified and unmodified copies of the Font
Software, subject to the following conditions:
1) Neither the Font Software nor any of its individual components,
in Original or Modified Versions, may be sold by itself.
2) Original or Modified Versions of the Font Software may be bundled,
redistributed and/or sold with any software, provided that each copy
contains the above copyright notice and this license. These can be
included either as stand-alone text files, human-readable headers or
in the appropriate machine-readable metadata fields within text or
binary files as long as those fields can be easily viewed by the user.
3) No Modified Version of the Font Software may use the Reserved Font
Name(s) unless explicit written permission is granted by the corresponding
Copyright Holder. This restriction only applies to the primary font name as
presented to the users.
4) The name(s) of the Copyright Holder(s) or the Author(s) of the Font
Software shall not be used to promote, endorse or advertise any
Modified Version, except to acknowledge the contribution(s) of the
Copyright Holder(s) and the Author(s) or with their explicit written
permission.
5) The Font Software, modified or unmodified, in part or in whole,
must be distributed entirely under this license, and must not be
distributed under any other license. The requirement for fonts to
remain under this license does not apply to any document created
using the Font Software.
TERMINATION
This license becomes null and void if any of the above conditions are
not met.
DISCLAIMER
THE FONT SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO ANY WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT
OF COPYRIGHT, PATENT, TRADEMARK, OR OTHER RIGHT. IN NO EVENT SHALL THE
COPYRIGHT HOLDER BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
INCLUDING ANY GENERAL, SPECIAL, INDIRECT, INCIDENTAL, OR CONSEQUENTIAL
DAMAGES, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF THE USE OR INABILITY TO USE THE FONT SOFTWARE OR FROM
OTHER DEALINGS IN THE FONT SOFTWARE.
README.md
+97
View File
@@ -0,0 +1,97 @@
# TRMNL Local Server
This is a self-hosted FastAPI backend that emulates the TRMNL cloud so e-paper devices can fetch fresh images and metadata from your local network.
![Obligatory screenshot](docs/playlist.png)
It is loosely based on a Flask implementation by [@ohAnd](https://github.com/ohAnd/trmnlServer), rewritten (nearly) from scratch to use FastAPI, async I/O, and a plugin-driven architecture for rendering various charts and images, prioritizing greyscale output suitable for later firmware versions but allowing you to force 1-bit BMP for legacy devices on a per-item basis
The server maintains device/playlists in SQLite, renders plugin-driven charts/photos into BMP/PNG assets, and exposes `/api/display` plus legacy-compatible endpoints expected by the firmware.
## Non-Goals
- **Full feature parity with the official TRMNL cloud** – this is a lightweight server for personal use, not a 1:1 clone of the official backend.
- **Advanced security features** – while SSL is supported (but disabled by default to save battery), user authentication, multi-user support, and other advanced features are out of scope.
- **Extensive plugin library** – only a few example plugins are provided; users are encouraged to write their own.
- **Web dashboard for management** – a minimal static UI is included for previewing plugin outputs, but no full-featured admin panel.
- **Browser-based rendering** – all image generation is done server-side using Python libraries to minimize system requirements.
## Highlights
- **FastAPI core** – `trmnl_server/main.py` hosts the HTTP API, static assets under `/web`, and middleware-level request logging.
- **Plugin rendering pipeline** – classes in `plugins/` generate images (BMP for the panel, grayscale PNG previews) using Pillow, httpx, pandas, etc.
- **Device + playlist persistence** – SQLAlchemy models in `models.py` keep per-device rotation state, playlists, logs, and battery samples in `var/db/trmnl.db`.
- **Autodiscovered plugin scheduler** – background workers keep assets fresh; see **Plugins & registry** for discovery rules and toggles.
- **Firmware compatibility** – `/api/display` always returns a single `image_url` plus a changing `filename` token so ESP32-based firmware knows when to refresh.
- **Batteries-included tooling** – `Makefile` wraps `make serve` (launch FastAPI via `python -m trmnl_server`) and `make test` (pytest). Plugins can be previewed via helper scripts under the repo root.
- **Color grading + dithering** – "color" grading and multiple dithering algorithms are available to improve image quality on e-ink panels.
## Deployment
I am deploying this with `kata`, a Docker-based service manager I wrote, but any method that can run a FastAPI app will work.
## Running Locally
```bash
git clone https://github.com/rcarmo/trmnlServer.git
cd trmnlServer
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
make serve
```
The server logs which port it binds to (default `SERVER_PORT=4567`). Point your TRMNL device at `http://<server_ip>:<port>`.
Useful commands:
- `make serve` – start FastAPI using the current working directory as the runtime root.
- `SERVER_PORT=8081 make serve` – override port for quick tests.
- `make serve ARGS=/path/to/workdir` – run the server against a different working directory (`var/` contents plus SSL/generated assets) without touching your source tree.
- `python -m trmnl_server --list-plugins` – print the plugin registry (names + defaults) and exit.
- `python -m trmnl_server --run-plugin WeatherPlugin --plugin-output /tmp --plugin-arg image_root=/path` – run a single plugin once for debugging with optional keyword arguments.
- `make test` – run `pytest` (`tests/test_rotation.py`, `tests/test_plugins.py`, `tests/test_weather.py`).
## Configuration
All settings come from environment variables:
- `SERVER_PORT`, `ENABLE_SSL` – networking defaults (4567/False out of the box, set `ENABLE_SSL=true` when you need TLS).
- `IMAGE_PATH`, `REFRESH_TIME`, `DITHERING_MODE` – rendering and dithering behaviour.
- `PHOTO_GRADING_ENABLED` – enable/disable photographic grading for image-heavy plugins (default: true).
- `EINK_TONE_POINTS`, `EINK_TONE_GAMMA` – optional grayscale response compensation points/gamma for panel-space quantization.
- `BATTERY_MAX_VOLTAGE`, `BATTERY_MIN_VOLTAGE`, `TIME_ZONE` – telemetry scaling.
- `SETUP_API_KEY`, `SETUP_FRIENDLY_ID`, `SETUP_MESSAGE` – `/api/setup` payload fields.
- `ASSETS_ROOT`, `STATIC_ROOT`, `GENERATED_ROOT` – relative directories (inside the working dir) for dashboard assets and generated BMP/PNG output (defaults: `web`, `web`, and `var/generated`).
- `CALIBRATION_PLUGIN_ENABLED` – set to `false` to remove calibration plugins from the registry and skip generating calibration assets.
Whenever a setting is changed via the `/settings/*` endpoints, the new value is written to SQLite (table `config_entries`). On startup, `config.py` loads environment variables first (highest precedence) and then applies any persisted entries that are not overridden by the environment, so API-driven tweaks survive restarts without fighting `SERVER_PORT=...` overrides in your shell.
Runtime artefacts now consolidate under `var/` inside your chosen working directory:
- `var/db/trmnl.db` – SQLite database plus future state.
- `var/logs/` – reserved for future log sinks.
- `var/generated/` – plugin BMP/PNG output served via `/generated/*`.
- `var/ssl/` – self-signed certs generated automatically if SSL is enabled.
FastAPI creates these directories during startup if they are missing, and `.gitignore` keeps `var/` out of version control.
## Plugins & registry
- Plugins are auto-discovered from `trmnl_server/plugins/` by the scheduler; any class inheriting `PluginBase` with `AUTO_REGISTER=True` is registered.
- Set `AUTO_REGISTER = False` on a plugin class to opt it out of the registry.
- Set `CALIBRATION_PLUGIN_ENABLED=false` (ENV or `/settings`) to remove all calibration plugins from the registry and skip generating calibration assets.
- `python -m trmnl_server --list-plugins` shows the active registry; `--run-plugin <Name>` respects these toggles.
## API + Static Surface
| Path | Description |
| --------------------------------------------- | ---------------------------------------------------------------------------------------------- |
| `GET /api/display` | Main firmware endpoint: returns `image_url`, `filename`, refresh hints, and playlist metadata. |
| `POST /api/log` | Device log ingestion recorded via SQLAlchemy. |
| `POST /api/battery` | Battery + RSSI samples persisted to `BatteryStatus`. |
| `GET /image/screen.bmp` / `screen1.bmp` | Alternating BMP endpoints to break caches. |
| `GET /image/grayscale.png` / `grayscale1.png` | Optional grayscale preview for firmware that supports it. |
| `GET /web/*` | Static dashboard assets (HTML/JS/CSS/fonts and fallback imagery). |
| `GET /generated/*` | Runtime plugin output (BMP/PNG) served as-is. |
The UI under `web/` shows plugin output previews and rotation metadata; templates in `templates/` are used by specific plugins (e.g., weather renderer).
docs/playlist.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 58 KiB

kata-compose.yaml
+25
View File
@@ -0,0 +1,25 @@
version: "3.9"
services:
trmnl-server:
container_name: trmnl-server
runtime: python
ports:
- "4567:4567"
environment:
# Override as needed (examples):
SERVER_PORT: "4567"
ENABLE_SSL: false
PYTHONIOENCODING: "UTF_8:replace"
PYTHONUNBUFFERED: 1
PYTHONPATH: "/app"
LANG: "en_US.UTF-8"
LC_ALL: "en_US.UTF-8"
TZ: "Europe/Lisbon"
# CALIBRATION_PLUGIN_ENABLED: "true"
# PHOTO_GRADING_ENABLED: "true"
# ASSETS_ROOT: "web"
# STATIC_ROOT: "web"
# GENERATED_ROOT: "var/generated"
command: python3 -u -m trmnl_server
restart: unless-stopped
pytest.ini
+2
View File
@@ -0,0 +1,2 @@
[pytest]
pythonpath = .
requirements.txt
+13
View File
@@ -0,0 +1,13 @@
pillow>=10.3.0
fastapi>=0.109.0
uvicorn>=0.27.0
python-multipart>=0.0.9
psutil>=5.9.4
PyYAML>=6.0
httpx>=0.27.0
sqlalchemy>=2.0.0
d3blocks>=1.0.0
html2image>=2.0.0
pandas>=2.0.0
feedparser>=6.0.10
pytest>=8.2.0
tests/test_plugins.py
+214
View File
@@ -0,0 +1,214 @@
from io import BytesIO
from pathlib import Path
from typing import Any, Dict, List, Tuple
from unittest.mock import AsyncMock, patch
import pytest
from PIL import Image
from trmnl_server.plugins.base import PluginOutput
from trmnl_server.plugins.bing import BingWallpaperPlugin
from trmnl_server.plugins.hn import HNPlugin
from trmnl_server.plugins.random_image import RandomImagePlugin
try:
from trmnl_server.plugins.charts import PageviewsPlugin, VisitorsPlugin # type: ignore[attr-defined]
except ModuleNotFoundError:
PageviewsPlugin = None
VisitorsPlugin = None
from trmnl_server.plugins.xkcd import XKCDPlugin
from trmnl_server.plugins.weather import WeatherPlugin
from trmnl_server.services import plugins as plugin_service
from trmnl_server import config
pytestmark = pytest.mark.asyncio
def _make_image_bytes(size: Tuple[int, int] = (32, 32), color: Tuple[int, int, int] = (255, 0, 0)) -> bytes:
img = Image.new('RGB', size, color=color)
buffer = BytesIO()
img.save(buffer, format='PNG')
return buffer.getvalue()
def _make_image(size: Tuple[int, int] = (32, 32), color: Tuple[int, int, int] = (0, 255, 0)) -> Image.Image:
return Image.open(BytesIO(_make_image_bytes(size=size, color=color))).convert('RGB')
def _plugin_available(plugin_name: str) -> bool:
available = {name.lower() for name in plugin_service.list_available_plugins()}
return plugin_name.lower() in available
def _assert_output_assets(output: PluginOutput) -> None:
mono = Path(output.monochrome_path)
gray = Path(output.grayscale_path)
assert mono.is_file() and mono.stat().st_size > 0
assert gray.is_file() and gray.stat().st_size > 0
async def test_bing_plugin_generates_assets(tmp_path: Path) -> None:
if not _plugin_available(BingWallpaperPlugin.__name__):
pytest.skip('BingWallpaperPlugin not available')
plugin_cls = plugin_service.get_plugin_schedule(BingWallpaperPlugin.__name__).plugin_cls
plugin = plugin_cls()
fake_metadata = {"url": "http://example.com/test_wallpaper.png"}
with patch.object(plugin, '_fetch_metadata', new=AsyncMock(return_value=fake_metadata)), \
patch.object(plugin, '_download_image', new=AsyncMock(return_value=_make_image_bytes())):
output = await plugin.run(output_dir=str(tmp_path))
assert isinstance(output, PluginOutput)
_assert_output_assets(output)
async def test_xkcd_plugin_generates_assets(tmp_path: Path) -> None:
if not _plugin_available(XKCDPlugin.__name__):
pytest.skip('XKCDPlugin not available')
plugin_cls = plugin_service.get_plugin_schedule(XKCDPlugin.__name__).plugin_cls
plugin = plugin_cls()
fake_metadata = {
"img": "http://example.com/comic.png",
"safe_title": "Test XKCD",
"alt": "Alt text",
"transcript": "Transcript text"
}
with patch.object(plugin, '_fetch_metadata', new=AsyncMock(return_value=fake_metadata)), \
patch.object(plugin, '_download_image', new=AsyncMock(return_value=_make_image_bytes())), \
patch.object(plugin, '_load_image', return_value=_make_image()):
output = await plugin.run(output_dir=str(tmp_path))
assert isinstance(output, PluginOutput)
_assert_output_assets(output)
async def test_hn_plugin_generates_assets(tmp_path: Path) -> None:
if not _plugin_available(HNPlugin.__name__):
pytest.skip('HNPlugin not available')
plugin_cls = plugin_service.get_plugin_schedule(HNPlugin.__name__).plugin_cls
plugin = plugin_cls()
fake_entries: List[Dict[str, Any]] = [
{"title": "Test headline", "points": 123, "comments": 45},
{"title": "Another headline", "points": 456, "comments": 78}
]
with patch.object(plugin, '_fetch_entries', new=AsyncMock(return_value=fake_entries)):
output = await plugin.run(output_dir=str(tmp_path))
assert isinstance(output, PluginOutput)
_assert_output_assets(output)
def _fake_stats_data() -> List[Tuple[str, int]]:
return [("00", 0), ("01", 10), ("02", 20), ("03", 5)]
async def test_pageviews_plugin_generates_assets(tmp_path: Path) -> None:
if PageviewsPlugin is None or not _plugin_available('PageviewsPlugin'):
pytest.skip('PageviewsPlugin not available')
plugin_cls = plugin_service.get_plugin_schedule('PageviewsPlugin').plugin_cls
plugin = plugin_cls()
fake_data = _fake_stats_data()
with patch.object(plugin, '_fetch_series', new=AsyncMock(return_value=fake_data)):
output = await plugin.run(output_dir=str(tmp_path))
assert isinstance(output, PluginOutput)
_assert_output_assets(output)
async def test_visitors_plugin_generates_assets(tmp_path: Path) -> None:
if VisitorsPlugin is None or not _plugin_available('VisitorsPlugin'):
pytest.skip('VisitorsPlugin not available')
plugin_cls = plugin_service.get_plugin_schedule('VisitorsPlugin').plugin_cls
plugin = plugin_cls()
fake_data = _fake_stats_data()
with patch.object(plugin, '_fetch_series', new=AsyncMock(return_value=fake_data)):
output = await plugin.run(output_dir=str(tmp_path))
assert isinstance(output, PluginOutput)
_assert_output_assets(output)
async def test_random_image_plugin_generates_assets(tmp_path: Path) -> None:
image_root = tmp_path / "images"
image_root.mkdir()
img_path = image_root / "sample.png"
img = Image.new('RGB', (64, 64), color=(0, 0, 255))
img.save(img_path)
if not _plugin_available(RandomImagePlugin.__name__):
pytest.skip('RandomImagePlugin not available')
plugin_cls = plugin_service.get_plugin_schedule(RandomImagePlugin.__name__).plugin_cls
plugin = plugin_cls()
output = await plugin.run(output_dir=str(tmp_path), image_root=str(image_root))
assert isinstance(output, PluginOutput)
_assert_output_assets(output)
async def test_photo_grading_toggle_changes_prepare_image() -> None:
if not _plugin_available(BingWallpaperPlugin.__name__):
pytest.skip('BingWallpaperPlugin not available')
plugin_cls = plugin_service.get_plugin_schedule(BingWallpaperPlugin.__name__).plugin_cls
plugin = plugin_cls()
gradient = Image.linear_gradient('L').resize((64, 64))
rgb = gradient.convert('RGB')
original = config.PHOTO_GRADING_ENABLED
try:
config.PHOTO_GRADING_ENABLED = False
disabled = plugin.prepare_image(rgb)
assert disabled.mode == 'L'
assert disabled.tobytes() == rgb.convert('L').tobytes()
config.PHOTO_GRADING_ENABLED = True
enabled = plugin.prepare_image(rgb)
assert enabled.mode == 'L'
assert enabled.tobytes() != rgb.convert('L').tobytes()
finally:
config.PHOTO_GRADING_ENABLED = original
async def test_plugin_runner_passes_extra_kwargs(monkeypatch, tmp_path: Path) -> None:
bmp = tmp_path / 'weather.bmp'
png = tmp_path / 'weather.png'
bmp.write_bytes(b'0')
png.write_bytes(b'0')
async def fake_run(self, **kwargs): # type: ignore[override]
assert kwargs['output_dir'] == str(tmp_path)
assert kwargs['extra'] == 'value'
return PluginOutput(monochrome_path=str(bmp), grayscale_path=str(png))
if not _plugin_available(WeatherPlugin.__name__):
pytest.skip('WeatherPlugin not available')
monkeypatch.setattr(WeatherPlugin, 'run', fake_run)
result = await plugin_service.run_single_plugin_by_name(
'WeatherPlugin',
output_dir=str(tmp_path),
plugin_kwargs={'extra': 'value'}
)
assert result.monochrome_path == str(bmp)
assert result.grayscale_path == str(png)
async def test_plugin_runner_errors_on_unknown_plugin(tmp_path: Path) -> None:
with pytest.raises(ValueError):
await plugin_service.run_single_plugin_by_name('NotAPlugin', output_dir=str(tmp_path))
async def test_calibration_plugins_can_be_disabled(monkeypatch) -> None:
original = config.CALIBRATION_PLUGIN_ENABLED
try:
monkeypatch.setattr(config, 'CALIBRATION_PLUGIN_ENABLED', False)
names = plugin_service.list_available_plugins()
assert all('calibration' not in name.lower() for name in names)
with pytest.raises(ValueError):
plugin_service.get_plugin_schedule('CalibrationPlugin')
finally:
monkeypatch.setattr(config, 'CALIBRATION_PLUGIN_ENABLED', original)
tests/test_rotation.py
+444
View File
@@ -0,0 +1,444 @@
from typing import Dict, List
from fastapi.testclient import TestClient
from trmnl_server import main, models
from trmnl_server.services import state
from trmnl_server import utils
from io import BytesIO
from urllib.parse import urlparse
def _prime_rotation_master() -> Dict:
master = state.global_state['rotation_master']
master.clear()
master.update({
'bmp_entries': [b'A', b'B'],
'png_entries': [b'A', b'B'],
'hashes': ['hashA', 'hashB'],
'meta': [
{
'id': 'hashA',
'hash': 'hashA',
'plugin': 'TestPlugin',
'label': 'Entry A',
'url_png': '/web/a.png',
'url_bmp': '/web/a.bmp'
},
{
'id': 'hashB',
'hash': 'hashB',
'plugin': 'TestPlugin',
'label': 'Entry B',
'url_png': '/web/b.png',
'url_bmp': '/web/b.bmp'
}
],
'selected_ids': [],
'version': 0
})
return master
def _prime_rotation_with_dummy_frames() -> Dict:
"""Populate rotation master with a real BMP and PNG so preview conversion succeeds."""
master = state.global_state['rotation_master']
master.clear()
dummy_path = utils.asset_path('img', 'dummy.bmp')
with open(dummy_path, 'rb') as handle:
bmp_bytes = handle.read()
png_bytes = utils.convert_bmp_bytes_to_png(BytesIO(bmp_bytes)).getvalue()
master.update({
'bmp_entries': [bmp_bytes],
'png_entries': [png_bytes],
'hashes': ['hashA'],
'meta': [
{
'id': 'hashA',
'hash': 'hashA',
'plugin': 'TestPlugin',
'label': 'Entry A',
'url_png': '/web/a.png',
'url_bmp': '/web/a.bmp'
}
],
'selected_ids': ['hashA'],
'version': 0
})
return master
def _prime_rotation_with_two_frames() -> Dict:
"""Populate rotation with two distinct frames so we can verify preview updates."""
master = state.global_state['rotation_master']
master.clear()
dummy_path = utils.asset_path('img', 'dummy.bmp')
with open(dummy_path, 'rb') as handle:
bmp_bytes1 = handle.read()
# Create a second BMP by toggling one pixel via PIL to ensure valid encoding
img = utils.load_image(str(dummy_path))
img = utils.ensure_image_mode(img, '1')
img_copy = img.copy()
img_copy.putpixel((0, 0), 0 if img_copy.getpixel((0, 0)) == 255 else 255)
bmp_buffer = BytesIO()
img_copy.save(bmp_buffer, format='BMP')
bmp_bytes2 = bmp_buffer.getvalue()
png_bytes1 = utils.convert_bmp_bytes_to_png(BytesIO(bmp_bytes1)).getvalue()
png_bytes2 = utils.convert_bmp_bytes_to_png(BytesIO(bmp_bytes2)).getvalue()
master.update({
'bmp_entries': [bytes(bmp_bytes1), bytes(bmp_bytes2)],
'png_entries': [png_bytes1, png_bytes2],
'hashes': ['hashA', 'hashB'],
'meta': [
{
'id': 'hashA',
'hash': 'hashA',
'plugin': 'TestPlugin',
'label': 'Entry A',
'url_png': '/web/a.png',
'url_bmp': '/web/a.bmp'
},
{
'id': 'hashB',
'hash': 'hashB',
'plugin': 'TestPlugin',
'label': 'Entry B',
'url_png': '/web/b.png',
'url_bmp': '/web/b.bmp'
}
],
'selected_ids': ['hashA', 'hashB'],
'version': 0
})
return master
def _reset_device_context(*device_ids: str) -> None:
for device_id in device_ids:
for key in ('devices', 'device_playlists', 'client_metrics', 'device_profiles'):
state.global_state.setdefault(key, {}).pop(device_id, None)
models.delete_device_state(device_id)
models.delete_rotation_playlist(device_id)
models.delete_device_playlist_binding(device_id)
def test_save_playlist_updates_rotation_and_order():
client = TestClient(main.app)
master = _prime_rotation_master()
device_id = 'test-device'
_reset_device_context(device_id)
try:
# Initial GET
resp = client.get('/rotation')
assert resp.status_code == 200
data = resp.json()
assert data['entries'][0]['id'] == 'hashA'
assert data['entries'][1]['id'] == 'hashB'
# Post new playlist selecting only hashB
resp = client.post('/rotation', json={'playlist': ['hashB']})
assert resp.status_code == 200
payload = resp.json()
assert payload['playlists']['default'] == ['hashB']
assert master['selected_ids'] == ['hashB']
# Version should bump
assert master['version'] == 1
# Trigger display request so device records the new playlist state
resp_display = client.get('/api/display', headers={'X-Device-Id': device_id})
assert resp_display.status_code == 200
device_state = state.get_device_state(device_id)
with state.STATE_LOCK:
assert device_state['playlist_indexes'] == [1]
assert device_state['playlist_ids'] == ['hashB']
assert device_state['request_count'] == 1
assert device_state['last_entry_hash'] == 'hashB'
finally:
_reset_device_context(device_id)
def test_device_specific_playlist_update():
client = TestClient(main.app)
_prime_rotation_master()
device_id = 'device-specific'
_reset_device_context(device_id)
try:
resp = client.post('/rotation', json={'playlist': ['hashA'], 'device_id': device_id})
assert resp.status_code == 200
payload = resp.json()
device_playlists: Dict[str, List[str]] = payload['playlists'].get('devices') or {}
assert device_playlists.get(device_id) == ['hashA']
assert state.get_playlist_selection(device_id) == ['hashA']
finally:
_reset_device_context(device_id)
def test_devices_endpoint_lists_known_devices():
client = TestClient(main.app)
_prime_rotation_master()
device_id = 'metrics-device'
_reset_device_context(device_id)
try:
state.get_device_state(device_id)
state.update_client_metrics(device_id, refresh_rate=90, battery_voltage=3.8, rssi=-55)
resp = client.get('/devices', params={'include_default': 'false'})
assert resp.status_code == 200
payload = resp.json()
devices = payload['devices']
assert all(entry['device_id'] != state.DEFAULT_DEVICE_ID for entry in devices)
assert any(entry['device_id'] == device_id for entry in devices)
finally:
_reset_device_context(device_id)
def test_device_profile_update_endpoint():
client = TestClient(main.app)
_prime_rotation_master()
device_id = 'kitchen-panel'
_reset_device_context(device_id)
try:
state.get_device_state(device_id)
resp = client.patch(
f'/devices/{device_id}',
json={
'friendly_name': 'Kitchen Display',
'refresh_interval': 180,
'playlist': ['hashB']
}
)
assert resp.status_code == 200
payload = resp.json()
assert payload['device_id'] == device_id
assert payload['friendly_name'] == 'Kitchen Display'
assert payload['refresh_interval'] == 180
assert payload['playlist'] == ['hashB']
profile = state.ensure_device_profile(device_id)
assert profile['friendly_name'] == 'Kitchen Display'
finally:
_reset_device_context(device_id)
def test_image_token_resolves_device_without_headers():
client = TestClient(main.app)
_prime_rotation_master()
device_id = 'token-device'
_reset_device_context(device_id)
try:
# Bind a named playlist so we can verify the resolved frame index.
state.set_named_playlist('Charts', ['hashB'])
state.set_device_playlist_binding(device_id, 'Charts')
resp_display = client.get('/api/display', headers={'X-Device-Id': device_id, 'fw-version': '1.6.9'})
assert resp_display.status_code == 200
image_url = resp_display.json().get('image_url')
assert isinstance(image_url, str) and 'token=' in image_url
token = image_url.split('token=')[-1]
assert token
device_state = state.get_device_state(device_id)
with state.STATE_LOCK:
pending_idx = device_state.get('pending_entry_index')
assert pending_idx == 1
resp_image = client.get(f'/image/grayscale.png?token={token}')
assert resp_image.status_code == 200
assert resp_image.content == state.get_rotation_png_bytes(pending_idx)
finally:
_reset_device_context(device_id)
def test_playlist_entry_forces_bmp_even_when_grayscale_supported():
client = TestClient(main.app)
_prime_rotation_master()
device_id = 'force-bmp-device'
_reset_device_context(device_id)
try:
resp = client.post('/rotation', json={'playlist': ['hashB@bmp'], 'device_id': device_id})
assert resp.status_code == 200
resp_display = client.get('/api/display', headers={'X-Device-Id': device_id, 'fw-version': '1.6.9'})
assert resp_display.status_code == 200
image_url = resp_display.json().get('image_url')
assert isinstance(image_url, str)
assert '/image/screen' in image_url
parsed = urlparse(image_url)
resp_image = client.get(f"{parsed.path}?{parsed.query}")
assert resp_image.status_code == 200
assert resp_image.content == state.get_rotation_bmp_bytes(1)
finally:
_reset_device_context(device_id)
def test_playlist_entry_forces_png_when_grayscale_supported():
client = TestClient(main.app)
_prime_rotation_master()
device_id = 'force-png-device'
_reset_device_context(device_id)
try:
resp = client.post('/rotation', json={'playlist': ['hashB@png'], 'device_id': device_id})
assert resp.status_code == 200
resp_display = client.get('/api/display', headers={'X-Device-Id': device_id, 'fw-version': '1.6.9'})
assert resp_display.status_code == 200
image_url = resp_display.json().get('image_url')
assert isinstance(image_url, str)
assert '/image/grayscale' in image_url
parsed = urlparse(image_url)
resp_image = client.get(f"{parsed.path}?{parsed.query}")
assert resp_image.status_code == 200
assert resp_image.content == state.get_rotation_png_bytes(1)
finally:
_reset_device_context(device_id)
def test_preview_endpoint_persists_last_frame():
client = TestClient(main.app)
_prime_rotation_with_dummy_frames()
device_id = 'preview-device'
_reset_device_context(device_id)
try:
# Trigger a display fetch which should capture the preview
resp = client.get('/api/display', headers={'X-Device-Id': device_id})
assert resp.status_code == 200
device_state = state.get_device_state(device_id)
with state.STATE_LOCK:
preview_index = device_state.get('current_preview_entry_index')
preview_url = device_state.get('current_preview_url')
preview_token = device_state.get('current_preview_token')
assert preview_index is not None, 'preview index should be cached after display fetch'
assert preview_url and preview_url.startswith(f'/preview/{device_id}')
assert preview_token, 'preview token should be set'
# Preview endpoint should return the cached PNG
resp_preview = client.get(f'/preview/{device_id}')
assert resp_preview.status_code == 200
assert resp_preview.headers.get('content-type') == 'image/png'
expected_bytes = state.get_rotation_png_bytes(preview_index)
assert resp_preview.content == expected_bytes
# A different device with no preview should 404
resp_missing = client.get('/preview/unknown-device')
assert resp_missing.status_code == 404
finally:
_reset_device_context(device_id)
def test_named_playlist_binding_affects_rotation_and_deletion_unbinds():
client = TestClient(main.app)
_prime_rotation_master()
device_id = 'bound-device'
_reset_device_context(device_id)
try:
resp = client.post('/playlists', json={'name': 'morning', 'playlist': ['hashA']})
assert resp.status_code == 200
resp = client.patch(f'/devices/{device_id}', json={'playlist_name': 'morning'})
assert resp.status_code == 200
resp_display = client.get('/api/display', headers={'X-Device-Id': device_id})
assert resp_display.status_code == 200
device_state = state.get_device_state(device_id)
with state.STATE_LOCK:
assert device_state['playlist_ids'] == ['hashA']
resp_delete = client.delete('/playlists/morning')
assert resp_delete.status_code == 200
device_state = state.get_device_state(device_id)
with state.STATE_LOCK:
assert device_state.get('request_count') == 0
resp_display = client.get('/api/display', headers={'X-Device-Id': device_id})
assert resp_display.status_code == 200
device_state = state.get_device_state(device_id)
with state.STATE_LOCK:
assert device_state['playlist_ids'] == ['hashA', 'hashB']
finally:
_reset_device_context(device_id)
def test_named_playlist_requires_at_least_one_entry():
client = TestClient(main.app)
_prime_rotation_master()
resp = client.post('/playlists', json={'name': 'empty', 'playlist': []})
assert resp.status_code == 400
def test_preview_updates_on_subsequent_frames():
client = TestClient(main.app)
_prime_rotation_with_two_frames()
device_id = 'preview-advancing'
_reset_device_context(device_id)
try:
# First frame
resp1 = client.get('/api/display', headers={'X-Device-Id': device_id})
assert resp1.status_code == 200
state_after_first = state.get_device_state(device_id)
with state.STATE_LOCK:
first_token = state_after_first.get('current_preview_token')
first_index = state_after_first.get('current_preview_entry_index')
assert first_index is not None
assert first_token
resp_preview_first = client.get(f'/preview/{device_id}')
assert resp_preview_first.status_code == 200
first_bytes = resp_preview_first.content
# Second frame (rotation advances)
resp2 = client.get('/api/display', headers={'X-Device-Id': device_id})
assert resp2.status_code == 200
state_after_second = state.get_device_state(device_id)
with state.STATE_LOCK:
second_token = state_after_second.get('current_preview_token')
second_index = state_after_second.get('current_preview_entry_index')
assert second_index is not None
assert second_token
resp_preview_second = client.get(f'/preview/{device_id}')
assert resp_preview_second.status_code == 200
second_bytes = resp_preview_second.content
assert second_bytes
assert second_bytes != first_bytes, 'Preview bytes should update to the next rotation frame'
assert second_token != first_token, 'Preview token should change when preview bytes change'
finally:
_reset_device_context(device_id)
def test_preview_sequence_stores_each_frame():
client = TestClient(main.app)
master = _prime_rotation_with_two_frames()
device_id = 'preview-sequence'
_reset_device_context(device_id)
try:
expected_sequence = master['png_entries']
for idx in range(3):
resp = client.get('/api/display', headers={'X-Device-Id': device_id})
assert resp.status_code == 200
device_state = state.get_device_state(device_id)
with state.STATE_LOCK:
preview_token = device_state.get('current_preview_token')
preview_index = device_state.get('current_preview_entry_index')
assert preview_index is not None
assert preview_token
expected_bytes = expected_sequence[idx % len(expected_sequence)]
resp_preview = client.get(f'/preview/{device_id}')
assert resp_preview.status_code == 200
assert resp_preview.content == expected_bytes
finally:
_reset_device_context(device_id)
tests/test_weather.py
+44
View File
@@ -0,0 +1,44 @@
from pathlib import Path
from typing import Any, Dict
from unittest.mock import AsyncMock, patch
import pytest
from trmnl_server.plugins.base import PluginOutput
from trmnl_server.plugins.weather import WeatherPlugin
def _sample_weather_payload() -> Dict[str, Any]:
# Minimal structure consumed by WeatherPlugin
times = [f"2025-12-10T{str(h).zfill(2)}:00" for h in range(24)]
temps = [10 + (h % 5) for h in range(24)]
precip = [0.1 * (h % 3) for h in range(24)]
return {
"current_weather": {
"temperature": 21.5,
"windspeed": 12.3
},
"hourly": {
"time": times,
"temperature_2m": temps,
"precipitation": precip
}
}
@pytest.mark.asyncio
async def test_weather_plugin_generates_assets(tmp_path: Path):
payload = _sample_weather_payload()
with patch.object(
WeatherPlugin,
'_fetch_weather_data',
new=AsyncMock(return_value=payload)
):
plugin = WeatherPlugin()
output = await plugin.run(output_dir=str(tmp_path))
assert isinstance(output, PluginOutput)
mono = Path(output.monochrome_path)
gray = Path(output.grayscale_path)
assert mono.is_file() and mono.stat().st_size > 0
assert gray.is_file() and gray.stat().st_size > 0
trmnl_server/__init__.py
+5
View File
@@ -0,0 +1,5 @@
"""TRMNL local server package."""
from . import config # re-export for convenience
__all__ = ['config']
trmnl_server/__main__.py
+6
View File
@@ -0,0 +1,6 @@
"""Module entry point for ``python -m trmnl_server``."""
from .main import run
if __name__ == '__main__':
run()
trmnl_server/config.py
+263
View File
@@ -0,0 +1,263 @@
from __future__ import annotations
import logging
from os import environ, getcwd
from os.path import abspath, isdir, join
from sys import stdout
# Logging Configuration
LOG_LEVEL = environ.get('LOG_LEVEL', 'DEBUG').upper()
logging.basicConfig(
level=getattr(logging, LOG_LEVEL, logging.DEBUG),
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
stream=stdout
)
logger = logging.getLogger('trmnlServer')
# Pillow emits very noisy DEBUG logs (PNG chunk dumps). Keep them at INFO+.
logging.getLogger('PIL').setLevel(logging.INFO)
logging.getLogger('PIL.PngImagePlugin').setLevel(logging.INFO)
logging.getLogger('httpx').setLevel(logging.WARNING)
logging.getLogger('httpcore').setLevel(logging.WARNING)
logger.info('[Config] loading module')
_TRUE_VALUES = {'true', '1', 't', 'yes', 'on'}
IMAGE_PATH = 'images/screen.bmp'
REFRESH_TIME = 900
BATTERY_MAX_VOLTAGE = 4.1
BATTERY_MIN_VOLTAGE = 2.3
TIME_ZONE = 'UTC'
SERVER_PORT = 4567
ENABLE_SSL = False
SERVER_SCHEME = 'http'
SETUP_API_KEY = ''
SETUP_FRIENDLY_ID = 'trmnl-byod'
SETUP_MESSAGE = 'Configured'
DITHERING_MODE = 'none'
ASSETS_ROOT = 'web'
STATIC_ROOT = 'web'
GENERATED_ROOT = 'var/generated'
# Photographic plugin grading
#
# When enabled, photographic plugins apply additional histogram and shadow grading
# prior to tone-curve-aware quantization/dithering.
# When disabled, photographic plugins output raw grayscale (no extra grading),
# which makes it easier to reason about the tone curve/LUT calibration.
PHOTO_GRADING_ENABLED = True
# Calibration plugin control
#
# When disabled, calibration plugins are excluded from the plugin registry and
# no calibration assets are generated.
CALIBRATION_PLUGIN_ENABLED = False
# E-ink grayscale response compensation
#
# These settings allow quantization and dithering to operate in a non-linear
# "panel space" so the resulting grays are closer to perceptually uniform on
# real e-ink panels.
#
# - EINK_TONE_POINTS: optional anchor list "in:out" in 0-255, comma-separated
# e.g. "0:0,32:40,128:160,255:255" (digital input -> observed panel output).
# - EINK_TONE_GAMMA: fallback forward gamma (digital -> panel) when points unset.
EINK_TONE_POINTS = '0:0,32:6,64:18,85:32,128:95,170:155,192:190,224:225,255:250'
EINK_TONE_GAMMA = 1.0
CONFIG_DIR = getcwd()
VAR_ROOT = join(CONFIG_DIR, 'var')
DATABASE_PATH = join(VAR_ROOT, 'db', 'trmnl.db')
LOGS_DIR = join(VAR_ROOT, 'logs')
SSL_DIR = join(VAR_ROOT, 'ssl')
WEB_ROOT_DIR = join(CONFIG_DIR, ASSETS_ROOT)
WEB_STATIC_DIR = join(CONFIG_DIR, STATIC_ROOT)
WEB_GENERATED_DIR = join(CONFIG_DIR, GENERATED_ROOT)
_ENV_OVERRIDES: set[str] = set()
def _env_str(name: str, default: str, config_key: str) -> str:
value = environ.get(name)
if value is None:
return default
_ENV_OVERRIDES.add(config_key)
return value
def _env_bool(name: str, default: bool, config_key: str) -> bool:
value = environ.get(name)
if value is None:
return default
_ENV_OVERRIDES.add(config_key)
return value.strip().lower() in _TRUE_VALUES
def _env_int(name: str, default: int, config_key: str) -> int:
value = environ.get(name)
if value is None:
return default
try:
number = int(value)
_ENV_OVERRIDES.add(config_key)
return number
except ValueError:
logger.warning('[Config] Invalid int for %s: %s', name, value)
return default
def _env_float(name: str, default: float, config_key: str) -> float:
value = environ.get(name)
if value is None:
return default
try:
number = float(value)
_ENV_OVERRIDES.add(config_key)
return number
except ValueError:
logger.warning('[Config] Invalid float for %s: %s', name, value)
return default
def _apply_environment_overrides() -> None:
global IMAGE_PATH, REFRESH_TIME
global BATTERY_MAX_VOLTAGE, BATTERY_MIN_VOLTAGE, TIME_ZONE
global SERVER_PORT, ENABLE_SSL, SERVER_SCHEME
global SETUP_API_KEY, SETUP_FRIENDLY_ID, SETUP_MESSAGE
global DITHERING_MODE, ASSETS_ROOT, STATIC_ROOT, GENERATED_ROOT
global EINK_TONE_POINTS, EINK_TONE_GAMMA
global PHOTO_GRADING_ENABLED, CALIBRATION_PLUGIN_ENABLED
_ENV_OVERRIDES.clear()
default_eink_tone_points = EINK_TONE_POINTS
default_eink_tone_gamma = EINK_TONE_GAMMA
IMAGE_PATH = _env_str('IMAGE_PATH', 'images/screen.bmp', 'image_path')
REFRESH_TIME = _env_int('REFRESH_TIME', 900, 'refresh_time')
BATTERY_MAX_VOLTAGE = _env_float('BATTERY_MAX_VOLTAGE', 4.1, 'battery_max_voltage')
BATTERY_MIN_VOLTAGE = _env_float('BATTERY_MIN_VOLTAGE', 2.3, 'battery_min_voltage')
TIME_ZONE = _env_str('TIME_ZONE', 'UTC', 'time_zone')
SERVER_PORT = _env_int('SERVER_PORT', 4567, 'server_port')
ENABLE_SSL = _env_bool('ENABLE_SSL', False, 'enable_ssl')
SETUP_API_KEY = _env_str('SETUP_API_KEY', '', 'setup_api_key')
SETUP_FRIENDLY_ID = _env_str('SETUP_FRIENDLY_ID', 'trmnl-byod', 'setup_friendly_id')
SETUP_MESSAGE = _env_str('SETUP_MESSAGE', 'Configured', 'setup_message')
DITHERING_MODE = _env_str('DITHERING_MODE', 'none', 'dithering_mode')
ASSETS_ROOT = _env_str('ASSETS_ROOT', 'web', 'assets_root')
STATIC_ROOT = _env_str('STATIC_ROOT', 'web', 'static_root')
GENERATED_ROOT = _env_str('GENERATED_ROOT', 'var/generated', 'generated_root')
EINK_TONE_POINTS = _env_str('EINK_TONE_POINTS', default_eink_tone_points, 'eink_tone_points')
EINK_TONE_GAMMA = _env_float('EINK_TONE_GAMMA', default_eink_tone_gamma, 'eink_tone_gamma')
PHOTO_GRADING_ENABLED = _env_bool('PHOTO_GRADING_ENABLED', PHOTO_GRADING_ENABLED, 'photo_grading_enabled')
CALIBRATION_PLUGIN_ENABLED = _env_bool('CALIBRATION_PLUGIN_ENABLED', CALIBRATION_PLUGIN_ENABLED, 'calibration_plugin_enabled')
_refresh_server_scheme()
def _refresh_server_scheme() -> None:
global SERVER_SCHEME
SERVER_SCHEME = 'https' if ENABLE_SSL else 'http'
def _refresh_path_constants() -> None:
global VAR_ROOT, DATABASE_PATH, LOGS_DIR, SSL_DIR
global WEB_ROOT_DIR, WEB_STATIC_DIR, WEB_GENERATED_DIR
VAR_ROOT = join(CONFIG_DIR, 'var')
DATABASE_PATH = join(VAR_ROOT, 'db', 'trmnl.db')
LOGS_DIR = join(VAR_ROOT, 'logs')
SSL_DIR = join(VAR_ROOT, 'ssl')
WEB_ROOT_DIR = join(CONFIG_DIR, ASSETS_ROOT)
WEB_STATIC_DIR = join(CONFIG_DIR, STATIC_ROOT)
WEB_GENERATED_DIR = join(CONFIG_DIR, GENERATED_ROOT)
def load_config(base_dir: str | None = None) -> None:
"""Apply environment overrides and update path constants for the provided base directory."""
global CONFIG_DIR
_apply_environment_overrides()
if base_dir:
candidate = abspath(base_dir)
if not isdir(candidate):
logger.warning('[Config] Provided base_dir %s is not a directory; using current working directory', base_dir)
candidate = getcwd()
CONFIG_DIR = candidate
else:
CONFIG_DIR = getcwd()
_refresh_path_constants()
def _coerce_bool(value) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in _TRUE_VALUES
return bool(value)
def update_config(key: str, value) -> None:
"""Update an in-memory configuration value."""
global IMAGE_PATH, REFRESH_TIME
global BATTERY_MAX_VOLTAGE, BATTERY_MIN_VOLTAGE, TIME_ZONE
global SERVER_PORT, ENABLE_SSL, SETUP_API_KEY
global SETUP_FRIENDLY_ID, SETUP_MESSAGE, DITHERING_MODE
global ASSETS_ROOT, STATIC_ROOT, GENERATED_ROOT
global EINK_TONE_POINTS, EINK_TONE_GAMMA
global PHOTO_GRADING_ENABLED, CALIBRATION_PLUGIN_ENABLED
logger.info('[Config] Updating %s to %s', key, value)
if key == 'image_path':
IMAGE_PATH = str(value)
elif key == 'refresh_time':
REFRESH_TIME = int(value)
elif key == 'battery_max_voltage':
BATTERY_MAX_VOLTAGE = float(value)
elif key == 'battery_min_voltage':
BATTERY_MIN_VOLTAGE = float(value)
elif key == 'time_zone':
TIME_ZONE = str(value)
elif key == 'server_port':
SERVER_PORT = int(value)
elif key == 'enable_ssl':
ENABLE_SSL = _coerce_bool(value)
_refresh_server_scheme()
elif key == 'setup_api_key':
SETUP_API_KEY = str(value)
elif key == 'setup_friendly_id':
SETUP_FRIENDLY_ID = str(value)
elif key == 'setup_message':
SETUP_MESSAGE = str(value)
elif key == 'dithering_mode':
DITHERING_MODE = str(value)
elif key == 'assets_root':
ASSETS_ROOT = str(value)
_refresh_path_constants()
elif key == 'static_root':
STATIC_ROOT = str(value)
_refresh_path_constants()
elif key == 'generated_root':
GENERATED_ROOT = str(value)
_refresh_path_constants()
elif key == 'eink_tone_points':
EINK_TONE_POINTS = str(value)
elif key == 'eink_tone_gamma':
EINK_TONE_GAMMA = float(value)
elif key == 'photo_grading_enabled':
PHOTO_GRADING_ENABLED = _coerce_bool(value)
elif key == 'calibration_plugin_enabled':
CALIBRATION_PLUGIN_ENABLED = _coerce_bool(value)
else:
logger.warning('[Config] Unknown config key: %s', key)
def apply_persisted_config(entries: dict[str, str]) -> None:
"""Apply database-backed configuration entries unless overridden by env vars."""
for key, raw_value in entries.items():
if key in _ENV_OVERRIDES:
continue
update_config(key, raw_value)
_apply_environment_overrides()
_refresh_path_constants()
trmnl_server/main.py
+341
View File
@@ -0,0 +1,341 @@
#! /usr/bin/env python
"""FastAPI entrypoint and CLI tooling for the TRMNL local server."""
from __future__ import annotations
import argparse
import asyncio
import os
import sys
from contextlib import asynccontextmanager
from typing import Awaitable, Callable, Dict, List, Optional, Tuple
import uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.staticfiles import StaticFiles
from . import config, models, utils
from .routes import api_router, image_router, page_router
from .services import plugins, state
###################################################################################################
logger = config.logger
logger.info('[Main] Starting trmnlServer')
API_LOG_PATH_PREFIXES = ('/api',)
MAX_REQUEST_LOG_BODY = 2048
MAX_RESPONSE_LOG_BODY = 2048
BINARY_CONTENT_PREFIXES = (
'application/octet-stream',
'application/pdf',
'application/zip',
'image/',
'audio/',
'video/'
)
BASE_PATH = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), os.pardir))
STATIC_MOUNT_PATHS: Optional[Tuple[str, str]] = None
def should_log_request(path: str) -> bool:
return any(path.startswith(prefix) for prefix in API_LOG_PATH_PREFIXES)
def format_request_body(body: bytes, limit: int = MAX_REQUEST_LOG_BODY) -> str:
if not body:
return '<empty>'
body_text = body.decode('utf-8', errors='replace')
if len(body_text) > limit:
return f"{body_text[:limit]}...<truncated>"
return body_text
def is_binary_content_type(content_type: str) -> bool:
lowered = (content_type or '').lower()
return any(lowered.startswith(prefix) for prefix in BINARY_CONTENT_PREFIXES)
def format_response_body(body: bytes, limit: int = MAX_RESPONSE_LOG_BODY) -> str:
if not body:
return '<empty>'
text = body.decode('utf-8', errors='replace')
if len(text) > limit:
return f"{text[:limit]}...<truncated>"
return text
@asynccontextmanager
async def lifespan(app: FastAPI): # noqa: ARG001
logger.info('Running initial plugin refresh')
await plugins.refresh_plugin_assets()
logger.info('Starting plugin refresh workers')
await plugins.start_plugin_refreshers()
yield
logger.info('Stopping plugin refresh workers')
await plugins.stop_plugin_refreshers()
app = FastAPI(lifespan=lifespan)
@app.middleware('http')
async def log_api_request(
request: Request,
call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
log_this_request = should_log_request(request.url.path)
if log_this_request:
body_bytes = await request.body()
logger.info(
'[RequestDump] method=%s path=%s query=%s headers=%s body=%s',
request.method,
request.url.path,
dict(request.query_params),
dict(request.headers),
format_request_body(body_bytes)
)
response = await call_next(request)
if log_this_request:
content_type = response.headers.get('content-type', '')
if not is_binary_content_type(content_type):
response_body_chunks = [chunk async for chunk in response.body_iterator]
response_body = b''.join(response_body_chunks)
logger.info(
'[ResponseDump] path=%s status=%s content_type=%s headers=%s body=%s',
request.url.path,
response.status_code,
content_type,
dict(response.headers),
format_response_body(response_body)
)
return Response(
content=response_body,
status_code=response.status_code,
headers=dict(response.headers),
media_type=response.media_type,
background=response.background
)
return response
app.include_router(image_router)
app.include_router(api_router)
app.include_router(page_router)
def _parse_cli_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description='TRMNL local server')
parser.add_argument('workdir', nargs='?', help='Runtime working directory', default=None)
parser.add_argument('--list-plugins', action='store_true', help='List registered plugins and exit')
parser.add_argument('--run-plugin', metavar='PLUGIN', help='Run a single plugin and exit')
parser.add_argument('--plugin-output', metavar='DIR', help='Override output directory when running a plugin')
parser.add_argument(
'--plugin-arg',
action='append',
default=[],
metavar='KEY=VALUE',
help='Additional keyword arguments for --run-plugin'
)
return parser.parse_args(argv)
def _parse_plugin_kwargs(pairs: List[str]) -> Dict[str, str]:
kwargs: Dict[str, str] = {}
for entry in pairs:
if '=' not in entry:
raise ValueError(f"Invalid plugin arg '{entry}'. Expected KEY=VALUE")
key, value = entry.split('=', 1)
kwargs[key] = value
return kwargs
def _resolve_workdir(candidate: Optional[str]) -> str:
if not candidate:
return BASE_PATH
if not os.path.isdir(candidate):
print(f"Path {candidate} is not a directory. Using default path {BASE_PATH}.")
return BASE_PATH
return candidate
def _ensure_static_mounts() -> None:
global STATIC_MOUNT_PATHS
desired = (config.WEB_STATIC_DIR, config.WEB_GENERATED_DIR)
if STATIC_MOUNT_PATHS == desired:
return
# Remove existing mounts if present so new directories are reflected
app.router.routes = [
route for route in app.router.routes
if getattr(route, 'name', None) not in {'web-static', 'web-generated', 'generated-static'}
]
app.mount('/web', StaticFiles(directory=config.WEB_STATIC_DIR), name='web-static')
# Serve generated, volatile plugin outputs directly under /generated
app.mount('/generated', StaticFiles(directory=config.WEB_GENERATED_DIR), name='generated-static')
STATIC_MOUNT_PATHS = desired
def _enforce_runtime_directory_defaults(entries: Optional[Dict[str, str]] = None) -> None:
desired_static = 'web'
desired_assets = 'web'
desired_generated = 'var/generated'
if entries is None:
entries = {}
persisted_static = entries.get('static_root')
if persisted_static and persisted_static != desired_static:
logger.info(
'Migrating static_root from %s to %s to align with reorganized web assets',
persisted_static,
desired_static
)
config.update_config('static_root', desired_static)
models.save_config_entry('static_root', desired_static)
persisted_assets = entries.get('assets_root')
if persisted_assets and persisted_assets != desired_assets:
logger.info(
'Migrating assets_root from %s to %s to align with reorganized web assets',
persisted_assets,
desired_assets
)
config.update_config('assets_root', desired_assets)
models.save_config_entry('assets_root', desired_assets)
persisted_generated = entries.get('generated_root')
if persisted_generated and persisted_generated != desired_generated:
logger.info(
'Migrating generated_root from %s to %s to keep volatile assets under var/',
persisted_generated,
desired_generated
)
config.update_config('generated_root', desired_generated)
models.save_config_entry('generated_root', desired_generated)
def _prepare_runtime(current_dir: str) -> str:
config.load_config(current_dir)
os.makedirs(config.VAR_ROOT, exist_ok=True)
os.makedirs(os.path.dirname(config.DATABASE_PATH), exist_ok=True)
models.init_db()
persisted_entries = models.load_config_entries()
config.apply_persisted_config(persisted_entries)
_enforce_runtime_directory_defaults(persisted_entries)
state.initialize_rotation_playlists_from_storage()
runtime_paths = {
config.VAR_ROOT,
os.path.dirname(config.DATABASE_PATH),
config.LOGS_DIR,
config.SSL_DIR,
config.WEB_STATIC_DIR,
config.WEB_GENERATED_DIR
}
for path in runtime_paths:
os.makedirs(path, exist_ok=True)
server_ip = utils.get_ip_address()
server_scheme = config.SERVER_SCHEME
server_base_url = f"{server_scheme}://{server_ip}:{config.SERVER_PORT}"
state.set_server_base_url(server_base_url)
logger.info(
'Server will be running on IP: %s and port: %s (scheme: %s)',
server_ip,
config.SERVER_PORT,
server_scheme
)
for path in (config.WEB_ROOT_DIR, config.WEB_STATIC_DIR, config.WEB_GENERATED_DIR, config.SSL_DIR):
os.makedirs(path, exist_ok=True)
_ensure_static_mounts()
return server_ip
def _print_available_plugins() -> None:
print('Registered plugins:')
for name in plugins.list_available_plugins():
print(f' - {name}')
def _run_plugin_command(plugin_name: str, output_dir: Optional[str], plugin_kwargs: Dict[str, str]) -> None:
try:
result = asyncio.run(
plugins.run_single_plugin_by_name(
plugin_name,
output_dir=output_dir,
plugin_kwargs=plugin_kwargs
)
)
except ValueError as exc:
logger.error('%s', exc)
sys.exit(2)
except RuntimeError as exc:
logger.error('%s', exc)
sys.exit(1)
logger.info(
'Plugin %s assets saved to %s and %s',
plugin_name,
result.monochrome_path,
result.grayscale_path
)
def _start_http_server(server_ip: str) -> None:
with state.STATE_LOCK:
state.get_device_state(state.DEFAULT_DEVICE_ID)['bmp_send_switch'] = True
if config.ENABLE_SSL:
cert_file = os.path.join(config.SSL_DIR, 'cert.pem')
key_file = os.path.join(config.SSL_DIR, 'key.pem')
if not os.path.exists(cert_file) or not os.path.exists(key_file):
logger.debug('[Main] cert.pem and key.pem not found, generating new ones')
os.system(
f'openssl req -x509 -newkey rsa:4096 -keyout {key_file} -out {cert_file} '
f'-days 365 -nodes '
f'-subj "/C=US/ST=Georgia/L=Atlanta/O=trmnlServer/OU=webapp/CN={server_ip}"'
)
logger.debug('[Main] Starting the server with uvicorn and SSL')
uvicorn.run(
app,
host='0.0.0.0',
port=config.SERVER_PORT,
ssl_keyfile=key_file,
ssl_certfile=cert_file,
log_level='info'
)
else:
logger.debug('[Main] Starting the server without SSL')
uvicorn.run(
app,
host='0.0.0.0',
port=config.SERVER_PORT,
log_level='info'
)
_prepare_runtime(BASE_PATH)
def run() -> None:
args = _parse_cli_args(sys.argv[1:])
current_dir = _resolve_workdir(args.workdir)
server_ip = _prepare_runtime(current_dir)
if args.list_plugins:
_print_available_plugins()
return
if args.run_plugin:
try:
plugin_kwargs = _parse_plugin_kwargs(args.plugin_arg)
except ValueError as exc:
logger.error('%s', exc)
sys.exit(2)
_run_plugin_command(args.run_plugin, args.plugin_output, plugin_kwargs)
return
_start_http_server(server_ip)
if __name__ == '__main__':
run()
trmnl_server/models.py
+519
View File
File diff suppressed because it is too large Load Diff

Some files were not shown because too many files have changed in this diff Show More