9 Commits

Author SHA1 Message Date
b69f4bd87d feat: copy build unix executable to other build artifacts
Some checks failed
Build RPi Pico firmware image / Build-Firmware (push) Has been cancelled
Check code formatting / Check-C-Format (push) Has been cancelled
Check code formatting / Check-Python-Flake8 (push) Has been cancelled
Check code formatting / Check-Bash-Shellcheck (push) Has been cancelled
Run unit tests on host / Run-Unit-Tests (push) Has been cancelled
Run pytests / Check-Pytest (push) Has been cancelled
2025-12-03 20:59:28 +01:00
82ed3a3c2e Merge pull request '27-configuration-storage' (#53) from 27-configuration-storage into main
All checks were successful
Build RPi Pico firmware image / Build-Firmware (push) Successful in 4m42s
Check code formatting / Check-C-Format (push) Successful in 8s
Check code formatting / Check-Python-Flake8 (push) Successful in 10s
Check code formatting / Check-Bash-Shellcheck (push) Successful in 6s
Run unit tests on host / Run-Unit-Tests (push) Successful in 9s
Run pytests / Check-Pytest (push) Successful in 12s
Reviewed-on: #53
Reviewed-by: Stefan Kratochwil <kratochwil-la@gmx.de>
2025-12-03 19:20:26 +00:00
a7e58853bb feat: Add api/v1/config to get, put config
All checks were successful
Build RPi Pico firmware image / Build-Firmware (push) Successful in 4m40s
Check code formatting / Check-C-Format (push) Successful in 9s
Check code formatting / Check-Python-Flake8 (push) Successful in 12s
Check code formatting / Check-Bash-Shellcheck (push) Successful in 6s
Run unit tests on host / Run-Unit-Tests (push) Successful in 10s
Run pytests / Check-Pytest (push) Successful in 12s
Signed-off-by: Matthias Blankertz <matthias@blankertz.org>
2025-12-03 20:12:29 +01:00
2e1bc7782b fix: Ensure each timer can only be scheduled once
All checks were successful
Build RPi Pico firmware image / Build-Firmware (push) Successful in 4m58s
Check code formatting / Check-C-Format (push) Successful in 8s
Check code formatting / Check-Python-Flake8 (push) Successful in 11s
Check code formatting / Check-Bash-Shellcheck (push) Successful in 6s
Run unit tests on host / Run-Unit-Tests (push) Successful in 10s
Run pytests / Check-Pytest (push) Successful in 13s
If a timer is scheduled with the timer manager while it is already
scheduled, the new exipry time should override the prev. expiry time
instead of adding a second instance of the same timer function.

Signed-off-by: Matthias Blankertz <matthias@blankertz.org>
2025-12-01 20:25:30 +01:00
fa0e23ee87 feat: Add play/pause and prev track function
Signed-off-by: Matthias Blankertz <matthias@blankertz.org>
2025-12-01 20:25:30 +01:00
856bf34161 feat: Make button mapping configurable
- Remove Pin to button mapping from hwconfig, replace with a simple list
  of Pins to which buttons are attached
- Add BUTTON_MAP to config.json which maps indices in the HW button list
  to button names
- Update utils.Buttons to use the button map to connect the correct pins
  to the matching key codes

Signed-off-by: Matthias Blankertz <matthias@blankertz.org>
2025-12-01 20:25:30 +01:00
83deb1b4c2 feat: Replace hardcoded timeouts in app with configurable
Signed-off-by: Matthias Blankertz <matthias@blankertz.org>
2025-12-01 20:25:30 +01:00
2225906664 feat: Move LED_COUNT from hwconfig to config.json
Signed-off-by: Matthias Blankertz <matthias@blankertz.org>
2025-12-01 20:25:30 +01:00
176fc66c17 feat: config store
Load a config.json file from the root of the littlefs flash filesystem.

Signed-off-by: Matthias Blankertz <matthias@blankertz.org>
2025-12-01 20:25:30 +01:00
12 changed files with 241 additions and 43 deletions

View File

@@ -17,6 +17,7 @@ set -eu
)
BUILDDIR=lib/micropython/ports/rp2/build-TONBERRY_RPI_PICO_W/
BUILDDIR_UNIX=lib/micropython/ports/unix/build-tonberry_unix/
OUTDIR=$(pwd)/build
mkdir -p "$OUTDIR"
FS_STAGE_DIR=$(mktemp -d)
@@ -51,5 +52,9 @@ for hwconfig in boards/RPI_PICO_W/manifest-*.py; do
$PICOTOOL uf2 convert $BUILDDIR/firmware-filesystem.bin "$OUTDIR"/firmware-filesystem-"$hwname".uf2
done
cp "$BUILDDIR_UNIX"/micropython "$OUTDIR"/micropython-tonberry_unix
chmod u+x "$OUTDIR"/micropython-tonberry_unix
echo "Output in" "${OUTDIR}"/firmware-*.uf2
echo "Images with filesystem in" "${OUTDIR}"/firmware-filesystem-*.uf2
echo "Unix build in" "${OUTDIR}"/micropython-tonberry_unix

