From 9df2a952092633ee41fc1e810df73def137d1542 Mon Sep 17 00:00:00 2001 From: Sassan Haradji Date: Tue, 20 Feb 2024 02:09:27 +0400 Subject: [PATCH] refactor: `HeadlessWidget` is not a singleton anymore, config module is now source of truth for config refactor: initializing display and config is now done in `setup_headless_kivy` feat: class method `get_instance` to get the closest parent instance of type `HeadlessWidget` --- .gitignore | 106 +------------- CHANGELOG.md | 7 + README.md | 4 +- headless_kivy_pi/__init__.py | 245 +++++++++---------------------- headless_kivy_pi/config.py | 267 ++++++++++++++++++++++++++++++++++ headless_kivy_pi/constants.py | 9 +- headless_kivy_pi/display.py | 17 +-- headless_kivy_pi/setup.py | 120 --------------- pyproject.toml | 8 +- 9 files changed, 371 insertions(+), 412 deletions(-) create mode 100644 headless_kivy_pi/config.py delete mode 100644 headless_kivy_pi/setup.py diff --git a/.gitignore b/.gitignore index 120fc87..0f9f385 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,4 @@ -# Gitignore for the following technologies: Python +.DS_Store # Byte-compiled / optimized / DLL files __pycache__/ @@ -57,69 +57,8 @@ cover/ *.mo *.pot -# Django stuff: -*.log -local_settings.py -db.sqlite3 -db.sqlite3-journal - -# Flask stuff: -instance/ -.webassets-cache - -# Scrapy stuff: -.scrapy - -# Sphinx documentation -docs/_build/ - -# PyBuilder -.pybuilder/ -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# IPython -profile_default/ -ipython_config.py - # pyenv -# For a library or package, you might want to ignore these files since the code is -# intended to run in multiple environments; otherwise, check them in: -# .python-version - -# pipenv -# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. -# However, in case of collaboration, if having platform-specific dependencies or dependencies -# having no cross-platform support, pipenv may install dependencies that don't work, or not -# install all needed dependencies. -#Pipfile.lock - -# poetry -# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -#poetry.lock - -# pdm -# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -#pdm.lock -# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it -# in version control. -# https://pdm.fming.dev/#use-with-ide -.pdm.toml - -# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm -__pypackages__/ - -# Celery stuff -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py +.python-version # Environments .env @@ -130,41 +69,8 @@ ENV/ env.bak/ venv.bak/ -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pyre type checker -.pyre/ - -# pytype static type analyzer -.pytype/ +# pyright +/typings/ -# Cython debug symbols -cython_debug/ - -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ - -# Poetry local configuration file - https://python-poetry.org/docs/configuration/#local-configuration -poetry.toml - -# ruff -.ruff_cache/ - -.DS_Store +# logs +*.log diff --git a/CHANGELOG.md b/CHANGELOG.md index c3ad049..eb4f546 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## Version 0.6.0 + +- refactor: `HeadlessWidget` is not a singleton anymore, config module is now source + of truth for config +- refactor: initializing display and config is now done in `setup_headless_kivy` +- feat: class method `get_instance` to get the closest parent instance of type `HeadlessWidget` + ## Version 0.5.12 - fix: make sure `kivy.core.window` is loaded in `setup_headless` to avoid diff --git a/README.md b/README.md index df13853..8444a1e 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,7 @@ poetry --group dev headless-kivy-pi width=240, height=240, baudrate=60000000, - debug_mode=False, + is_debug_mode=False, display_class=ST7789, double_buffering=True, synchronous_clock=True, @@ -87,7 +87,7 @@ The height of the display in pixels. The baud rate for the display connection. -#### `debug_mode` +#### `is_debug_mode` If set to True, the application will print debug information, including FPS. diff --git a/headless_kivy_pi/__init__.py b/headless_kivy_pi/__init__.py index 79dd002..61de737 100644 --- a/headless_kivy_pi/__init__.py +++ b/headless_kivy_pi/__init__.py @@ -14,70 +14,42 @@ """ from __future__ import annotations -import atexit import time from pathlib import Path from queue import Empty, Queue from threading import Thread -from typing import TYPE_CHECKING, ClassVar +from typing import TYPE_CHECKING -import kivy import numpy as np -from kivy.app import App, ObjectProperty, Widget +from kivy.app import App from kivy.clock import Clock -from kivy.config import Config -from kivy.graphics import ( - Canvas, - ClearBuffers, - ClearColor, - Color, - Fbo, - Rectangle, -) +from kivy.graphics.context_instructions import Color +from kivy.graphics.fbo import Fbo +from kivy.graphics.gl_instructions import ClearBuffers, ClearColor +from kivy.graphics.instructions import Canvas +from kivy.graphics.vertex_instructions import Rectangle +from kivy.properties import ObjectProperty +from kivy.uix.widget import Widget from typing_extensions import Any -from headless_kivy_pi.constants import ( - BAUDRATE, - BITS_PER_BYTE, - BYTES_PER_PIXEL, - CLEAR_AT_EXIT, - IS_RPI, -) +from headless_kivy_pi import config +from headless_kivy_pi.constants import IS_TEST_ENVIRONMENT from headless_kivy_pi.display import transfer_to_display from headless_kivy_pi.logger import logger -from headless_kivy_pi.setup import SetupHeadlessConfig, setup_kivy if TYPE_CHECKING: from adafruit_rgb_display.rgb import DisplaySPI from kivy.graphics.texture import Texture -import board -import digitalio -from adafruit_rgb_display.st7789 import ST7789 - -kivy.require('2.2.1') - class HeadlessWidget(Widget): """Headless Kivy widget class rendering on SPI connected display.""" - instance: ClassVar[HeadlessWidget | None] = None - - _is_setup_headless_called: ClassVar = False - should_ignore_hash: ClassVar = False + _is_setup_headless_called: bool = False + should_ignore_hash: bool = False texture = ObjectProperty(None, allownone=True) _display: DisplaySPI - min_fps: int - max_fps: int - is_paused: bool - width: int - height: int - debug_mode: bool - double_buffering: bool - synchronous_clock: bool - automatic_fps_control: bool - last_second: int rendered_frames: int skipped_frames: int @@ -88,27 +60,24 @@ class HeadlessWidget(Widget): fbo: Fbo fbo_rect: Rectangle - def __init__(self: HeadlessWidget, **kwargs: Any) -> None: # noqa: ANN401 + def __init__(self: HeadlessWidget, **kwargs: dict[str, object]) -> None: """Initialize a `HeadlessWidget`.""" - if HeadlessWidget.instance: - msg = 'Only one instantiation of `HeadlessWidget` is possible' - raise RuntimeError(msg) + if not IS_TEST_ENVIRONMENT: + config.check_initialized() + + self.should_ignore_hash = False - if not HeadlessWidget._is_setup_headless_called: - msg = ( - 'You need to run `setup_headless` before instantiating `HeadlessWidget`' - ) - raise RuntimeError(msg) + __import__('kivy.core.window') - if self.debug_mode: + if config.is_debug_mode(): self.last_second = int(time.time()) self.rendered_frames = 0 self.skipped_frames = 0 - self.pending_render_threads = Queue(2 if HeadlessWidget.double_buffering else 1) + self.pending_render_threads = Queue(2 if config.double_buffering() else 1) self.last_hash = 0 self.last_change = time.time() - self.fps = self.max_fps + self.fps = config.max_fps() self.canvas = Canvas() with self.canvas: @@ -124,8 +93,6 @@ def __init__(self: HeadlessWidget, **kwargs: Any) -> None: # noqa: ANN401 super().__init__(**kwargs) - HeadlessWidget.instance = self - self.render_trigger = Clock.create_trigger( lambda _: self.render_on_display(), 1 / self.fps, @@ -133,8 +100,12 @@ def __init__(self: HeadlessWidget, **kwargs: Any) -> None: # noqa: ANN401 ) self.render_trigger() app = App.get_running_app() + + def clear(*_: object) -> None: + self.render_trigger.cancel() + if app: - app.bind(on_stop=lambda _: self.render_trigger.cancel()) + app.bind(on_stop=clear) def add_widget( self: HeadlessWidget, @@ -184,62 +155,44 @@ def on_alpha(self: HeadlessWidget, _: HeadlessWidget, value: float) -> None: """Update alpha value of `fbo_rect` when widget's alpha value changes.""" self.fbo_color.rgba = (1, 1, 1, value) - @classmethod - def pause(cls: type[HeadlessWidget]) -> None: - """Pause writing to the display.""" - HeadlessWidget.is_paused = True - - @classmethod - def resume(cls: type[HeadlessWidget]) -> None: - """Resume writing to the display.""" - cls.is_paused = False - cls.should_ignore_hash = True - - @classmethod - def render(cls: type[HeadlessWidget]) -> None: + def render(self: HeadlessWidget) -> None: """Schedule a force render.""" - if not cls.instance: + if not self: return - Clock.schedule_once(cls.instance.render_on_display, 0) + Clock.schedule_once(self.render_on_display, 0) - @classmethod - def _activate_high_fps_mode(cls: type[HeadlessWidget]) -> None: + def _activate_high_fps_mode(self: HeadlessWidget) -> None: """Increase fps to `max_fps`.""" - if not cls.instance: + if not self: return logger.info('Activating high fps mode, setting FPS to `max_fps`') - cls.instance.fps = cls.max_fps - cls.instance.render_trigger.timeout = 1.0 / cls.instance.fps - cls.instance.last_hash = 0 + self.fps = config.max_fps() + self.render_trigger.timeout = 1.0 / self.fps + self.last_hash = 0 - @classmethod - def activate_high_fps_mode(cls: type[HeadlessWidget]) -> None: + def activate_high_fps_mode(self: HeadlessWidget) -> None: """Schedule increasing fps to `max_fps`.""" - cls.render() - Clock.schedule_once(lambda _: cls._activate_high_fps_mode(), 0) + self.render() + Clock.schedule_once(lambda _: self._activate_high_fps_mode(), 0) - @classmethod - def _activate_low_fps_mode(cls: type[HeadlessWidget]) -> None: + def _activate_low_fps_mode(self: HeadlessWidget) -> None: """Drop fps to `min_fps`.""" - if not cls.instance: - return logger.info('Activating low fps mode, dropping FPS to `min_fps`') - cls.instance.fps = cls.min_fps - cls.instance.render_trigger.timeout = 1.0 / cls.instance.fps + self.fps = config.min_fps() + self.render_trigger.timeout = 1.0 / self.fps - @classmethod - def activate_low_fps_mode(cls: type[HeadlessWidget]) -> None: + def activate_low_fps_mode(self: HeadlessWidget) -> None: """Schedule dropping fps to `min_fps`.""" - cls.render() - Clock.schedule_once(lambda _: cls._activate_low_fps_mode(), 0) + self.render() + Clock.schedule_once(lambda _: self._activate_low_fps_mode(), 0) - def render_on_display(self: HeadlessWidget, *_: Any) -> None: # noqa: ANN401 + def render_on_display(self: HeadlessWidget, *_: object) -> None: """Render the widget on display connected to the SPI controller.""" - if HeadlessWidget.is_paused: + if config.is_paused(): return # Log the number of skipped and rendered frames in the last second - if self.debug_mode: + if config.is_debug_mode(): # Increment rendered_frames/skipped_frames count every frame and reset their # values to zero every second. current_second = int(time.time()) @@ -256,22 +209,20 @@ def render_on_display(self: HeadlessWidget, *_: Any) -> None: # noqa: ANN401 # If `synchronous_clock` is False, skip frames if there are more than one # pending render in case `double_buffering` is enabled, or if there are ANY # pending render in case `double_buffering` is disabled. - if ( - not HeadlessWidget.synchronous_clock - and self.pending_render_threads.qsize() - > (1 if HeadlessWidget.double_buffering else 0) + if not config.synchronous_clock() and self.pending_render_threads.qsize() > ( + 1 if config.double_buffering() else 0 ): self.skipped_frames += 1 return data = np.frombuffer(self.texture.pixels, dtype=np.uint8) data_hash = hash(data.data.tobytes()) - if data_hash == self.last_hash and not HeadlessWidget.should_ignore_hash: + if data_hash == self.last_hash and not self.should_ignore_hash: # Only drop FPS when the screen has not changed for at least one second if ( - self.automatic_fps_control + config.automatic_fps() and time.time() - self.last_change > 1 - and self.fps != self.min_fps + and self.fps != config.min_fps() ): logger.debug('Frame content has not changed for 1 second') self.activate_low_fps_mode() @@ -279,14 +230,14 @@ def render_on_display(self: HeadlessWidget, *_: Any) -> None: # noqa: ANN401 # Considering the content has not changed, this frame can safely be ignored return - if self.debug_mode: + if config.is_debug_mode(): self.rendered_frames += 1 with Path('headless_kivy_pi_buffer.raw').open('wb') as snapshot_file: snapshot_file.write( bytes( data.reshape( - HeadlessWidget.width, - HeadlessWidget.height, + config.width(), + config.height(), -1, )[::-1, :, :3] .flatten() @@ -294,11 +245,11 @@ def render_on_display(self: HeadlessWidget, *_: Any) -> None: # noqa: ANN401 ), ) - HeadlessWidget.should_ignore_hash = False + self.should_ignore_hash = False self.last_change = time.time() self.last_hash = data_hash - if self.automatic_fps_control and self.fps != self.max_fps: + if config.automatic_fps and self.fps != config.max_fps(): logger.debug('Frame content has changed') self.activate_high_fps_mode() @@ -314,82 +265,22 @@ def render_on_display(self: HeadlessWidget, *_: Any) -> None: # noqa: ANN401 data_hash, last_thread, ), + daemon=True, ) self.pending_render_threads.put(thread) thread.start() @classmethod - def setup_headless( + def get_instance( cls: type[HeadlessWidget], - config: SetupHeadlessConfig | None = None, - ) -> None: - """Set up `HeadlessWidget`.""" - if config is None: - config = {} - - setup_kivy(config) - baudrate = config.get('baudrate', BAUDRATE) - display_class: DisplaySPI = config.get('st7789', ST7789) - clear_at_exit = config.get('clear_at_exit', CLEAR_AT_EXIT) - - if HeadlessWidget.min_fps > HeadlessWidget.max_fps: - msg = f"""Invalid value "{HeadlessWidget.min_fps}" for "min_fps", it can't \ -be higher than 'max_fps' which is set to '{HeadlessWidget.max_fps}'.""" - raise ValueError(msg) - - fps_cap = baudrate / ( - HeadlessWidget.width - * HeadlessWidget.height - * BYTES_PER_PIXEL - * BITS_PER_BYTE - ) - - if HeadlessWidget.max_fps > fps_cap: - msg = f"""Invalid value "{HeadlessWidget.max_fps}" for "max_fps", it can't \ -be higher than "{fps_cap:.1f}" (baudrate={baudrate} ÷ (width={HeadlessWidget.width} x \ -height={HeadlessWidget.height} x bytes per pixel={BYTES_PER_PIXEL} x bits per byte=\ -{BITS_PER_BYTE}))""" - raise ValueError( - msg, - ) - - from kivy.core.window import Window - - if IS_RPI: - Config.set('graphics', 'window_state', 'hidden') - spi = board.SPI() - # Configuration for CS and DC pins (these are PiTFT defaults): - cs_pin = digitalio.DigitalInOut(board.CE0) - dc_pin = digitalio.DigitalInOut(board.D25) - reset_pin = digitalio.DigitalInOut(board.D24) - display = display_class( - spi, - height=HeadlessWidget.height, - width=HeadlessWidget.width, - y_offset=80, - x_offset=0, - cs=cs_pin, - dc=dc_pin, - rst=reset_pin, - baudrate=baudrate, - ) - HeadlessWidget._display = display - if clear_at_exit: - atexit.register( - lambda: display._block( # noqa: SLF001 - 0, - 0, - HeadlessWidget.width - 1, - HeadlessWidget.height - 1, - bytes(HeadlessWidget.width * HeadlessWidget.height * 2), - ), - ) - else: - from screeninfo import get_monitors + widget: Widget, + ) -> HeadlessWidget | None: + """Get the nearest instance of `HeadlessWidget`.""" + if isinstance(widget, HeadlessWidget): + return widget + if widget.parent: + return cls.get_instance(widget.parent) + return None - monitor = get_monitors()[0] - Window._win.set_always_on_top(True) # noqa: SLF001, FBT003 - Window._set_top(200) # noqa: SLF001 - Window._set_left(monitor.width - Window._size[0]) # noqa: SLF001 - HeadlessWidget._is_setup_headless_called = True +__all__ = ['HeadlessWidget'] diff --git a/headless_kivy_pi/config.py b/headless_kivy_pi/config.py new file mode 100644 index 0000000..12e966b --- /dev/null +++ b/headless_kivy_pi/config.py @@ -0,0 +1,267 @@ +# pyright: reportMissingImports=false +"""Implement `setup_kivy`, it configures Kivy.""" +from __future__ import annotations + +import atexit +import sys +from functools import cache +from typing import TYPE_CHECKING, NoReturn, NotRequired, TypedDict + +import kivy +from kivy.config import Config + +from headless_kivy_pi.constants import ( + AUTOMATIC_FPS, + BAUDRATE, + BITS_PER_BYTE, + BYTES_PER_PIXEL, + CLEAR_AT_EXIT, + DOUBLE_BUFFERING, + HEIGHT, + IS_DEBUG_MODE, + IS_RPI, + IS_TEST_ENVIRONMENT, + MAX_FPS, + MIN_FPS, + SYNCHRONOUS_CLOCK, + WIDTH, +) +from headless_kivy_pi.fake import Fake +from headless_kivy_pi.logger import add_file_handler, add_stdout_handler + +if not IS_RPI: + sys.modules['board'] = Fake() + sys.modules['digitalio'] = Fake() + sys.modules['adafruit_rgb_display.st7789'] = Fake() + +import board +import digitalio +from adafruit_rgb_display.st7789 import ST7789 + +kivy.require('2.3.0') + +if TYPE_CHECKING: + from adafruit_rgb_display.rgb import DisplaySPI + + +class SetupHeadlessConfig(TypedDict): + """Arguments of `setup_headless_kivy` function.""" + + """Minimum frames per second for when the Kivy application is idle.""" + min_fps: NotRequired[int] + """Maximum frames per second for the Kivy application.""" + max_fps: NotRequired[int] + """The width of the display in pixels.""" + width: NotRequired[int] + """The height of the display in pixels.""" + height: NotRequired[int] + """The baud rate for the display connection.""" + baudrate: NotRequired[int] + """If set to True, the application will consume computational resources to log + additional debug information.""" + is_debug_mode: NotRequired[bool] + """The display class to use (default is ST7789).""" + display_class: NotRequired[type[DisplaySPI]] + """Is set to `True`, it will let Kivy generate the next frame while sending the + last frame to the display.""" + double_buffering: NotRequired[bool] + """If set to `True`, Kivy will wait for the LCD before rendering next frames. This + will cause Headless to skip frames if they are rendered before the LCD has finished + displaying the previous frames. If set to False, frames will be rendered + asynchronously, letting Kivy render frames regardless of display being able to catch + up or not at the expense of possible frame skipping.""" + synchronous_clock: NotRequired[bool] + """If set to `True`, it will monitor the hash of the screen data, if this hash + changes, it will increase the fps to the maximum and if the hash doesn't change for + a while, it will drop the fps to the minimum.""" + automatic_fps: NotRequired[bool] + """If set to `True`, it will clear the screen before exiting.""" + clear_at_eixt: NotRequired[bool] + + +_config: SetupHeadlessConfig | None = None +_display: DisplaySPI | None = None + + +def report_uninitialized() -> NoReturn: + """Report that the module has not been initialized.""" + msg = """You need to run `setup_headless_kivy` before importing \ +`kivy.core.window` module. \ +Note that it might have been imported by another module unintentionally.""" + raise RuntimeError(msg) + + +def setup_headless_kivy(config: SetupHeadlessConfig) -> None: + """Configure the headless mode for the Kivy application. + + Arguments: + --------- + config: `SetupHeadlessConfig` + + """ + global _config, _display # noqa: PLW0603 + _config = config + + if is_debug_mode(): + add_stdout_handler() + add_file_handler() + + Config.set('kivy', 'kivy_clock', 'default') + Config.set('graphics', 'fbo', 'force-hardware') + Config.set('graphics', 'fullscreen', '0') + Config.set('graphics', 'maxfps', f'{max_fps()}') + Config.set('graphics', 'multisamples', '1') + Config.set('graphics', 'resizable', '0') + Config.set('graphics', 'vsync', '0') + Config.set('graphics', 'width', f'{width()}') + Config.set('graphics', 'height', f'{height()}') + + baudrate = config.get('baudrate', BAUDRATE) + display_class: DisplaySPI = config.get('st7789', ST7789) + clear_at_exit = config.get('clear_at_exit', CLEAR_AT_EXIT) + + if min_fps() > max_fps(): + msg = f"""Invalid value "{min_fps()}" for "min_fps", it can't \ +be higher than 'max_fps' which is set to '{max_fps()}'.""" + raise ValueError(msg) + + fps_cap = baudrate / (width() * height() * BYTES_PER_PIXEL * BITS_PER_BYTE) + + if max_fps() > fps_cap: + msg = f"""Invalid value "{max_fps()}" for "max_fps", it can't \ +be higher than "{fps_cap:.1f}" (baudrate={baudrate} ÷ (width={width()} x \ +height={height()} x bytes per pixel={BYTES_PER_PIXEL} x bits per byte=\ +{BITS_PER_BYTE}))""" + raise ValueError(msg) + + if IS_TEST_ENVIRONMENT: + Config.set('graphics', 'window_state', 'hidden') + from kivy.core.window import Window + elif IS_RPI: + Config.set('graphics', 'window_state', 'hidden') + spi = board.SPI() + # Configuration for CS and DC pins (these are PiTFT defaults): + cs_pin = digitalio.DigitalInOut(board.CE0) + dc_pin = digitalio.DigitalInOut(board.D25) + reset_pin = digitalio.DigitalInOut(board.D24) + _display = display_class( + spi, + height=height(), + width=width(), + y_offset=80, + x_offset=0, + cs=cs_pin, + dc=dc_pin, + rst=reset_pin, + baudrate=baudrate, + ) + if clear_at_exit: + atexit.register( + lambda: _display + and _display._block( # noqa: SLF001 + 0, + 0, + width() - 1, + height() - 1, + bytes(width() * height() * 2), + ), + ) + else: + from kivy.core.window import Window + from screeninfo import get_monitors + + monitor = get_monitors()[0] + + Window._win.set_always_on_top(True) # noqa: SLF001 + Window._set_top(200) # noqa: SLF001 + Window._set_left(monitor.width - Window._size[0]) # noqa: SLF001 + + +def check_initialized() -> None: + """Check if the module has been initialized.""" + if not _config: + report_uninitialized() + + +@cache +def min_fps() -> int: + """Return the minimum frames per second for when the Kivy application is idle.""" + if _config: + return _config.get('min_fps', MIN_FPS) + report_uninitialized() + + +@cache +def max_fps() -> int: + """Return the maximum frames per second for the Kivy application.""" + if _config: + return _config.get('max_fps', MAX_FPS) + report_uninitialized() + + +@cache +def width() -> int: + """Return the width of the display in pixels.""" + if _config: + return _config.get('width', WIDTH) + report_uninitialized() + + +@cache +def height() -> int: + """Return the height of the display in pixels.""" + if _config: + return _config.get('height', HEIGHT) + report_uninitialized() + + +@cache +def is_debug_mode() -> bool: + """Return `True` if the application will consume computational resources to log.""" + if _config: + return _config.get('is_debug_mode', IS_DEBUG_MODE) + report_uninitialized() + + +@cache +def double_buffering() -> bool: + """Generate the next frame while sending the last frame to the display.""" + if _config: + return _config.get('double_buffering', DOUBLE_BUFFERING) + report_uninitialized() + + +@cache +def synchronous_clock() -> bool: + """headless-kivy-pi will wait for the LCD before rendering next frames.""" + if _config: + return _config.get('synchronous_clock', SYNCHRONOUS_CLOCK) + report_uninitialized() + + +@cache +def automatic_fps() -> bool: + """headless-kivy-pi adjusts the FPS automatically.""" + if _config: + return _config.get('automatic_fps', AUTOMATIC_FPS) + report_uninitialized() + + +_is_paused: bool = False + + +def is_paused() -> bool: + """Return `True` if rendering the application is paused.""" + return _is_paused + + +def pause() -> None: + """Pause rendering the application.""" + global _is_paused # noqa: PLW0603 + _is_paused = True + + +def resume() -> None: + """Resume rendering the application.""" + global _is_paused # noqa: PLW0603 + _is_paused = False diff --git a/headless_kivy_pi/constants.py b/headless_kivy_pi/constants.py index f20a44e..5da51b4 100644 --- a/headless_kivy_pi/constants.py +++ b/headless_kivy_pi/constants.py @@ -12,12 +12,15 @@ # Configure the headless mode for the Kivy application and initialize the display +IS_TEST_ENVIRONMENT = ( + strtobool(os.environ.get('HEADLESS_KIVY_PI_TEST_ENVIRONMENT', 'False')) == 1 +) MIN_FPS = int(os.environ.get('HEADLESS_KIVY_PI_MIN_FPS', '1')) MAX_FPS = int(os.environ.get('HEADLESS_KIVY_PI_MAX_FPS', '32')) WIDTH = int(os.environ.get('HEADLESS_KIVY_PI_WIDTH', '240')) HEIGHT = int(os.environ.get('HEADLESS_KIVY_PI_HEIGHT', '240')) BAUDRATE = int(os.environ.get('HEADLESS_KIVY_PI_BAUDRATE', '60000000')) -DEBUG_MODE = ( +IS_DEBUG_MODE = ( strtobool( os.environ.get('HEADLESS_KIVY_PI_DEBUG', 'False' if IS_RPI else 'True'), ) @@ -35,9 +38,9 @@ ) == 1 ) -AUTOMATIC_FPS_CONTROL = ( +AUTOMATIC_FPS = ( strtobool( - os.environ.get('HEADLESS_KIVY_PI_AUTOMATIC_FPS_CONTROL', 'True'), + os.environ.get('HEADLESS_KIVY_PI_AUTOMATIC_FPS', 'True'), ) == 1 ) diff --git a/headless_kivy_pi/display.py b/headless_kivy_pi/display.py index aaa3224..78e810a 100644 --- a/headless_kivy_pi/display.py +++ b/headless_kivy_pi/display.py @@ -5,7 +5,7 @@ import numpy as np -from headless_kivy_pi.constants import IS_RPI +from headless_kivy_pi import config from headless_kivy_pi.logger import logger if TYPE_CHECKING: @@ -20,14 +20,12 @@ def transfer_to_display( last_render_thread: Thread, ) -> None: """Transfer data to the display via SPI controller.""" - from headless_kivy_pi import HeadlessWidget - logger.debug(f'Rendering frame with hash "{data_hash}"') # Flip the image vertically data = data.reshape( - HeadlessWidget.width, - HeadlessWidget.height, + config.width(), + config.height(), -1, )[::-1, :, :3].astype(np.uint16) @@ -45,11 +43,12 @@ def transfer_to_display( last_render_thread.join() # Only render when running on a Raspberry Pi - if IS_RPI: - HeadlessWidget._display._block( # noqa: SLF001 + display = config._display # noqa: SLF001 + if display: + display._block( # noqa: SLF001 0, 0, - HeadlessWidget.width - 1, - HeadlessWidget.height - 1, + config.width() - 1, + config.height() - 1, data_bytes, ) diff --git a/headless_kivy_pi/setup.py b/headless_kivy_pi/setup.py deleted file mode 100644 index d74b727..0000000 --- a/headless_kivy_pi/setup.py +++ /dev/null @@ -1,120 +0,0 @@ -# pyright: reportMissingImports=false -"""Implement `setup_kivy`, it configures Kivy.""" -from __future__ import annotations - -import sys -from typing import TYPE_CHECKING, NotRequired, TypedDict - -from kivy import Config - -from headless_kivy_pi.constants import ( - AUTOMATIC_FPS_CONTROL, - DEBUG_MODE, - DOUBLE_BUFFERING, - HEIGHT, - IS_RPI, - MAX_FPS, - MIN_FPS, - SYNCHRONOUS_CLOCK, - WIDTH, -) -from headless_kivy_pi.fake import Fake -from headless_kivy_pi.logger import add_file_handler, add_stdout_handler - -if TYPE_CHECKING: - from adafruit_rgb_display.rgb import DisplaySPI - -if not IS_RPI: - sys.modules['board'] = Fake() - sys.modules['digitalio'] = Fake() - sys.modules['adafruit_rgb_display.st7789'] = Fake() - - -class SetupHeadlessConfig(TypedDict): - """Arguments of `setup_headless` function.""" - - """Minimum frames per second for when the Kivy application is idle.""" - min_fps: NotRequired[int] - """Maximum frames per second for the Kivy application.""" - max_fps: NotRequired[int] - """The width of the display in pixels.""" - width: NotRequired[int] - """The height of the display in pixels.""" - height: NotRequired[int] - """The baud rate for the display connection.""" - baudrate: NotRequired[int] - """If set to True, the application will consume computational resources to log - additional debug information.""" - debug_mode: NotRequired[bool] - """The display class to use (default is ST7789).""" - display_class: NotRequired[type[DisplaySPI]] - """Is set to `True`, it will let Kivy generate the next frame while sending the - last frame to the display.""" - double_buffering: NotRequired[bool] - """If set to `True`, Kivy will wait for the LCD before rendering next frames. This - will cause Headless to skip frames if they are rendered before the LCD has finished - displaying the previous frames. If set to False, frames will be rendered - asynchronously, letting Kivy render frames regardless of display being able to catch - up or not at the expense of possible frame skipping.""" - synchronous_clock: NotRequired[bool] - """If set to `True`, it will monitor the hash of the screen data, if this hash - changes, it will increase the fps to the maximum and if the hash doesn't change for - a while, it will drop the fps to the minimum.""" - automatic_fps: NotRequired[bool] - """If set to `True`, it will clear the screen before exiting.""" - clear_at_eixt: NotRequired[bool] - - -def setup_kivy(config: SetupHeadlessConfig) -> None: - """Configure the headless mode for the Kivy application. - - Arguments: - --------- - config: `SetupHeadlessConfig`, optional - """ - from headless_kivy_pi import HeadlessWidget - - if 'kivy.core.window' in sys.modules: - msg = """You need to run `setup_headless` before importing \ -`kivy.core.window` module. \ -Note that it might have been imported by another module unintentionally.""" - raise RuntimeError(msg) - - min_fps = config.get('min_fps', MIN_FPS) - max_fps = config.get('max_fps', MAX_FPS) - width = config.get('width', WIDTH) - height = config.get('height', HEIGHT) - HeadlessWidget.debug_mode = config.get('debug_mode', DEBUG_MODE) - HeadlessWidget.double_buffering = config.get( - 'double_buffering', - DOUBLE_BUFFERING, - ) - HeadlessWidget.synchronous_clock = config.get( - 'synchronous_clock', - SYNCHRONOUS_CLOCK, - ) - HeadlessWidget.automatic_fps_control = config.get( - 'automatic_fps', - AUTOMATIC_FPS_CONTROL, - ) - - if HeadlessWidget.debug_mode: - add_stdout_handler() - add_file_handler() - - HeadlessWidget.min_fps = min_fps - HeadlessWidget.max_fps = max_fps - HeadlessWidget.width = width - HeadlessWidget.height = height - - HeadlessWidget.is_paused = False - - Config.set('kivy', 'kivy_clock', 'default') - Config.set('graphics', 'fbo', 'force-hardware') - Config.set('graphics', 'fullscreen', '0') - Config.set('graphics', 'maxfps', f'{max_fps}') - Config.set('graphics', 'multisamples', '1') - Config.set('graphics', 'resizable', '0') - Config.set('graphics', 'vsync', '0') - Config.set('graphics', 'width', f'{width}') - Config.set('graphics', 'height', f'{height}') diff --git a/pyproject.toml b/pyproject.toml index dfe9f7f..9f27852 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "headless-kivy-pi" -version = "0.5.12" +version = "0.6.0" description = "Headless renderer for Kivy framework on Raspberry Pi" authors = ["Sassan Haradji "] license = "Apache-2.0" @@ -49,3 +49,9 @@ multiline-quotes = "double" [tool.ruff.format] quote-style = 'single' + +[tool.isort] +profile = "black" + +[tool.pyright] +exclude = ['typings']