Merge branch 'mock' into 'master' (f05e287f) · Commits · PoolPay Heidelberg / PoolPay

README.md

+10 −0

Changes for README.md: 10 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -269,6 +269,16 @@ cd /coli/poolpay
		uv run -m poolpay.card id
		```

		## Development

		You can run the application in mock mode, which will simulate display, touch screen and card reader:

		```python
		uv run -m poolpay --mock
		```

		The mock card reader will listen on the command line for an integer card id, usually followed by `r` to simulate card removal. Quitting the program should work via <kbd>Ctrl</kbd>+<kbd>C/<kbd>, although it seems like not all threads immediately terminate, i.e., you might need to press it several times.

		## License

		This project is licensed under the [MIT license](https://opensource.org/licenses/MIT), for details see the file [`LICENSE`](LICENSE).

poolpay/main.py

+32 −6

Changes for poolpay/__main__.py: 32 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		import argparse
		import logging
		import signal
		from pathlib import Path
		@@ -6,11 +7,8 @@ from types import FrameType

		from poolpay import db, paths
		from poolpay.card.CardReaderException import CardReaderException
		from poolpay.card.Pirc522CardReader import Pirc522CardReader
		from poolpay.director.Director import Director
		from poolpay.display.PygameDisplay import PygameDisplay
		from poolpay.model.messages.decode import decode
		from poolpay.touch.EvdevTouchScreen import EvdevTouchScreen
		from poolpay.ui.pygame.play import play
		from poolpay.wire.Packet import Packet
		from poolpay.wire.Server import Server
		@@ -19,16 +17,44 @@ logging.basicConfig(
		level=logging.DEBUG, format="[%(levelname)s] <%(module)s.%(funcName)s> %(message)s"
		)


		def _get_args() -> argparse.Namespace:
		parser = argparse.ArgumentParser(prog="uv run -m poolpay")
		parser.add_argument(
		"--mock",
		action="store_true",
		help="Simulate display, touch screen and card reader",
		)
		return parser.parse_args()


		args = _get_args()

		logging.info("Starting PoolPay")
		logging.info("Creating Vault, Display and Director")
		logging.info("Creating Display, Touch Screen and Director")

		if args.mock:
		from poolpay.card.MockCardReader import MockCardReader
		from poolpay.display.PygameMockDisplay import PygameMockDisplay
		from poolpay.touch.MockTouchScreen import MockTouchScreen

		display = PygameMockDisplay(Path("/dev/null"), width=480, height=320)
		touch_screen = MockTouchScreen()
		else:
		from poolpay.card.Pirc522CardReader import Pirc522CardReader
		from poolpay.display.PygameDisplay import PygameDisplay
		from poolpay.touch.EvdevTouchScreen import EvdevTouchScreen

		display = PygameDisplay(Path("/dev/fb1"), width=480, height=320)
		touch_screen = EvdevTouchScreen(Path("/dev/input/event0"))

		director = Director(play, display)


		def _initialize_card_reader_and_run() -> None:
		logging.info("Initializing Pirc522CardReader")
		with Pirc522CardReader() as card_reader:
		CardReader = MockCardReader if args.mock else Pirc522CardReader
		logging.info(f"Initializing {CardReader.__name__}")
		with CardReader() as card_reader:
		logging.info("Registering card removal handler")
		card_reader.on_card_removed(lambda reader: director.card_removed())

poolpay/card/MockCardReader.py

0 → 100644

+35 −0

Changes for poolpay/card/MockCardReader.py: 35 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from collections.abc import Callable
		from dataclasses import dataclass, field

		from poolpay.card.CardReader import CardReader


		@dataclass
		class MockCardReader(CardReader):
		_handle_removal: Callable[[CardReader], None] = field(
		init=False, default=lambda _: None
		)
		_card_id: int \| None = field(init=False, default=None)

		def read_id(self) -> int \| None:
		return self._card_id

		def on_card_placed(self, handle_card: Callable[[CardReader], None]) -> None:
		while True:
		line = input("Card ID (int) or 'r' to remove card> ").strip()

		if line.lower() == "r":
		if self._card_id is not None:
		self._card_id = None
		self._handle_removal(self)
		else:
		self._card_id = int(line)
		handle_card(self)

		def on_card_removed(self, handle_removal: Callable[[CardReader], None]) -> None:
		self._handle_removal = handle_removal

		def close(self) -> None:
		pass

poolpay/display/PygameMockDisplay.py

0 → 100644

+64 −0

Changes for poolpay/display/PygameMockDisplay.py: 64 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from dataclasses import dataclass, field
		from functools import cached_property
		from queue import SimpleQueue
		from threading import Thread
		from typing import Literal

		import pygame
		from pygame import Surface

		from poolpay.display.Display import Display

		type Request = tuple[int, int] \| Literal["refresh"]
		type Response = Surface


		class MockDisplayThread(Thread):
		def __init__(self) -> None:
		super().__init__(name="mock_display", daemon=True)
		self._request_queue: SimpleQueue[Request] = SimpleQueue()
		self._response_queue: SimpleQueue[Response] = SimpleQueue()

		def get_response(self) -> Response:
		return self._response_queue.get(block=True)

		def request(self, request: Request) -> None:
		self._request_queue.put(request)

		def run(self) -> None:
		pygame.init()

		while True:
		request = self._request_queue.get(block=True)

		match request:
		case (width, height):
		screen = pygame.display.set_mode((width, height))
		self._response_queue.put(screen)
		case "refresh":
		pygame.display.flip()


		@dataclass(frozen=True)
		class PygameMockDisplay(Display):
		_display_thread: MockDisplayThread = field(
		init=False, default_factory=MockDisplayThread
		)

		@cached_property
		def surface(self) -> Surface:
		self._display_thread.start()
		self._display_thread.request((self.width, self.height))

		return self._display_thread.get_response()

		def refresh(self) -> None:
		self._display_thread.request("refresh")

		def show(self, picture: bytes) -> None:
		pass


		__all__ = ["PygameMockDisplay"]

poolpay/touch/MockTouchScreen.py

0 → 100644

+23 −0

Changes for poolpay/touch/MockTouchScreen.py: 23 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from collections.abc import Callable
		from threading import Thread

		import pygame

		from poolpay.touch.TouchScreen import TouchScreen


		class MockTouchScreen(TouchScreen):
		def on_screen_touched(self, handle_touch: Callable[[], None]) -> None:
		def listen() -> None:
		while True:
		for event in pygame.event.get():
		if event.type == pygame.MOUSEBUTTONDOWN:
		handle_touch()
		pygame.time.wait(10)

		Thread(target=listen, daemon=True, name="touchscreen").start()

		def close(self) -> None:
		pass