View File

@@ -6,7 +6,8 @@ import time
from utils import TimerManager
Dependencies = namedtuple('Dependencies', ('mp3player', 'nfcreader', 'buttons', 'playlistdb', 'hwconfig', 'leds'))
Dependencies = namedtuple('Dependencies', ('mp3player', 'nfcreader', 'buttons', 'playlistdb', 'hwconfig', 'leds',
'config'))
# Should be ~ 6dB steps
VOLUME_CURVE = [1, 2, 4, 8, 16, 32, 63, 126, 251]
@@ -14,11 +15,12 @@ VOLUME_CURVE = [1, 2, 4, 8, 16, 32, 63, 126, 251]
class PlayerApp:
class TagStateMachine:
def __init__(self, parent, timer_manager):
def __init__(self, parent, timer_manager, timeout=5000):
self.parent = parent
self.timer_manager = timer_manager
self.current_tag = None
self.current_tag_time = time.ticks_ms()
self.timeout = timeout
def onTagChange(self, new_tag):
if new_tag is not None:
@@ -31,7 +33,7 @@ class PlayerApp:
self.current_tag = new_tag
self.parent.onNewTag(new_tag)
else:
self.timer_manager.schedule(time.ticks_ms() + 5000, self.onTagRemoveDelay)
self.timer_manager.schedule(time.ticks_ms() + self.timeout, self.onTagRemoveDelay)
def onTagRemoveDelay(self):
if self.current_tag is not None:
@@ -40,7 +42,10 @@ class PlayerApp:
def __init__(self, deps: Dependencies):
self.timer_manager = TimerManager()
self.tag_state_machine = self.TagStateMachine(self, self.timer_manager)
self.config = deps.config(self)
self.tag_timeout_ms = self.config.get_tag_timeout() * 1000
self.idle_timeout_ms = self.config.get_idle_timeout() * 1000
self.tag_state_machine = self.TagStateMachine(self, self.timer_manager, self.tag_timeout_ms)
self.player = deps.mp3player(self)
self.nfc = deps.nfcreader(self.tag_state_machine)
self.playlist_db = deps.playlistdb(self)
@@ -52,6 +57,7 @@ class PlayerApp:
self.buttons = deps.buttons(self) if deps.buttons is not None else None
self.mp3file = None
self.volume_pos = 3
self.paused = False
self.player.set_volume(VOLUME_CURVE[self.volume_pos])
self._onIdle()
@@ -91,6 +97,10 @@ class PlayerApp:
self.player.set_volume(VOLUME_CURVE[self.volume_pos])
elif what == self.buttons.NEXT:
self._play_next()
elif what == self.buttons.PREV:
self._play_prev()
elif what == self.buttons.PLAY_PAUSE:
self._pause_toggle()
def onPlaybackDone(self):
assert self.mp3file is not None
@@ -103,7 +113,7 @@ class PlayerApp:
self.hwconfig.power_off()
else:
# Check again in a minute
self.timer_manager.schedule(time.ticks_ms() + 60*1000, self.onIdleTimeout)
self.timer_manager.schedule(time.ticks_ms() + self.idle_timeout_ms, self.onIdleTimeout)
def _set_playlist(self, tag: bytes):
if self.playlist is not None:
@@ -131,6 +141,15 @@ class PlayerApp:
self.playlist = None
self.playing_tag = None
def _play_prev(self):
if self.playlist is None:
return
filename = self.playlist.getPrevPath()
self._play(filename)
if filename is None:
self.playlist = None
self.playing_tag = None
def _play(self, filename: bytes | None, offset=0):
if self.mp3file is not None:
self.player.stop()
@@ -141,10 +160,21 @@ class PlayerApp:
print(f'Playing {filename!r}')
self.mp3file = open(filename, 'rb')
self.player.play(self.mp3file, offset)
self.paused = False
self._onActive()
def _pause_toggle(self):
if self.playlist is None:
return
if self.paused:
self._play(self.playlist.getCurrentPath(), self.pause_offset)
else:
self.pause_offset = self.player.stop()
self.paused = True
self._onIdle()
def _onIdle(self):
self.timer_manager.schedule(time.ticks_ms() + 60*1000, self.onIdleTimeout)
self.timer_manager.schedule(time.ticks_ms() + self.idle_timeout_ms, self.onIdleTimeout)
self.leds.set_state(self.leds.IDLE)
def _onActive(self):

View File

@@ -28,13 +28,14 @@ RC522_SS = Pin.board.GP13
# WS2812
LED_DIN = Pin.board.GP16
LED_COUNT = 1
# Buttons
BUTTON_VOLUP = Pin.board.GP17
BUTTON_VOLDOWN = Pin.board.GP19
BUTTON_NEXT = Pin.board.GP18
BUTTON_POWER = Pin.board.GP21
BUTTONS = [Pin.board.GP17,
Pin.board.GP18,
Pin.board.GP19,
Pin.board.GP20,
Pin.board.GP21,
]
# Power
POWER_EN = Pin.board.GP22

View File

@@ -27,13 +27,12 @@ RC522_SS = Pin.board.GP13
# WS2812
LED_DIN = Pin.board.GP16
LED_COUNT = 1
# Buttons
BUTTON_VOLUP = Pin.board.GP17
BUTTON_VOLDOWN = Pin.board.GP19
BUTTON_NEXT = Pin.board.GP18
BUTTON_POWER = None
BUTTONS = [Pin.board.GP17,
Pin.board.GP18,
Pin.board.GP19,
]
# Power
POWER_EN = None

View File

@@ -18,7 +18,7 @@ from mfrc522 import MFRC522
from mp3player import MP3Player
from nfc import Nfc
from rp2_neopixel import NeoPixel
from utils import BTreeFileManager, Buttons, SDContext, TimerManager, LedManager
from utils import BTreeFileManager, Buttons, SDContext, TimerManager, LedManager, Configuration
from webserver import start_webserver
try:
@@ -55,15 +55,17 @@ def setup_wifi():
DB_PATH = '/sd/tonberry.db'
config = Configuration()
def run():
asyncio.new_event_loop()
# Setup LEDs
np = NeoPixel(hwconfig.LED_DIN, hwconfig.LED_COUNT, sm=1)
np = NeoPixel(hwconfig.LED_DIN, config.get_led_count(), sm=1)
# Wifi with default config
setup_wifi()
start_webserver()
start_webserver(config)
# Setup MP3 player
with SDContext(mosi=hwconfig.SD_DI, miso=hwconfig.SD_DO, sck=hwconfig.SD_SCK, ss=hwconfig.SD_CS,
@@ -87,12 +89,11 @@ def run():
# Setup app
deps = app.Dependencies(mp3player=lambda the_app: MP3Player(audioctx, the_app),
nfcreader=lambda the_app: Nfc(reader, the_app),
buttons=lambda the_app: Buttons(the_app, pin_volup=hwconfig.BUTTON_VOLUP,
pin_voldown=hwconfig.BUTTON_VOLDOWN,
pin_next=hwconfig.BUTTON_NEXT),
buttons=lambda the_app: Buttons(the_app, config, hwconfig),
playlistdb=lambda _: playlistdb,
hwconfig=lambda _: hwconfig,
leds=lambda _: LedManager(np))
leds=lambda _: LedManager(np),
config=lambda _: config)
the_app = app.PlayerApp(deps)
# Start
@@ -121,5 +122,5 @@ def builddb():
if __name__ == '__main__':
time.sleep(1)
if machine.Pin(hwconfig.BUTTON_VOLUP, machine.Pin.IN, machine.Pin.PULL_UP).value() != 0:
if machine.Pin(hwconfig.BUTTONS[0], machine.Pin.IN, machine.Pin.PULL_UP).value() != 0:
run()

View File

@@ -2,6 +2,7 @@
# Copyright (c) 2025 Matthias Blankertz <matthias@blankertz.org>
from utils.buttons import Buttons
from utils.config import Configuration
from utils.leds import LedManager
from utils.mbrpartition import MBRPartition
from utils.pinindex import get_pin_index
@@ -9,5 +10,5 @@ from utils.playlistdb import BTreeDB, BTreeFileManager
from utils.sdcontext import SDContext
from utils.timer import TimerManager
__all__ = ["BTreeDB", "BTreeFileManager", "Buttons", "get_pin_index", "LedManager", "MBRPartition", "SDContext",
"TimerManager"]
__all__ = ["BTreeDB", "BTreeFileManager", "Buttons", "Configuration", "get_pin_index", "LedManager", "MBRPartition",
"SDContext", "TimerManager"]

View File

@@ -17,14 +17,27 @@ if TYPE_CHECKING:
class Buttons:
def __init__(self, cb: "ButtonCallback", pin_volup=17, pin_voldown=19, pin_next=18):
self.VOLUP = micropython.const(1)
self.VOLDOWN = micropython.const(2)
self.NEXT = micropython.const(3)
VOLUP = micropython.const(1)
VOLDOWN = micropython.const(2)
NEXT = micropython.const(3)
PREV = micropython.const(4)
PLAY_PAUSE = micropython.const(5)
KEYMAP = {VOLUP: 'VOLUP',
VOLDOWN: 'VOLDOWN',
NEXT: 'NEXT',
PREV: 'PREV',
PLAY_PAUSE: 'PLAY_PAUSE'}
def __init__(self, cb: "ButtonCallback", config, hwconfig):
self.button_map = config.get_button_map()
self.hw_buttons = hwconfig.BUTTONS
self.cb = cb
self.buttons = {machine.Pin(pin_volup, machine.Pin.IN, machine.Pin.PULL_UP): self.VOLUP,
machine.Pin(pin_voldown, machine.Pin.IN, machine.Pin.PULL_UP): self.VOLDOWN,
machine.Pin(pin_next, machine.Pin.IN, machine.Pin.PULL_UP): self.NEXT}
self.buttons = dict()
for key_id, key_name in self.KEYMAP.items():
pin = self._get_pin(key_name)
if pin is None:
continue
self.buttons[pin] = key_id
self.int_flag = asyncio.ThreadSafeFlag()
self.pressed: list[int] = []
self.last: dict[int, int] = {}
@@ -32,6 +45,17 @@ class Buttons:
button.irq(handler=self._interrupt, trigger=machine.Pin.IRQ_FALLING | machine.Pin.IRQ_RISING)
asyncio.create_task(self.task())
def _get_pin(self, key):
key_id = self.button_map.get(key, None)
if key_id is None:
return None
if key_id < 0 or key_id >= len(self.hw_buttons):
return None
pin = self.hw_buttons[key_id]
if pin is not None:
pin.init(machine.Pin.IN, machine.Pin.PULL_UP)
return pin
def _interrupt(self, button):
keycode = self.buttons[button]
last = self.last.get(keycode, 0)

View File

@@ -0,0 +1,91 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2025 Matthias Blankertz <matthias@blankertz.org>
from errno import ENOENT
import json
import os
try:
from typing import TYPE_CHECKING, Mapping, Any
except ImportError:
TYPE_CHECKING = False
class Configuration:
DEFAULT_CONFIG = {
'LED_COUNT': 1,
'IDLE_TIMEOUT_SECS': 60,
'TAG_TIMEOUT_SECS': 5,
'BUTTON_MAP': {
'PLAY_PAUSE': 4,
'VOLUP': 0,
'VOLDOWN': 2,
'PREV': None,
'NEXT': 1,
}
}
def __init__(self, config_path='/config.json'):
self.config_path = config_path
try:
with open(self.config_path, 'r') as conf_file:
self.config = json.load(conf_file)
except OSError as ex:
if ex.errno == ENOENT:
self.config = Configuration.DEFAULT_CONFIG
self._save()
else:
raise
except ValueError as ex:
print(f"Warning: Could not load configuration {self.config_path}:\n{ex}")
self._move_config_to_backup()
self.config = Configuration.DEFAULT_CONFIG
def _move_config_to_backup(self):
# Remove old backup
try:
os.remove(self.config_path + '.bup')
os.rename(self.config_path, self.config_path + '.bup')
except OSError as ex:
if ex.errno != ENOENT:
raise
os.sync()
def _save(self):
with open(self.config_path + '.new', 'w') as conf_file:
json.dump(self.config, conf_file)
self._move_config_to_backup()
os.rename(self.config_path + '.new', self.config_path)
os.sync()
def _get(self, key):
return self.config.get(key, self.DEFAULT_CONFIG[key])
def get_led_count(self) -> int:
return self._get('LED_COUNT')
def get_idle_timeout(self) -> int:
return self._get('IDLE_TIMEOUT_SECS')
def get_tag_timeout(self) -> int:
return self._get('TAG_TIMEOUT_SECS')
def get_button_map(self) -> Mapping[str, int | None]:
return self._get('BUTTON_MAP')
# For the web API
def get_config(self) -> Mapping[str, Any]:
return self.config
def _validate(self, default, config, path=''):
for k in config.keys():
if k not in default:
raise ValueError(f'Invalid config key {path}/{k}')
if isinstance(default[k], dict):
if not isinstance(config[k], dict):
raise ValueError(f'Invalid config: Value of {path}/{k} must be mapping')
self._validate(default[k], config[k], f'{path}/{k}')
def set_config(self, config):
self._validate(self.DEFAULT_CONFIG, config)
self.config = config
self._save()

View File

@@ -107,6 +107,17 @@ class BTreeDB(IPlaylistDB):
self.setPlaybackOffset(0)
return self.getCurrentPath()
def getPrevPath(self):
"""
Select prev track and return path.
"""
if self.pos > 0:
self.pos -= 1
if self.persist != BTreeDB.PERSIST_NO:
self.parent._setPlaylistPos(self.tag, self.pos)
self.setPlaybackOffset(0)
return self.getCurrentPath()
def setPlaybackOffset(self, offset):
"""
Store the current position in the track for PERSIST_OFFSET mode

View File

@@ -22,6 +22,7 @@ class TimerManager(object):
def schedule(self, when, what):
cur_nearest = self.timers[0][0] if len(self.timers) > 0 else None
self._remove_timer(what) # Ensure timer is not already scheduled
heapq.heappush(self.timers, (when, what))
if cur_nearest is None or cur_nearest > self.timers[0][0]:
# New timer is closer than previous closest timer
@@ -31,18 +32,22 @@ class TimerManager(object):
self.worker_event.set()
def cancel(self, what):
remove_idx = self._remove_timer(what)
if remove_idx == 0:
# Cancel timer was closest timer
if self.timer_debug:
print("cancel: wake")
self.worker_event.set()
return True
def _remove_timer(self, what):
try:
(when, _), i = next(filter(lambda item: item[0][1] == what, zip(self.timers, range(len(self.timers)))))
except StopIteration:
return False
del self.timers[i]
heapq.heapify(self.timers)
if i == 0:
# Cancel timer was closest timer
if self.timer_debug:
print("cancel: wake")
self.worker_event.set()
return True
return i
async def _timer_worker(self):
while True:

View File

@@ -9,11 +9,13 @@ from microdot import Microdot
webapp = Microdot()
server = None
config = None
def start_webserver():
global server
def start_webserver(config_):
global server, config
server = asyncio.create_task(webapp.start_server(port=80))
config = config_
@webapp.route('/')
@@ -39,3 +41,17 @@ async def filesystem_post(request):
async def playlist_post(request):
print(request)
return {'success': False}
@webapp.route('/api/v1/config', methods=['GET'])
async def config_get(request):
return config.get_config()
@webapp.route('/api/v1/config', methods=['PUT'])
async def config_put(request):
try:
config.set_config(request.json)
except ValueError as ex:
return str(ex), 400
return '', 204

View File

@@ -124,6 +124,19 @@ class FakeLeds:
self.state = state
class FakeConfig:
def __init__(self): pass
def get_led_count(self):
return 1
def get_idle_timeout(self):
return 60
def get_tag_timeout(self):
return 5
def fake_open(filename, mode):
return FakeFile(filename, mode)
@@ -136,13 +149,14 @@ def faketimermanager(monkeypatch):
def _makedeps(mp3player=FakeMp3Player, nfcreader=FakeNfcReader, buttons=FakeButtons,
playlistdb=FakePlaylistDb, hwconfig=FakeHwconfig, leds=FakeLeds):
playlistdb=FakePlaylistDb, hwconfig=FakeHwconfig, leds=FakeLeds, config=FakeConfig):
return app.Dependencies(mp3player=lambda _: mp3player() if callable(mp3player) else mp3player,
nfcreader=lambda x: nfcreader(x) if callable(nfcreader) else nfcreader,
buttons=lambda _: buttons() if callable(buttons) else buttons,
playlistdb=lambda _: playlistdb() if callable(playlistdb) else playlistdb,
hwconfig=lambda _: hwconfig() if callable(hwconfig) else hwconfig,
leds=lambda _: leds() if callable(leds) else leds)
leds=lambda _: leds() if callable(leds) else leds,
config=lambda _: config() if callable(config) else config)
def test_construct_app(micropythonify, faketimermanager):