Merge pull request 'feat/filesystem_viewer' (#70) from feat/filesystem_viewer into main

Reviewed-on: #70
Reviewed-by: Matthias Blankertz <matthias@blankertz.org>

.gitea/workflows/build.yaml

@@ -11,15 +11,17 @@ jobs:
         uses: actions/checkout@v4
       - name: Initialize submodules
         run: cd software && ./update-submodules.sh
+      - name: Prepare venv
+        run: python -m venv build-venv && source build-venv/bin/activate && pip install freezefs
       - name: Build
-        run: cd software && ./build.sh
+        run: source build-venv/bin/activate && cd software && ./build.sh
       - name: Upload firmware
         uses: actions/upload-artifact@v3
         with:
           name: firmware-RPi-Pico-W
-          path: software/lib/micropython/ports/rp2/build-TONBERRY_RPI_PICO_W/firmware.uf2
+          path: software/build/firmware-*.uf2
       - name: Upload firmware w/ filesystem
         uses: actions/upload-artifact@v3
         with:
           name: firmware-RPi-Pico-W-with-fs
-          path: software/lib/micropython/ports/rp2/build-TONBERRY_RPI_PICO_W/firmware-filesystem-*.uf2
+          path: software/build/firmware-filesystem-*.uf2

DEVELOP.md

@@ -52,3 +52,23 @@ would be stored in the following key/value pairs in the btree db:
  * 00aa11bb22/playlist/00000: a.mp3
  * 00aa11bb22/playlist/00001: b.mp3
  * 00aa11bb22/playlistpos: 00000
+
+## Notes for UI development with chromium
+
+Features for the web interface are best prototyped in a browser directly. By using the built-in developmer tools and
+and their "override" feature, the web contents are replaced by a locally stored copy, which can be used to directly
+test the modifications without going all the way through the build and flash process.
+
+However, modern browsers may restrict or even completely forbid the execution of dynamic content like JavaScript, if
+the content is stored on the local machine and/or the content is accessed using http. In such a case, chromium issues
+an error message similar to the following one:
+
+> Access to fetch at 'http://192.168.4.1/api/v1/audiofiles' from origin 'http://192.168.4.1' has been blocked by CORS
+> policy: The request client is not a secure context and the resource is in more-private address space `local`.
+
+To mitigate this, chromium offers two flags that need modification:
+- 'chrome://flags/#local-network-access-check' must be `Disabled`
+- 'chrome://flags/#unsafely-treat-insecure-origin-as-secure' must be `Enabled`
+
+Note that these settings leave the browser susceptible to security issues and should be returned to
+their default values as soon as possible.

software/boards/RPI_PICO_W/board.c

@@ -0,0 +1,29 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2025 Matthias Blankertz <matthias@blankertz.org>
+
+#include "py/obj.h"
+#include "py/runtime.h"
+#include "py/objstr.h"
+
+#ifndef TONBERRY_GIT_REVISION
+#define TONBERRY_GIT_REVISION "unknown"
+#endif
+#ifndef TONBERRY_VERSION
+#define TONBERRY_VERSION "unknown"
+#endif
+
+static const MP_DEFINE_STR_OBJ(tonberry_git_revision_obj, TONBERRY_GIT_REVISION);
+static const MP_DEFINE_STR_OBJ(tonberry_version_obj, TONBERRY_VERSION);
+
+static const mp_rom_map_elem_t board_module_globals_table[] = {
+    {MP_ROM_QSTR(MP_QSTR___name__), MP_ROM_QSTR(MP_QSTR_board)},
+    {MP_ROM_QSTR(MP_QSTR_revision), MP_ROM_PTR(&tonberry_git_revision_obj)},
+    {MP_ROM_QSTR(MP_QSTR_version), MP_ROM_PTR(&tonberry_version_obj)},
+};
+static MP_DEFINE_CONST_DICT(board_module_globals, board_module_globals_table);
+
+const mp_obj_module_t board_cmodule = {
+    .base = {&mp_type_module},
+    .globals = (mp_obj_dict_t *)&board_module_globals,
+};
+MP_REGISTER_MODULE(MP_QSTR_board, board_cmodule);

software/boards/RPI_PICO_W/manifest-Rev1.py

@@ -0,0 +1,3 @@
+include("manifest.py")
+
+module("hwconfig.py", "../../src/hwconfig_Rev1")

software/boards/RPI_PICO_W/manifest-breadboard.py

@@ -0,0 +1,3 @@
+include("manifest.py")
+
+module("hwconfig.py", "../../src/hwconfig_breadboard")

software/boards/RPI_PICO_W/manifest.py

@@ -15,3 +15,12 @@ module("microdot.py", "../../lib/microdot/src/microdot/")
 # TonberryPico modules
 module("audiocore.py", "../../modules/audiocore")
 module("rp2_neopixel.py", "../../modules")
+
+module("main.py", "../../src")
+module("app.py", "../../src")
+module("mp3player.py", "../../src")
+module("webserver.py", "../../src")
+package("utils", base_path="../../src")
+package("nfc", base_path="../../src")
+
+module("frozen_frontend.py", "../../build")

software/boards/RPI_PICO_W/mpconfigboard.cmake

@@ -20,3 +20,22 @@ set(GEN_PINS_CSV_ARG --board-csv "${GEN_PINS_BOARD_CSV}")
 
 add_link_options("-Wl,--print-memory-usage")
 set(PICO_USE_FASTEST_SUPPORTED_CLOCK 1)
+
+find_program(GIT git)
+
+execute_process(COMMAND ${GIT} rev-parse HEAD
+    WORKING_DIRECTORY "${CMAKE_CURRENT_LIST_DIR}"
+    OUTPUT_VARIABLE TONBERRY_GIT_REVISION
+    OUTPUT_STRIP_TRAILING_WHITESPACE
+)
+
+execute_process(COMMAND ${GIT} describe --match 'v*' --always --dirty
+    WORKING_DIRECTORY "${CMAKE_CURRENT_LIST_DIR}"
+    OUTPUT_VARIABLE TONBERRY_VERSION
+    OUTPUT_STRIP_TRAILING_WHITESPACE
+)
+
+set(MICROPY_SOURCE_BOARD "${CMAKE_CURRENT_LIST_DIR}/board.c")
+set(MICROPY_DEF_BOARD
+    TONBERRY_GIT_REVISION="${TONBERRY_GIT_REVISION}"
+    TONBERRY_VERSION="${TONBERRY_VERSION}")

software/boards/tonberry_unix/manifest.py

@@ -0,0 +1,5 @@
+include("$(PORT_DIR)/variants/manifest.py")
+
+include("$(MPY_DIR)/extmod/asyncio")
+
+module("microdot.py", "../../lib/microdot/src/microdot/")

software/boards/tonberry_unix/mpconfigvariant.h

@@ -0,0 +1,31 @@
+/*
+ * This file is part of the MicroPython project, http://micropython.org/
+ *
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2019 Damien P. George
+ *
+ * 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.
+ */
+
+// Set base feature level.
+#define MICROPY_CONFIG_ROM_LEVEL (MICROPY_CONFIG_ROM_LEVEL_EXTRA_FEATURES)
+
+// Enable extra Unix features.
+#include "mpconfigvariant_common.h"

software/boards/tonberry_unix/mpconfigvariant.mk

@@ -0,0 +1,3 @@
+# This is the default variant when you `make` the Unix port.
+
+FROZEN_MANIFEST ?= $(VARIANT_DIR)/manifest.py

software/boards/tonberry_unix/mpconfigvariant_common.h

@@ -0,0 +1,126 @@
+/*
+ * This file is part of the MicroPython project, http://micropython.org/
+ *
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2022 Jim Mussared
+ *
+ * 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.
+ */
+
+// This file enables and configures features common to all variants
+// other than "minimal".
+
+// Send raise KeyboardInterrupt directly from the signal handler rather than
+// scheduling it into the VM.
+#define MICROPY_ASYNC_KBD_INTR         (!MICROPY_PY_THREAD_GIL)
+
+// Enable helpers for printing debugging information.
+#ifndef MICROPY_DEBUG_PRINTERS
+#define MICROPY_DEBUG_PRINTERS         (1)
+#endif
+
+// Enable floating point by default.
+#ifndef MICROPY_FLOAT_IMPL
+#define MICROPY_FLOAT_IMPL             (MICROPY_FLOAT_IMPL_DOUBLE)
+#endif
+
+// Don't use native _Float16 because it increases code size by a lot.
+#ifndef MICROPY_FLOAT_USE_NATIVE_FLT16
+#define MICROPY_FLOAT_USE_NATIVE_FLT16 (0)
+#endif
+
+// Enable arbitrary precision long-int by default.
+#ifndef MICROPY_LONGINT_IMPL
+#define MICROPY_LONGINT_IMPL           (MICROPY_LONGINT_IMPL_MPZ)
+#endif
+
+// Enable use of C libraries that need read/write/lseek/fsync, e.g. axtls.
+#define MICROPY_STREAMS_POSIX_API      (1)
+
+// REPL conveniences.
+#define MICROPY_REPL_EMACS_WORDS_MOVE  (1)
+#define MICROPY_REPL_EMACS_EXTRA_WORDS_MOVE (1)
+#define MICROPY_USE_READLINE_HISTORY   (1)
+#ifndef MICROPY_READLINE_HISTORY_SIZE
+#define MICROPY_READLINE_HISTORY_SIZE  (50)
+#endif
+
+// Seed random on import.
+#define MICROPY_PY_RANDOM_SEED_INIT_FUNC (mp_random_seed_init())
+
+// Allow exception details in low-memory conditions.
+#define MICROPY_ENABLE_EMERGENCY_EXCEPTION_BUF (1)
+#define MICROPY_EMERGENCY_EXCEPTION_BUF_SIZE (256)
+
+// Allow loading of .mpy files.
+#define MICROPY_PERSISTENT_CODE_LOAD   (1)
+
+// Extra memory debugging.
+#define MICROPY_MALLOC_USES_ALLOCATED_SIZE (1)
+#define MICROPY_MEM_STATS              (1)
+
+// Enable a small performance boost for the VM.
+#define MICROPY_OPT_COMPUTED_GOTO      (1)
+
+// Return number of collected objects from gc.collect().
+#define MICROPY_PY_GC_COLLECT_RETVAL   (1)
+
+// Enable detailed error messages and warnings.
+#define MICROPY_ERROR_REPORTING     (MICROPY_ERROR_REPORTING_DETAILED)
+#define MICROPY_WARNINGS               (1)
+#define MICROPY_PY_STR_BYTES_CMP_WARN  (1)
+
+// Configure the "sys" module with features not usually enabled on bare-metal.
+#define MICROPY_PY_SYS_ATEXIT          (1)
+#define MICROPY_PY_SYS_EXC_INFO        (1)
+
+// Configure the "os" module with extra unix features.
+#define MICROPY_PY_OS_INCLUDEFILE      "ports/unix/modos.c"
+#define MICROPY_PY_OS_ERRNO            (1)
+#define MICROPY_PY_OS_GETENV_PUTENV_UNSETENV (1)
+#define MICROPY_PY_OS_SYSTEM           (1)
+#define MICROPY_PY_OS_URANDOM          (1)
+
+// Enable the unix-specific "time" module.
+#define MICROPY_PY_TIME                (1)
+#define MICROPY_PY_TIME_TIME_TIME_NS   (1)
+#define MICROPY_PY_TIME_CUSTOM_SLEEP   (1)
+#define MICROPY_PY_TIME_INCLUDEFILE    "ports/unix/modtime.c"
+
+#if MICROPY_PY_SSL
+#define MICROPY_PY_HASHLIB_MD5         (1)
+#define MICROPY_PY_HASHLIB_SHA1        (1)
+#define MICROPY_PY_CRYPTOLIB           (1)
+#endif
+
+// The "select" module is enabled by default, but disable select.select().
+#define MICROPY_PY_SELECT_POSIX_OPTIMISATIONS (1)
+#define MICROPY_PY_SELECT_SELECT       (0)
+
+// Enable the "websocket" module.
+#define MICROPY_PY_WEBSOCKET           (1)
+
+// Enable the "machine" module, mostly for machine.mem*.
+#define MICROPY_PY_MACHINE             (1)
+#define MICROPY_PY_MACHINE_PULSE       (1)
+#define MICROPY_PY_MACHINE_PIN_BASE    (1)
+
+#define MICROPY_VFS_ROM                (1)
+#define MICROPY_VFS_ROM_IOCTL          (0)

software/build.sh

@@ -6,15 +6,41 @@ set -eu
 
 ( cd lib/micropython
   make -C mpy-cross -j "$(nproc)"
-  make -C ports/rp2 BOARD=TONBERRY_RPI_PICO_W BOARD_DIR="$TOPDIR"/boards/RPI_PICO_W clean
-  make -C ports/rp2 BOARD=TONBERRY_RPI_PICO_W BOARD_DIR="$TOPDIR"/boards/RPI_PICO_W \
-       USER_C_MODULES="$TOPDIR"/modules/micropython.cmake -j "$(nproc)"
+
+  # build tonberry specific unix port of micropython
+  make -C ports/unix VARIANT_DIR="$TOPDIR"/boards/tonberry_unix clean
+  make -C ports/unix VARIANT_DIR="$TOPDIR"/boards/tonberry_unix -j "$(nproc)"
 )
 
 ( cd tools/mklittlefs
   make -j "$(nproc)"
 )
 
+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)
+mkdir  "$FS_STAGE_DIR"/fs
+trap 'rm -rf $FS_STAGE_DIR' EXIT
+tools/mklittlefs/mklittlefs -p 256 -s 868352 -c "$FS_STAGE_DIR"/fs "$FS_STAGE_DIR"/filesystem.bin
+
+FRONTEND_STAGE_DIR=$(mktemp -d)
+trap 'rm -rf $FRONTEND_STAGE_DIR' EXIT
+gzip -c frontend/index.html > "$FRONTEND_STAGE_DIR"/index.html.gz
+python -m freezefs "$FRONTEND_STAGE_DIR" build/frozen_frontend.py --target=/frontend
+
+for hwconfig in boards/RPI_PICO_W/manifest-*.py; do
+    hwconfig_base=$(basename "$hwconfig")
+    hwname=${hwconfig_base##manifest-}
+    hwname=${hwname%%.py}
+    hwconfig_abs=$(realpath "$hwconfig")
+    ( cd lib/micropython
+      make -C ports/rp2 BOARD=TONBERRY_RPI_PICO_W BOARD_DIR="$TOPDIR"/boards/RPI_PICO_W clean
+      make -C ports/rp2 BOARD=TONBERRY_RPI_PICO_W BOARD_DIR="$TOPDIR"/boards/RPI_PICO_W \
+           USER_C_MODULES="$TOPDIR"/modules/micropython.cmake \
+           FROZEN_MANIFEST="$hwconfig_abs" -j "$(nproc)"
+    )
     PICOTOOL=picotool
     if ! command -v $PICOTOOL >/dev/null 2>&1; then
         echo "system picotool not found, checking SDK build dir"
@@ -24,22 +50,16 @@ if ! command -v $PICOTOOL >/dev/null 2>&1; then
             exit 1
         fi
     fi
-BUILDDIR=lib/micropython/ports/rp2/build-TONBERRY_RPI_PICO_W/
-FS_STAGE_DIR=$(mktemp -d)
-trap 'rm -rf $FS_STAGE_DIR' EXIT
-for hwconfig in src/hwconfig_*.py; do
-    hwconfig_base=$(basename "$hwconfig")
-    hwname=${hwconfig_base##hwconfig_}
-    hwname=${hwname%%.py}
-    find src/ -iname '*.py' \! -iname 'hwconfig_*.py' | cpio -pdm "$FS_STAGE_DIR"
-    cp "$hwconfig" "$FS_STAGE_DIR"/src/hwconfig.py
-    tools/mklittlefs/mklittlefs -p 256 -s 868352 -c "$FS_STAGE_DIR"/src $BUILDDIR/filesystem.bin
     truncate -s 2M $BUILDDIR/firmware-filesystem.bin
     dd if=$BUILDDIR/firmware.bin of=$BUILDDIR/firmware-filesystem.bin bs=1k
-    dd if=$BUILDDIR/filesystem.bin of=$BUILDDIR/firmware-filesystem.bin bs=1k seek=1200
-    $PICOTOOL uf2 convert $BUILDDIR/firmware-filesystem.bin $BUILDDIR/firmware-filesystem-"$hwname".uf2
-    rm -r "${FS_STAGE_DIR:?}"/*
+    dd if="$FS_STAGE_DIR"/filesystem.bin of=$BUILDDIR/firmware-filesystem.bin bs=1k seek=1200
+    cp $BUILDDIR/firmware.uf2 "$OUTDIR"/firmware-"$hwname".uf2
+    $PICOTOOL uf2 convert $BUILDDIR/firmware-filesystem.bin "$OUTDIR"/firmware-filesystem-"$hwname".uf2
 done
 
-echo "Output in $BUILDDIR/firmware.uf2"
-echo "Images with filesystem in" ${BUILDDIR}firmware-filesystem-*.uf2
+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

software/flash.sh

@@ -2,6 +2,8 @@
 
 set -eu
 
+TOPDIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+
 check_command()
 {
     name=$1
@@ -16,7 +18,7 @@ check_command lsusb
 check_command picotool
 
 DEVICEPATH=/dev/disk/by-label/RPI-RP2
-IMAGEPATH=lib/micropython/ports/rp2/build-TONBERRY_RPI_PICO_W/
+IMAGEPATH=${TOPDIR}/build
 REVISION=Rev1
 
 flash_via_mountpoint()
@@ -41,7 +43,7 @@ flash_via_picotool()
     local device="${bus_device[1]//[!0-9]/}"
     echo "Found RP2 with serial $serial on Bus $bus Device $device"
     
-    picotool load --bus "$bus" --address "$device" "$IMAGEFILE"
+    picotool load --update --bus "$bus" --address "$device" "$IMAGEFILE"
 }
 
 FLASH_VIA_MOUNTPOINT=0
@@ -83,7 +85,7 @@ if [ $# -gt 0 ]; then
     usage
 fi
 
-IMAGEFILE="$IMAGEPATH"/firmware-filesystem-$REVISION.uf2
+IMAGEFILE="$IMAGEPATH"/firmware-$REVISION.uf2
 
 if [ "$FLASH_VIA_MOUNTPOINT" -eq 0 ]; then
     flash_via_picotool

software/requirements.txt

@@ -0,0 +1 @@
+freezefs

software/src/app.py

@@ -6,19 +6,21 @@ import time
 from utils import TimerManager
 
 
-Dependencies = namedtuple('Dependencies', ('mp3player', 'nfcreader', 'buttons', 'playlistdb'))
+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]
+# Should be ~ 3dB steps
+VOLUME_CURVE = [1, 2, 3, 4, 6, 8, 11, 16, 23, 32, 45, 64, 91, 128, 181, 255]
 
 
 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,17 +42,33 @@ 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)
-        self.tag_mode = self.playlist_db.getSetting('tagmode')
+        self.hwconfig = deps.hwconfig(self)
+        self.leds = deps.leds(self)
+        self.tag_mode = self.config.get_tagmode()
+        self.volume_max = self.config.get_volume_max()
+        self.volume_pos = 3  # fallback if config.get_volume_boot is nonsense
+        try:
+            for idx, val in enumerate(VOLUME_CURVE):
+                if val >= self.config.get_volume_boot():
+                    self.volume_pos = idx
+                    break
+        except (TypeError, ValueError):
+            pass
         self.playing_tag = None
         self.playlist = None
         self.buttons = deps.buttons(self) if deps.buttons is not None else None
         self.mp3file = None
-        self.volume_pos = 3
+        self.paused = False
+        self.playing = False
         self.player.set_volume(VOLUME_CURVE[self.volume_pos])
+        self._onIdle()
 
     def __del__(self):
         if self.mp3file is not None:
@@ -64,7 +82,7 @@ class PlayerApp:
         uid_str = b''.join('{:02x}'.format(x).encode() for x in new_tag)
         if self.tag_mode == 'tagremains' or (self.tag_mode == 'tagstartstop' and new_tag != self.playing_tag):
             self._set_playlist(uid_str)
-            self.playing_tag = new_tag
+            self.playing_tag = new_tag if self.playlist is not None else None
         elif self.tag_mode == 'tagstartstop':
             print('Tag presented again, stopping playback')
             self._unset_playlist()
@@ -81,22 +99,42 @@ class PlayerApp:
     def onButtonPressed(self, what):
         assert self.buttons is not None
         if what == self.buttons.VOLUP:
-            self.volume_pos = min(self.volume_pos + 1, len(VOLUME_CURVE) - 1)
+            new_volume = min(self.volume_pos + 1, len(VOLUME_CURVE) - 1)
+            if VOLUME_CURVE[new_volume] <= self.volume_max:
+                self.volume_pos = new_volume
                 self.player.set_volume(VOLUME_CURVE[self.volume_pos])
         elif what == self.buttons.VOLDOWN:
             self.volume_pos = max(self.volume_pos - 1, 0)
             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
-        self.mp3file.close()
-        self.mp3file = None
         self._play_next()
 
+    def onIdleTimeout(self):
+        if self.hwconfig.get_on_battery():
+            self.hwconfig.power_off()
+        else:
+            # Check again in a minute
+            self.timer_manager.schedule(time.ticks_ms() + self.idle_timeout_ms, self.onIdleTimeout)
+
+    def reset_idle_timeout(self):
+        if not self.playing:
+            self.timer_manager.schedule(time.ticks_ms() + self.idle_timeout_ms, self.onIdleTimeout)
+
+    def is_playing(self) -> bool:
+        return self.playing
+
     def _set_playlist(self, tag: bytes):
-        self._unset_playlist()
+        if self.playlist is not None:
+            pos = self.player.stop()
+            if pos is not None:
+                self.playlist.setPlaybackOffset(pos)
         self.playlist = self.playlist_db.getPlaylistForTag(tag)
         self._play(self.playlist.getCurrentPath() if self.playlist is not None else None,
                    self.playlist.getPlaybackOffset() if self.playlist is not None else 0)
@@ -104,14 +142,22 @@ class PlayerApp:
     def _unset_playlist(self):
         if self.playlist is not None:
             pos = self.player.stop()
+            self._onIdle()
             if pos is not None:
                 self.playlist.setPlaybackOffset(pos)
             self.playlist = None
 
     def _play_next(self):
+        filename = self.playlist.getNextPath() if self.playlist is not None else None
+        self._play(filename)
+        if filename is None:
+            self.playlist = None
+            self.playing_tag = None
+
+    def _play_prev(self):
         if self.playlist is None:
             return
-        filename = self.playlist.getNextPath()
+        filename = self.playlist.getPrevPath()
         self._play(filename)
         if filename is None:
             self.playlist = None
@@ -122,7 +168,43 @@ class PlayerApp:
             self.player.stop()
             self.mp3file.close()
             self.mp3file = None
+            self._onIdle()
         if filename is not None:
             print(f'Playing {filename!r}')
+            try:
                 self.mp3file = open(filename, 'rb')
+            except OSError as ex:
+                print(f"Could not play file {filename}: {ex}")
+                return
             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() + self.idle_timeout_ms, self.onIdleTimeout)
+        self.leds.set_state(self.leds.IDLE)
+        self.playing = False
+
+    def _onActive(self):
+        self.timer_manager.cancel(self.onIdleTimeout)
+        self.leds.set_state(self.leds.PLAYING)
+        self.playing = True
+
+    def get_nfc(self):
+        return self.nfc
+
+    def get_playlist_db(self):
+        return self.playlist_db
+
+    def get_leds(self):
+        return self.leds

software/src/hwconfig_Rev1/hwconfig.py

@@ -28,17 +28,19 @@ 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
 VBAT_ADC = Pin.board.GP26
+VBUS_DET = Pin.board.WL_GPIO2
 
 
 def board_init():
@@ -61,3 +63,13 @@ def get_battery_voltage():
     adc = machine.ADC(VBAT_ADC)     # create ADC object on ADC pin
     battv = adc.read_u16()/65535.0*3.3*2
     return battv
+
+
+def power_off():
+    POWER_EN.init(mode=Pin.OUT)
+    POWER_EN.value(0)
+
+
+def get_on_battery():
+    vbus = VBUS_DET.value()
+    return not vbus

software/src/hwconfig_breadboard/hwconfig.py

@@ -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
@@ -47,3 +46,12 @@ def board_init():
 def get_battery_voltage():
     # Not supported on breadboard
     return None
+
+
+def power_off():
+    # Not supported on breadboard
+    pass
+
+
+def get_on_battery():
+    return False

software/src/led_test.py

@@ -1,58 +0,0 @@
-# SPDX-License-Identifier: MIT
-# Copyright (c) 2024 Matthias Blankertz <matthias@blankertz.org>
-
-# Run with mpremote.py run src/led_test.py
-
-from machine import Pin
-from math import pi, sin, pow
-from micropython import const
-from rp2_neopixel import NeoPixel
-from time import sleep, ticks_ms
-import asyncio
-
-pin = Pin.board.GP16
-leds = const(10)
-brightness = 0.5
-
-np = NeoPixel(pin, leds)
-
-# test fill and write
-
-print("LEDs should now turn red")
-np.fill((255, 0, 0))
-np.write()
-sleep(1)
-
-print("LEDs should now turn green")
-np.fill((0, 255, 0))
-np.write()
-sleep(1)
-
-print("LEDs should now turn blue")
-np.fill((0, 0, 255))
-np.write()
-sleep(1)
-
-
-# test async
-def gamma(value, X=2.2):
-    return min(max(int(brightness * pow(value / 255.0, X) * 255.0 + 0.5), 0), 255)
-
-
-async def rainbow(np, period=10):
-    count = 0.0
-    while True:
-        for i in range(leds):
-            ofs = (count + i) % leds
-            np[i] = (gamma((sin(ofs / leds * 2 * pi) + 1) * 127),
-                     gamma((sin(ofs / leds * 2 * pi + 2/3*pi) + 1) * 127),
-                     gamma((sin(ofs / leds * 2 * pi + 4/3*pi) + 1) * 127))
-        count += 0.2
-        before = ticks_ms()
-        await np.async_write()
-        now = ticks_ms()
-        if before + 20 > now:
-            await asyncio.sleep_ms(20 - (now - before))
-
-print("LEDs should now start rainbowing")
-asyncio.run(rainbow(np))

software/src/main.py

@@ -3,22 +3,23 @@
 
 import aiorepl  # type: ignore
 import asyncio
-from errno import ENOENT
 import machine
 import micropython
 import network
-import os
 import time
-from math import pi, sin, pow
+import ubinascii
+import sys
 
 # Own modules
 import app
 from audiocore import AudioContext
+import frozen_frontend  # noqa: F401
 from mfrc522 import MFRC522
 from mp3player import MP3Player
 from nfc import Nfc
 from rp2_neopixel import NeoPixel
-from utils import BTreeFileManager, Buttons, SDContext, TimerManager
+from utils import BTreeFileManager, Buttons, SDContext, TimerManager, LedManager, Configuration
+from webserver import start_webserver
 
 try:
     import hwconfig
@@ -31,63 +32,77 @@ micropython.alloc_emergency_exception_buf(100)
 # Machine setup
 hwconfig.board_init()
 
-
-async def rainbow(np, period=10):
-    def gamma(value, X=2.2):
-        return min(max(int(brightness * pow(value / 255.0, X) * 255.0 + 0.5), 0), 255)
-
-    brightness = 0.05
-    count = 0.0
-    leds = len(np)
-    while True:
-        for i in range(leds):
-            ofs = (count + i) % leds
-            np[i] = (gamma((sin(ofs / leds * 2 * pi) + 1) * 127),
-                     gamma((sin(ofs / leds * 2 * pi + 2/3*pi) + 1) * 127),
-                     gamma((sin(ofs / leds * 2 * pi + 4/3*pi) + 1) * 127))
-        count += 0.02 * leds
-        before = time.ticks_ms()
-        await np.async_write()
-        now = time.ticks_ms()
-        if before + 20 > now:
-            await asyncio.sleep_ms(20 - (now - before))
-
-
 # high prio for proc 1
 machine.mem32[0x40030000 + 0x00] = 0x10
 
 
-def setup_wifi():
+def setup_wifi(ssid='', passphrase='', security=network.WLAN.SEC_WPA_WPA2):
     network.hostname("TonberryPico")
+    if ssid is None or ssid == '':
+        apname = f"TonberryPicoAP_{machine.unique_id().hex()}"
+        print(f"Create AP {apname}")
         wlan = network.WLAN(network.WLAN.IF_AP)
-    wlan.config(ssid=f"TonberryPicoAP_{machine.unique_id().hex()}", security=wlan.SEC_OPEN)
+        wlan.config(ssid=apname, password=passphrase if passphrase is not None else '', security=security)
         wlan.active(True)
+    else:
+        print(f"Connect to SSID {ssid} with passphrase {passphrase}...")
+        wlan = network.WLAN()
+        wlan.active(True)
+        wlan.connect(ssid, passphrase if passphrase is not None else '', security=security)
 
+    # configure power management
+    wlan.config(pm=network.WLAN.PM_PERFORMANCE)
+
+    mac = ubinascii.hexlify(network.WLAN().config('mac'), ':').decode()
+    print(f"     mac: {mac}")
+    print(f" channel: {wlan.config('channel')}")
+    print(f"   essid: {wlan.config('essid')}")
+    print(f" txpower: {wlan.config('txpower')}")
+    print(f"ifconfig: {wlan.ifconfig()}")
+
+
+async def wdt_task(wdt):
+    # TODO: more checking of app health
+    # Right now this only protects against the asyncio executor crashing completely
+    while True:
+        await asyncio.sleep_ms(100)
+        wdt.feed()
 
 DB_PATH = '/sd/tonberry.db'
 
+config = Configuration()
+
+# Setup LEDs
+np = NeoPixel(hwconfig.LED_DIN, config.get_led_count(), sm=1)
+led_max = config.get_led_max()
+np.fill((led_max, led_max, 0))
+np.write()
+
 
 def run():
     asyncio.new_event_loop()
-    # Setup LEDs
-    np = NeoPixel(hwconfig.LED_DIN, hwconfig.LED_COUNT, sm=1)
-    asyncio.create_task(rainbow(np))
 
-    # Wifi with default config
-    setup_wifi()
+    if machine.Pin(hwconfig.BUTTONS[1], machine.Pin.IN, machine.Pin.PULL_UP).value() == 0:
+        np.fill((0, 0, led_max))
+        np.write()
+        # Force default access point
+        setup_wifi('', '', network.WLAN.SEC_OPEN)
+    else:
+        secstring = config.get_wifi_security()
+        security = network.WLAN.SEC_WPA_WPA2
+        if secstring == 'open':
+            security = network.WLAN.SEC_OPEN
+        elif secstring == 'wpa_wpa2':
+            security = network.WLAN.SEC_WPA_WPA2
+        elif secstring == 'wpa3':
+            security = network.WLAN.SEC_WPA3
+        elif secstring == 'wpa2_wpa3':
+            security = network.WLAN.SEC_WPA2_WPA3
+        setup_wifi(config.get_wifi_ssid(), config.get_wifi_passphrase(), security)
 
     # Setup MP3 player
     with SDContext(mosi=hwconfig.SD_DI, miso=hwconfig.SD_DO, sck=hwconfig.SD_SCK, ss=hwconfig.SD_CS,
                    baudrate=hwconfig.SD_CLOCKRATE):
-        # Temporary hack: build database from folders if no database exists
-        # Can be removed once playlists can be created via API
-        try:
-            _ = os.stat(DB_PATH)
-        except OSError as ex:
-            if ex.errno == ENOENT:
-                print("No playlist DB found, trying to build DB from tag dirs")
-                builddb()
-
         with BTreeFileManager(DB_PATH) as playlistdb, \
              AudioContext(hwconfig.I2S_DIN, hwconfig.I2S_DCLK, hwconfig.I2S_LRCLK) as audioctx:
 
@@ -98,37 +113,42 @@ 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),
-                                    playlistdb=lambda _: playlistdb)
+                                    buttons=lambda the_app: Buttons(the_app, config, hwconfig),
+                                    playlistdb=lambda _: playlistdb,
+                                    hwconfig=lambda _: hwconfig,
+                                    leds=lambda _: LedManager(np, config),
+                                    config=lambda _: config)
             the_app = app.PlayerApp(deps)
 
+            start_webserver(config, the_app)
             # Start
+            wdt = machine.WDT(timeout=2000)
             asyncio.create_task(aiorepl.task({'timer_manager': TimerManager(),
                                               'app': the_app}))
+            asyncio.create_task(wdt_task(wdt))
             asyncio.get_event_loop().run_forever()
 
 
-def builddb():
-    """
-    For testing, build a playlist db based on the previous tag directory format.
-    Can be removed once uploading files / playlist via the web api is possible.
-    """
-    try:
-        os.unlink(DB_PATH)
-    except OSError:
-        pass
-    with BTreeFileManager(DB_PATH) as db:
-        for name, type_, _, _ in os.ilistdir(b'/sd'):
-            if type_ != 0x4000:
-                continue
-            fl = [b'/sd/' + name + b'/' + x for x in os.listdir(b'/sd/' + name) if x.endswith(b'.mp3')]
-            db.createPlaylistForTag(name, fl)
-    os.sync()
+def error_blink():
+    while True:
+        if machine.Pin(hwconfig.BUTTONS[0], machine.Pin.IN, machine.Pin.PULL_UP).value() == 0:
+            machine.reset()
+        np.fill((led_max, 0, 0))
+        np.write()
+        time.sleep_ms(500)
+        np.fill((0, 0, 0))
+        np.write()
+        time.sleep_ms(500)
 
 
 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:
+        try:
             run()
+        except Exception as ex:
+            sys.print_exception(ex)
+            error_blink()
+    else:
+        np.fill((led_max, 0, 0))
+        np.write()

software/src/mfrc522_test.py

@@ -1,73 +0,0 @@
-from mfrc522 import MFRC522
-import asyncio
-import time
-
-delay_sum = 0
-delay_count = 0
-max_delay = 0
-
-
-async def latency_test():
-    global delay_sum
-    global delay_count
-    global max_delay
-    global min_delay
-    min_delay = 0xffffffff
-    await asyncio.sleep_ms(1)
-    while True:
-        for _ in range(2000):
-            before = time.ticks_us()
-            await asyncio.sleep(0)
-            after = time.ticks_us()
-            delay = after - before
-            delay_sum += delay
-            delay_count += 1
-            if delay > max_delay:
-                max_delay = delay
-            if delay < min_delay:
-                min_delay = delay
-            await asyncio.sleep_ms(1)
-        print(f"delay (min / max / avg) [µs]: ({min_delay} / {max_delay} / {delay/delay_sum})")
-
-
-def uid_to_string(uid: list):
-    return '0x' + ''.join(f'{i:02x}' for i in uid)
-
-
-async def get_tag_uid(reader: MFRC522, poll_interval_ms: int = 50) -> list:
-    '''
-    The maximum measured delay with poll_interval_ms=50 and a reader with tocard_retries=5 is
-    15.9 ms:
-        delay (min / max / avg) [µs]: (360 / 15945 / 1.892923e-06)
-
-    The maximum measured delay dropped to 11.6 ms by setting tocard_retries=1:
-        delay (min / max / avg) [µs]: (368 / 11696 / 6.204211e-06)
-    '''
-    while True:
-        reader.init()
-
-        # For now we omit the tag type
-        (stat, _) = reader.request(reader.REQIDL)
-        if stat == reader.OK:
-            (stat, uid) = reader.SelectTagSN()
-            if stat == reader.OK:
-                print(f"uid={uid_to_string(uid)}")
-
-        await asyncio.sleep_ms(poll_interval_ms)
-
-
-def main():
-    reader = MFRC522(spi_id=1, sck=10, miso=12, mosi=11, cs=13, rst=9, tocard_retries=1)
-
-    print("")
-    print("Please place card on reader")
-    print("")
-
-    asyncio.create_task(get_tag_uid(reader))
-    asyncio.create_task(latency_test())
-
-    asyncio.get_event_loop().run_forever()
-
-
-if __name__ == "__main__":
-    main()

software/src/microdot_test.py

@@ -1,38 +0,0 @@
-import rp2
-import network
-import ubinascii
-from microdot import Microdot
-
-rp2.country('DE')
-
-wlan = network.WLAN(network.AP_IF)
-wlan.config(ssid='TonberryPico', security=network.WLAN.SEC_OPEN)
-# Important: we cannot change the ip in station mode, otherwise dhcp won't work!
-# wlan.ipconfig(addr4='10.0.0.1')
-wlan.active(True)   # loads the firmware
-while wlan.active() is False:
-    pass
-wlan.config(pm=network.WLAN.PM_NONE)
-
-mac = ubinascii.hexlify(network.WLAN().config('mac'), ':').decode()
-print(f"     mac: {mac}")
-print(f" channel: {wlan.config('channel')}")
-print(f"   essid: {wlan.config('essid')}")
-print(f" txpower: {wlan.config('txpower')}")
-print(f"ifconfig: {wlan.ifconfig()}")
-
-app = Microdot()
-
-
-@app.route('/')
-async def index(request):
-    print("wohoo, a guest :)")
-    print(f"  app: {request.app}")
-    print(f"  client: {request.client_addr}")
-    print(f"  method: {request.method}")
-    print(f"  url: {request.url}")
-    print(f"  headers: {request.headers}")
-    print(f"  cookies: {request.cookies}")
-    return "TonberryPico says 'Hello World!'"
-
-app.run(port=80)

software/src/nfc/nfc.py

@@ -6,6 +6,7 @@ Copyright (c) 2025 Matthias Blankertz <matthias@blankertz.org>
 
 import asyncio
 import time
+from utils import safe_callback
 
 from mfrc522 import MFRC522
 try:
@@ -74,7 +75,7 @@ class Nfc:
                 self.last_uid = uid
                 self.last_uid_timestamp = time.ticks_us()
             if self.cb is not None and last_callback_uid != uid:
-                self.cb.onTagChange(uid)
+                safe_callback(lambda: self.cb.onTagChange(uid), "nfc tag change")
                 last_callback_uid = uid
 
             await asyncio.sleep_ms(poll_interval_ms)

software/src/test.py

@@ -1,113 +0,0 @@
-# SPDX-License-Identifier: MIT
-# Copyright (c) 2024-2025 Matthias Blankertz <matthias@blankertz.org>
-
-import aiorepl
-import asyncio
-import machine
-import micropython
-import os
-import time
-from machine import Pin
-from math import pi, sin, pow
-from micropython import const
-
-# Own modules
-from audiocore import Audiocore
-from mp3player import MP3Player
-from rp2_neopixel import NeoPixel
-from rp2_sd import SDCard
-
-micropython.alloc_emergency_exception_buf(100)
-
-leds = const(10)
-brightness = 0.5
-
-
-def gamma(value, X=2.2):
-    return min(max(int(brightness * pow(value / 255.0, X) * 255.0 + 0.5), 0), 255)
-
-
-async def rainbow(np, period=10):
-    count = 0.0
-    while True:
-        for i in range(leds):
-            ofs = (count + i) % leds
-            np[i] = (gamma((sin(ofs / leds * 2 * pi) + 1) * 127),
-                     gamma((sin(ofs / leds * 2 * pi + 2/3*pi) + 1) * 127),
-                     gamma((sin(ofs / leds * 2 * pi + 4/3*pi) + 1) * 127))
-        count += 0.2
-        before = time.ticks_ms()
-        await np.async_write()
-        now = time.ticks_ms()
-        if before + 20 > now:
-            await asyncio.sleep_ms(20 - (now - before))
-
-
-# Set 8 mA drive strength and fast slew rate
-machine.mem32[0x4001c004 + 6*4] = 0x67
-machine.mem32[0x4001c004 + 7*4] = 0x67
-machine.mem32[0x4001c004 + 8*4] = 0x67
-
-
-def list_sd():
-    try:
-        sd = SDCard(mosi=Pin(3), miso=Pin(4), sck=Pin(2), ss=Pin(5), baudrate=15000000)
-    except OSError:
-        for i in range(leds):
-            np[i] = (255, 0, 0)
-        np.write()
-        return
-    try:
-        os.mount(sd, '/sd')
-        print(os.listdir(b'/sd'))
-    except OSError as ex:
-        print(f"{ex}")
-
-
-delay_sum = 0
-delay_count = 0
-max_delay = 0
-
-
-async def latency_test():
-    global delay_sum
-    global delay_count
-    global max_delay
-    await asyncio.sleep_ms(1)
-    while True:
-        for _ in range(2000):
-            before = time.ticks_us()
-            await asyncio.sleep(0)
-            after = time.ticks_us()
-            delay = after - before
-            delay_sum += delay
-            delay_count += 1
-            if delay > max_delay:
-                max_delay = delay
-            await asyncio.sleep_ms(1)
-        print(f"Max delay {max_delay} us, average {delay/delay_sum} us")
-
-pin = Pin.board.GP16
-np = NeoPixel(pin, leds)
-
-# Test SD card
-list_sd()
-
-# Test NeoPixel
-asyncio.create_task(rainbow(np))
-
-# Test audio
-audioctx = Audiocore(Pin(8), Pin(6))
-
-player = MP3Player(audioctx)
-
-# high prio for proc 1
-machine.mem32[0x40030000 + 0x00] = 0x10
-
-testfiles = [b'/sd/' + name for name in os.listdir(b'/sd') if name.endswith(b'mp3')]
-player.set_playlist(testfiles)
-asyncio.create_task(player.task())
-
-
-asyncio.create_task(aiorepl.task({'player': player}))
-asyncio.get_event_loop().run_forever()

software/src/tonberry.schema.json

@@ -0,0 +1,53 @@
+{
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "definitions": {
+        "PlaybackPosition": {
+            "type": "object",
+            "properties": {
+                "position_seconds": { "type": "number" },
+                "device_uptime": { "type": "number" }
+            },
+            "required": ["position_seconds"]
+        },
+        "AudioFile": {
+            "type": "object",
+            "properties": {
+                "id": { "type": "string", "format": "uuid" },
+                "filename": { "type": "string" },
+                "size_bytes": { "type": "integer" },
+                "duration_seconds": { "type": "number" },
+                "last_played_uptime": { "type": "number" },
+                "playback_position": { "$ref": "#/definitions/PlaybackPosition" }
+            },
+            "required": ["id", "filename"]
+        },
+        "Playlist": {
+            "type": "object",
+            "properties": {
+                "id": { "type": "string", "format": "uuid" },
+                "name": { "type": "string" },
+                "audio_files": {
+                    "type": "array",
+                    "items": { "$ref": "#/definitions/AudioFile" }
+                },
+                "current_track_index": { "type": "integer", "minimum": 0 },
+                "last_played_uptime": { "type": "number" },
+                "playback_position": { "$ref": "#/definitions/PlaybackPosition" }
+            },
+            "required": ["id", "name", "audio_files"]
+        },
+        "NfcTag": {
+            "type": "object",
+            "properties": {
+                "uid": { "type": "string" },
+                "name": { "type": "string" },
+                "linked_type": {
+                    "type": "string",
+                    "enum": ["audio_file", "playlist"]
+                },
+                "linked_id": { "type": "string", "format": "uuid" }
+            },
+            "required": ["uid", "linked_type", "linked_id"]
+        }
+    }
+}

software/src/utils/__init__.py

@@ -1,11 +1,15 @@
 # SPDX-License-Identifier: MIT
 # Copyright (c) 2025 Matthias Blankertz <matthias@blankertz.org>
 
+from utils.helpers import safe_callback
+from utils.timer import TimerManager
 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
 from utils.playlistdb import BTreeDB, BTreeFileManager
 from utils.sdcontext import SDContext
-from utils.timer import TimerManager
 
-__all__ = ["BTreeDB", "BTreeFileManager", "Buttons", "get_pin_index", "MBRPartition", "SDContext", "TimerManager"]
+__all__ = ["BTreeDB", "BTreeFileManager", "Buttons", "Configuration", "get_pin_index", "LedManager", "MBRPartition",
+           "safe_callback", "SDContext", "TimerManager"]

software/src/utils/buttons.py

@@ -5,6 +5,7 @@ import asyncio
 import machine
 import micropython
 import time
+from utils import safe_callback, TimerManager
 try:
     from typing import TYPE_CHECKING  # type: ignore
 except ImportError:
@@ -17,21 +18,47 @@ 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.hwconfig = hwconfig
         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] = {}
+        self.timer_manager = TimerManager()
         for button in self.buttons.keys():
             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)
@@ -44,10 +71,20 @@ class Buttons:
             # print(f'B{keycode} {now}')
             self.pressed.append(keycode)
             self.int_flag.set()
+            if keycode == self.VOLDOWN:
+                self.timer_manager.schedule(time.ticks_ms() + 5000, self.long_press_shutdown)
+        if button.value() == 1 and keycode == self.VOLDOWN:
+            self.timer_manager.cancel(self.long_press_shutdown)
 
     async def task(self):
         while True:
             await self.int_flag.wait()
             while len(self.pressed) > 0:
                 what = self.pressed.pop()
-                self.cb.onButtonPressed(what)
+                safe_callback(lambda: self.cb.onButtonPressed(what), "button callback")
+
+    def long_press_shutdown(self):
+        if self.hwconfig.get_on_battery():
+            self.hwconfig.power_off()
+        else:
+            machine.reset()

software/src/utils/config.py

@@ -0,0 +1,138 @@
+# 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,
+        },
+        'TAGMODE': 'tagremains',
+        'WIFI': {
+            'SSID': '',
+            'PASSPHRASE': '',
+            'SECURITY': 'wpa_wpa2',
+        },
+        'VOLUME_MAX': 255,
+        'VOLUME_BOOT': 16,
+        'LED_MAX': 255,
+    }
+
+    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)
+            self._merge_configs(self.DEFAULT_CONFIG, self.config)
+        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 _merge_configs(self, default, config):
+        for k in default.keys():
+            if k not in config:
+                if isinstance(default[k], dict):
+                    config[k] = default[k].copy()
+                else:
+                    config[k] = default[k]
+            elif isinstance(default[k], dict):
+                self._merge_configs(default[k], config[k])
+
+    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[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')
+
+    def get_tagmode(self) -> str:
+        return self._get('TAGMODE')
+
+    def get_wifi_ssid(self) -> str:
+        return self._get('WIFI')['SSID']
+
+    def get_wifi_passphrase(self) -> str:
+        return self._get('WIFI')['PASSPHRASE']
+
+    def get_wifi_security(self) -> str:
+        return self._get('WIFI')['SECURITY']
+
+    def get_volume_max(self) -> int:
+        return self._get('VOLUME_MAX')
+
+    def get_led_max(self) -> int:
+        return self._get('LED_MAX')
+
+    def get_volume_boot(self) -> int:
+        return self._get('VOLUME_BOOT')
+
+    # 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)
+        if 'TAGMODE' in config and config['TAGMODE'] not in ['tagremains', 'tagstartstop']:
+            raise ValueError("Invalid TAGMODE: Must be 'tagremains' or 'tagstartstop'")
+        if 'WLAN' in config and 'SECURITY' in config['WLAN'] and \
+           config['WLAN']['SECURITY'] not in ['open', 'wpa_wpa2', 'wpa3', 'wpa2_wpa3']:
+            raise ValueError("Invalid WLAN SECURITY: Must be 'open', 'wpa_wpa2', 'wpa3' or 'wpa2_wpa3'")
+        self._merge_configs(self.config, config)
+        self.config = config
+        self._save()

software/src/utils/helpers.py

@@ -0,0 +1,12 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2025 Matthias Blankertz <matthias@blankertz.org>
+
+import sys
+
+
+def safe_callback(func, name="callback"):
+    try:
+        func()
+    except Exception as ex:
+        print(f"Uncaught exception in {name}")
+        sys.print_exception(ex)

software/src/utils/leds.py

@@ -0,0 +1,61 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2025 Matthias Blankertz <matthias@blankertz.org>
+
+import asyncio
+from math import sin, pi
+from micropython import const
+import time
+
+
+class LedManager:
+    IDLE = const(0)
+    PLAYING = const(1)
+    REBOOTING = const(2)
+
+    def __init__(self, np, config):
+        self.led_state = LedManager.IDLE
+        self.brightness = config.get_led_max() / 255
+        self.np = np
+        self.leds = len(self.np)
+        asyncio.create_task(self.run())
+
+    def set_state(self, state):
+        assert state in [LedManager.IDLE, LedManager.PLAYING, LedManager.REBOOTING]
+        self.led_state = state
+
+    def _gamma(self, value, X=2.2):
+        result = min(max(int(self.brightness * pow(value / 255.0, X) * 255.0 + 0.5), 0), 255)
+        if value > 0:
+            result = max(1, result)
+        return result
+
+    def _rainbow(self, time):
+        for i in range(self.leds):
+            ofs = (time * self.leds + i) % self.leds
+            self.np[i] = (self._gamma((sin(ofs / self.leds * 2 * pi) + 1) * 127),
+                          self._gamma((sin(ofs / self.leds * 2 * pi + 2/3*pi) + 1) * 127),
+                          self._gamma((sin(ofs / self.leds * 2 * pi + 4/3*pi) + 1) * 127))
+
+    def _pulse(self, time, color, speed):
+        scaled_sin = max(1, abs(sin(time / speed * 2 * pi)) * 255)
+        val = (self._gamma(color[0]*scaled_sin),
+               self._gamma(color[1]*scaled_sin),
+               self._gamma(color[2]*scaled_sin))
+        for i in range(self.leds):
+            self.np[i] = val
+
+    async def run(self):
+        time_ = 0.0
+        while True:
+            if self.led_state == LedManager.IDLE:
+                self._pulse(time_, (0, 1, 0), 3)
+            elif self.led_state == LedManager.PLAYING:
+                self._rainbow(time_)
+            elif self.led_state == LedManager.REBOOTING:
+                self._pulse(time_, (1, 0, 1), 0.2)
+            time_ += 0.02
+            before = time.ticks_ms()
+            await self.np.async_write()
+            now = time.ticks_ms()
+            if before + 20 > now:
+                await asyncio.sleep_ms(20 - (now - before))

software/src/utils/playlistdb.py

@@ -31,17 +31,15 @@ class BTreeDB(IPlaylistDB):
     PERSIST_NO = b'no'
     PERSIST_TRACK = b'track'
     PERSIST_OFFSET = b'offset'
-    DEFAULT_SETTINGS = {
-        b'tagmode': b'tagremains'
-    }
 
     class Playlist(IPlaylist):
-        def __init__(self, parent: "BTreeDB", tag: bytes, pos: int, persist, shuffle):
+        def __init__(self, parent: "BTreeDB", tag: bytes, pos: int, persist, shuffle, name):
             self.parent = parent
             self.tag = tag
             self.pos = pos
             self.persist = persist
             self.shuffle = shuffle
+            self.name = name
             self.length = self.parent._getPlaylistLength(self.tag)
             self._shuffle()
 
@@ -107,6 +105,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
@@ -160,6 +169,10 @@ class BTreeDB(IPlaylistDB):
         return (b''.join([tag, b'/playlist/']),
                 b''.join([tag, b'/playlist0']))
 
+    @staticmethod
+    def _keyPlaylistName(tag):
+        return b''.join([tag, b'/playlistname'])
+
     def _flush(self):
         """
         Flush the database and call the flush_func if it was provided.
@@ -209,18 +222,18 @@ class BTreeDB(IPlaylistDB):
             if len(elements) >= 2 and elements[1] == b'playlist':
                 last = k
                 break
-        print(last)
         elements = last.split(b'/')
         if len(elements) != 3:
             raise RuntimeError("Malformed playlist key")
         return int(elements[2])+1
 
-    def _savePlaylist(self, tag, entries, persist, shuffle, flush=True):
+    def _savePlaylist(self, tag, entries, persist, shuffle, name, flush=True):
         self._deletePlaylist(tag, False)
         for idx, entry in enumerate(entries):
             self.db[self._keyPlaylistEntry(tag, idx)] = entry
         self.db[self._keyPlaylistPersist(tag)] = persist
         self.db[self._keyPlaylistShuffle(tag)] = shuffle
+        self.db[self._keyPlaylistName(tag)] = name.encode()
         if flush:
             self._flush()
 
@@ -233,7 +246,7 @@ class BTreeDB(IPlaylistDB):
                 pass
         for k in (self._keyPlaylistPos(tag), self._keyPlaylistPosOffset(tag),
                   self._keyPlaylistPersist(tag), self._keyPlaylistShuffle(tag),
-                  self._keyPlaylistShuffleSeed(tag)):
+                  self._keyPlaylistShuffleSeed(tag), self._keyPlaylistName(tag)):
             try:
                 del self.db[k]
             except KeyError:
@@ -241,6 +254,19 @@ class BTreeDB(IPlaylistDB):
         if flush:
             self._flush()
 
+    def getPlaylists(self):
+        """
+        Get a list of all defined playlists with their tag and names.
+        """
+        playlist_tags = set()
+        for item in self.db:
+            playlist_tags.add(item.split(b'/')[0])
+        playlists = []
+        for tag in playlist_tags:
+            name = self.db.get(self._keyPlaylistName(tag), b'').decode()
+            playlists.append({'tag': tag, 'name': name})
+        return playlists
+
     def getPlaylistForTag(self, tag: bytes):
         """
         Lookup the playlist for 'tag' and return the Playlist object. Return None if no playlist exists for the given
@@ -258,24 +284,23 @@ class BTreeDB(IPlaylistDB):
             return None
         if self._keyPlaylistEntry(tag, pos) not in self.db:
             pos = 0
+        name = self.db.get(self._keyPlaylistName(tag), b'').decode()
         shuffle = self.db.get(self._keyPlaylistShuffle(tag), self.SHUFFLE_NO)
-        return self.Playlist(self, tag, pos, persist, shuffle)
+        return self.Playlist(self, tag, pos, persist, shuffle, name)
 
     def createPlaylistForTag(self, tag: bytes, entries: typing.Iterable[bytes], persist=PERSIST_TRACK,
-                             shuffle=SHUFFLE_NO):
+                             shuffle=SHUFFLE_NO, name: str = ''):
         """
         Create and save a playlist for 'tag' and return the Playlist object. If a playlist already existed for 'tag' it
         is overwritten.
         """
         assert persist in (self.PERSIST_NO, self.PERSIST_TRACK, self.PERSIST_OFFSET)
         assert shuffle in (self.SHUFFLE_NO, self.SHUFFLE_YES)
-        self._savePlaylist(tag, entries, persist, shuffle)
+        self._savePlaylist(tag, entries, persist, shuffle, name)
         return self.getPlaylistForTag(tag)
 
-    def getSetting(self, key: bytes | str) -> str:
-        if key is str:
-            key = key.encode()
-        return self.db.get(b'settings/' + key, self.DEFAULT_SETTINGS[key]).decode()
+    def deletePlaylistForTag(self, tag: bytes):
+        self._deletePlaylist(tag)
 
     def validate(self, dump=False):
         """
@@ -296,8 +321,7 @@ class BTreeDB(IPlaylistDB):
                 fail(f'Malformed key {k!r}')
                 continue
             if fields[0] == b'settings':
-                val = self.db[k].decode()
-                print(f'Setting {fields[1].decode()} = {val}')
+                # Legacy, not used any more
                 continue
             if last_tag != fields[0]:
                 last_tag = fields[0]
@@ -356,6 +380,14 @@ class BTreeDB(IPlaylistDB):
                     _ = int(val)
                 except ValueError:
                     fail(f' Bad playlistposoffset value for {last_tag}: {val!r}')
+            elif fields[1] == b'playlistname':
+                val = self.db[k]
+                try:
+                    name = val.decode()
+                    if dump:
+                        print(f'\tName: {name}')
+                except UnicodeError:
+                    fail(f' Bad playlistname for {last_tag}: Not valid unicode')
             else:
                 fail(f'Unknown key {k!r}')
         return result

software/src/utils/timer.py

@@ -4,6 +4,7 @@
 import asyncio
 import heapq
 import time
+from utils import safe_callback
 
 TIMER_DEBUG = True
 
@@ -22,6 +23,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,41 +33,53 @@ class TimerManager(object):
             self.worker_event.set()
 
     def cancel(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:
+        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
 
-    async def _timer_worker(self):
-        while True:
-            if len(self.timers) == 0:
-                # Nothing to do
-                await self.worker_event.wait()
-                if self.timer_debug:
-                    print("_timer_worker: event 0")
-                self.worker_event.clear()
-                continue
-            cur_nearest = self.timers[0][0]
-            wait_time = cur_nearest - time.ticks_ms()
-            if wait_time > 0:
-                if self.timer_debug:
-                    print(f"_timer_worker: next is {self.timers[0]}, sleep {wait_time} ms")
+    def _remove_timer(self, what):
         try:
-                    await asyncio.wait_for_ms(self.worker_event.wait(), wait_time)
+            (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)
+        return i
+
+    def _next_timeout(self):
+        if len(self.timers) == 0:
             if self.timer_debug:
-                        print("_timer_worker: event 1")
+                print("timer: worker: queue empty")
+            return None
+        cur_nearest = self.timers[0][0]
+        next_timeout = cur_nearest - time.ticks_ms()
+        if self.timer_debug:
+            if next_timeout > 0:
+                print(f"timer: worker: next is {self.timers[0]}, sleep {next_timeout} ms")
+            else:
+                print(f"timer: worker: {self.timers[0]} elapsed @{cur_nearest}, delay {-next_timeout} ms")
+        return next_timeout
+
+    async def _wait(self, timeout):
+        try:
+            await asyncio.wait_for_ms(self.worker_event.wait(), timeout)
+            if self.timer_debug:
+                print("timer: worker: event")
             # got woken up due to event
             self.worker_event.clear()
-                    continue
+            return True
         except asyncio.TimeoutError:
-                    pass
+            return False
+
+    async def _timer_worker(self):
+        while True:
+            next_timeout = self._next_timeout()
+            if next_timeout is None or next_timeout > 0:
+                await self._wait(next_timeout)
+            else:
                 _, callback = heapq.heappop(self.timers)
-            callback()
+                safe_callback(callback, "timer callback")

software/src/webserver.py

@@ -0,0 +1,299 @@
+'''
+SPDX-License-Identifier: MIT
+Copyright (c) 2024-2025 Stefan Kratochwil <Kratochwil-LA@gmx.de>
+'''
+
+import asyncio
+import board
+import errno
+import hwconfig
+import json
+import machine
+import network
+import os
+import time
+import ubinascii
+
+from array import array
+from microdot import Microdot, redirect, send_file, Request
+from utils import TimerManager, LedManager
+
+webapp = Microdot()
+server = None
+config = None
+app = None
+nfc = None
+playlist_db = None
+leds = None
+timer_manager = None
+
+Request.max_content_length = 128 * 1024 * 1024  # 128MB requests allowed
+
+
+def start_webserver(config_, app_):
+    global server, config, app, nfc, playlist_db, leds, timer_manager
+    server = asyncio.create_task(webapp.start_server(host='::0', port=80))
+    config = config_
+    app = app_
+    nfc = app.get_nfc()
+    playlist_db = app.get_playlist_db()
+    leds = app.get_leds()
+    timer_manager = TimerManager()
+
+
+@webapp.before_request
+async def before_request_handler(request):
+    if request.method in ['PUT', 'POST', 'DELETE'] and app.is_playing():
+        return "Cannot write to device while playback is active", 503
+    app.reset_idle_timeout()
+
+
+@webapp.route('/api/v1/hello')
+async def index(request):
+    print("wohoo, a guest :)")
+    print(f"  app: {request.app}")
+    print(f"  client: {request.client_addr}")
+    print(f"  method: {request.method}")
+    print(f"  url: {request.url}")
+    print(f"  headers: {request.headers}")
+    print(f"  cookies: {request.cookies}")
+    return "TonberryPico says 'Hello World!'"
+
+
+@webapp.route('/api/v1/filesystem', methods=['POST'])
+async def filesystem_post(request):
+    # curl -X POST -d "burp" http://192.168.4.1/api/v1/filesystem
+    print(request)
+    return {'success': False}
+
+
+@webapp.route('/api/v1/playlist', methods=['POST'])
+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
+
+
+@webapp.route('/api/v1/last_tag_uid', methods=['GET'])
+async def last_tag_uid_get(request):
+    tag, _ = nfc.get_last_uid()
+    return {'tag': tag}
+
+
+@webapp.route('/', methods=['GET'])
+async def root_get(request):
+    return redirect('/index.html')
+
+
+@webapp.route('/index.html', methods=['GET'])
+async def index_get(request):
+    return send_file('/frontend/index.html.gz', content_type='text/html', compressed='gzip')
+
+
+@webapp.route('/static/<path:path>', methods=['GET'])
+async def static(request, path):
+    if '..' in path:
+        # directory traversal is not allowed
+        return 'Not found', 404
+    return send_file('/frontend/static/' + path, max_age=86400)
+
+
+@webapp.route('/api/v1/playlists', methods=['GET'])
+async def playlists_get(request):
+    return playlist_db.getPlaylists()
+
+
+def is_hex(s):
+    hex_chars = '0123456789abcdef'
+    return all(c in hex_chars for c in s)
+
+
+fsroot = b'/sd'
+
+
+@webapp.route('/api/v1/playlist/<tag>', methods=['GET'])
+async def playlist_get(request, tag):
+    if not is_hex(tag):
+        return 'invalid tag', 400
+
+    playlist = playlist_db.getPlaylistForTag(tag.encode())
+    if playlist is None:
+        return None, 404
+
+    return {
+            'shuffle': playlist.shuffle,
+            'persist': playlist.persist,
+            'paths': [(p[len(fsroot):] if p.startswith(fsroot) else p).decode()
+                      for p in playlist.getPaths()],
+            'name': playlist.name
+    }
+
+
+@webapp.route('/api/v1/playlist/<tag>', methods=['PUT'])
+async def playlist_put(request, tag):
+    if not is_hex(tag):
+        return 'invalid tag', 400
+
+    playlist = request.json
+    if 'persist' in playlist and \
+       playlist['persist'] not in ['no', 'track', 'offset']:
+        return "Invalid 'persist' setting", 400
+    if 'shuffle' in playlist and \
+       playlist['shuffle'] not in ['no', 'yes']:
+        return "Invalid 'shuffle' setting", 400
+
+    playlist_db.createPlaylistForTag(tag.encode(),
+                                     (fsroot + path.encode() for path in playlist.get('paths', [])),
+                                     playlist.get('persist', 'track').encode(),
+                                     playlist.get('shuffle', 'no').encode(),
+                                     playlist.get('name', ''))
+    return '', 204
+
+
+@webapp.route('/api/v1/playlist/<tag>', methods=['DELETE'])
+async def playlist_delete(request, tag):
+    if not is_hex(tag):
+        return 'invalid tag', 400
+    playlist_db.deletePlaylistForTag(tag.encode())
+    return '', 204
+
+
+@webapp.route('/api/v1/audiofiles', methods=['GET'])
+async def audiofiles_get(request):
+    def directory_iterator():
+        yield '['
+        first = True
+
+        def make_json_str(obj):
+            nonlocal first
+            jsonpath = json.dumps(obj)
+            if not first:
+                jsonpath = ',' + jsonpath
+            first = False
+            return jsonpath
+
+        dirstack = [fsroot]
+        while dirstack:
+            current_dir = dirstack.pop()
+            for entry in os.ilistdir(current_dir):
+                name = entry[0]
+                type_ = entry[1]
+                current_path = current_dir + b'/' + name
+                if type_ == 0x4000:
+                    yield make_json_str({'name': current_path[len(fsroot):], 'type': 'directory'})
+                    dirstack.append(current_path)
+                elif type_ == 0x8000:
+                    if name.lower().endswith('.mp3'):
+                        yield make_json_str({'name': current_path[len(fsroot):], 'type': 'file'})
+        yield ']'
+
+    return directory_iterator(), {'Content-Type': 'application/json; charset=UTF-8'}
+
+
+async def stream_to_file(stream, file_, length):
+    data = array('b', range(4096))
+    bytes_copied = 0
+    while True:
+        bytes_read = await stream.readinto(data)
+        if bytes_read == 0:
+            # End of body
+            break
+        bytes_written = file_.write(data[:bytes_read])
+        if bytes_written != bytes_read:
+            # short writes shouldn't happen
+            raise OSError(errno.EIO, 'unexpected short write')
+        bytes_copied += bytes_written
+        if bytes_copied == length:
+            break
+        app.reset_idle_timeout()
+    return bytes_copied
+
+
+@webapp.route('/api/v1/audiofiles', methods=['POST'])
+async def audiofile_upload(request):
+    if 'type' not in request.args or request.args['type'] not in ['file', 'directory']:
+        return 'invalid or missing type', 400
+    if 'location' not in request.args:
+        return 'missing location', 400
+    path = fsroot + '/' + request.args['location']
+    type_ = request.args['type']
+    length = request.content_length
+    print(f'Got upload request of type {type_} to {path} with length {length}')
+    if type_ == 'directory':
+        if length != 0:
+            return 'directory request may not have content', 400
+        try:
+            os.mkdir(path)
+        except OSError as ex:
+            return f'error creating directory: {ex}', 500
+        return '', 204
+    try:
+        with open(path, 'wb') as newfile:
+            if length > Request.max_body_length:
+                bytes_copied = await stream_to_file(request.stream, newfile, length)
+            else:
+                bytes_copied = newfile.write(request.body)
+    except OSError as ex:
+        return f'error writing data to file: {ex}', 500
+    if bytes_copied == length:
+        return '', 204
+    else:
+        return 'size mismatch', 500
+
+
+def recursive_delete(path):
+    stat = os.stat(path)
+    if stat[0] == 0x8000:
+        os.remove(path)
+    elif stat[0] == 0x4000:
+        for entry in os.ilistdir(path):
+            entry_path = path + '/' + entry[0]
+            recursive_delete(entry_path)
+        os.rmdir(path)
+
+
+@webapp.route('/api/v1/audiofiles', methods=['DELETE'])
+async def audiofile_delete(request):
+    if 'location' not in request.args:
+        return 'missing location', 400
+    location = request.args['location']
+    if '..' in location or len(location) == 0:
+        return 'bad location', 400
+    path = fsroot + '/' + request.args['location']
+    recursive_delete(path)
+    return '', 204
+
+
+@webapp.route('/api/v1/reboot/<method>', methods=['POST'])
+async def reboot(request, method):
+    if method == 'bootloader':
+        if hwconfig.get_on_battery():
+            return 'not possible: connect USB first', 403
+        leds.set_state(LedManager.REBOOTING)
+        timer_manager.schedule(time.ticks_ms() + 1500, machine.bootloader)
+    elif method == 'application':
+        leds.set_state(LedManager.REBOOTING)
+        timer_manager.schedule(time.ticks_ms() + 1500, machine.reset)
+    else:
+        return 'method not supported', 400
+    return '', 204
+
+
+@webapp.route('/api/v1/info', methods=['GET'])
+async def get_info(request):
+    mac = ubinascii.hexlify(network.WLAN().config('mac'), ':').decode()
+    return {'version': board.version,
+            'mac': mac}

software/tests/mocks/micropython.py

@@ -1,2 +1,5 @@
 # SPDX-License-Identifier: MIT
 # Copyright (c) 2025 Matthias Blankertz <matthias@blankertz.org>
+
+def const(x):
+    return x

software/tests/test_playerapp.py

@@ -43,10 +43,20 @@ class FakeMp3Player:
 
 
 class FakeTimerManager:
-    def __init__(self): pass
-    def cancel(self, timer): pass
+    def __init__(self):
+        self.queued = []
+
+    def cancel(self, timer):
+        self.queued = [(elem[0], elem[1], True) if elem[1] == timer else elem for elem in self.queued]
 
     def schedule(self, when, what):
+        self.queued.append((when, what, False))
+
+    def testing_run_queued(self):
+        queued = self.queued
+        self.queued = []
+        for when, what, canceled in queued:
+            if not canceled:
                 what()
 
 
@@ -91,32 +101,85 @@ class FakePlaylistDb:
         return None
 
 
+class FakeHwconfig:
+    def __init__(self):
+        self.powered = True
+        self.on_battery = False
+
+    def power_off(self):
+        self.powered = False
+
+    def get_on_battery(self):
+        return self.on_battery
+
+
+class FakeLeds:
+    IDLE = 0
+    PLAYING = 1
+
+    def __init__(self):
+        self.state = None
+
+    def set_state(self, state):
+        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 get_tagmode(self):
+        return 'tagremains'
+
+    def get_volume_max(self):
+        return 255
+
+    def get_volume_boot(self):
+        return 16
+
+
 def fake_open(filename, mode):
     return FakeFile(filename, mode)
 
 
 @pytest.fixture
 def faketimermanager(monkeypatch):
-    monkeypatch.setattr(utils.timer.TimerManager, '_instance', FakeTimerManager())
+    fake_timer_manager = FakeTimerManager()
+    monkeypatch.setattr(utils.timer.TimerManager, '_instance', fake_timer_manager)
+    yield fake_timer_manager
+
+
+def _makedeps(mp3player=FakeMp3Player, nfcreader=FakeNfcReader, buttons=FakeButtons,
+              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,
+                            config=lambda _: config() if callable(config) else config)
 
 
 def test_construct_app(micropythonify, faketimermanager):
     fake_mp3 = FakeMp3Player()
-    deps = app.Dependencies(mp3player=lambda _: fake_mp3,
-                            nfcreader=lambda _: FakeNfcReader(),
-                            buttons=lambda _: FakeButtons(),
-                            playlistdb=lambda _: FakePlaylistDb())
-    _ = app.PlayerApp(deps)
-    assert fake_mp3.volume is not None
+    deps = _makedeps(mp3player=fake_mp3)
+    dut = app.PlayerApp(deps)
+    fake_mp3 = dut.player
+    assert fake_mp3.volume is not None and fake_mp3.volume >= 16
 
 
 def test_load_playlist_on_tag(micropythonify, faketimermanager, monkeypatch):
     fake_db = FakePlaylistDb()
     fake_mp3 = FakeMp3Player()
-    deps = app.Dependencies(mp3player=lambda _: fake_mp3,
-                            nfcreader=lambda x: FakeNfcReader(x),
-                            buttons=lambda _: FakeButtons(),
-                            playlistdb=lambda _: fake_db)
+    deps = _makedeps(mp3player=fake_mp3, playlistdb=fake_db)
     app.PlayerApp(deps)
     with monkeypatch.context() as m:
         m.setattr(builtins, 'open', fake_open)
@@ -130,10 +193,7 @@ def test_load_playlist_on_tag(micropythonify, faketimermanager, monkeypatch):
 def test_playlist_seq(micropythonify, faketimermanager, monkeypatch):
     fake_db = FakePlaylistDb([b'track1.mp3', b'track2.mp3', b'track3.mp3'])
     fake_mp3 = FakeMp3Player()
-    deps = app.Dependencies(mp3player=lambda _: fake_mp3,
-                            nfcreader=lambda x: FakeNfcReader(x),
-                            buttons=lambda _: FakeButtons(),
-                            playlistdb=lambda _: fake_db)
+    deps = _makedeps(mp3player=fake_mp3, playlistdb=fake_db)
     dut = app.PlayerApp(deps)
     with monkeypatch.context() as m:
         m.setattr(builtins, 'open', fake_open)
@@ -166,10 +226,7 @@ def test_playlist_unknown_tag(micropythonify, faketimermanager, monkeypatch):
 
     fake_db = FakeNoPlaylistDb()
     fake_mp3 = FakeMp3Player()
-    deps = app.Dependencies(mp3player=lambda _: fake_mp3,
-                            nfcreader=lambda x: FakeNfcReader(x),
-                            buttons=lambda _: FakeButtons(),
-                            playlistdb=lambda _: fake_db)
+    deps = _makedeps(mp3player=fake_mp3, playlistdb=fake_db)
     app.PlayerApp(deps)
     with monkeypatch.context() as m:
         m.setattr(builtins, 'open', fake_open)
@@ -178,21 +235,16 @@ def test_playlist_unknown_tag(micropythonify, faketimermanager, monkeypatch):
 
 
 def test_tagmode_startstop(micropythonify, faketimermanager, monkeypatch):
-    class MyFakePlaylistDb(FakePlaylistDb):
-        def __init__(self, tracklist=[b'test/path.mp3']):
-            super().__init__(tracklist)
+    class FakeStartStopConfig(FakeConfig):
+        def __init__(self):
+            super().__init__()
 
-        def getSetting(self, key: bytes | str):
-            if key == 'tagmode':
+        def get_tagmode(self):
             return 'tagstartstop'
-            return None
 
-    fake_db = MyFakePlaylistDb()
+    fake_db = FakePlaylistDb([b'test/path.mp3'])
     fake_mp3 = FakeMp3Player()
-    deps = app.Dependencies(mp3player=lambda _: fake_mp3,
-                            nfcreader=lambda x: FakeNfcReader(x),
-                            buttons=lambda _: FakeButtons(),
-                            playlistdb=lambda _: fake_db)
+    deps = _makedeps(mp3player=fake_mp3, playlistdb=fake_db, config=FakeStartStopConfig)
     app.PlayerApp(deps)
     with monkeypatch.context() as m:
         m.setattr(builtins, 'open', fake_open)
@@ -202,6 +254,7 @@ def test_tagmode_startstop(micropythonify, faketimermanager, monkeypatch):
         assert fake_mp3.track.filename == b'test/path.mp3'
         # Removing tag should not stop playback
         FakeNfcReader.tag_callback.onTagChange(None)
+        faketimermanager.testing_run_queued()
         assert fake_mp3.track is not None
         assert fake_mp3.track.filename == b'test/path.mp3'
         # Presenting tag should stop playback
@@ -209,6 +262,7 @@ def test_tagmode_startstop(micropythonify, faketimermanager, monkeypatch):
         assert fake_mp3.track is None
         # Nothing should change here
         FakeNfcReader.tag_callback.onTagChange(None)
+        faketimermanager.testing_run_queued()
         assert fake_mp3.track is None
         # Presenting tag again should start playback again
         FakeNfcReader.tag_callback.onTagChange([23, 42, 1, 2, 3])
@@ -217,21 +271,9 @@ def test_tagmode_startstop(micropythonify, faketimermanager, monkeypatch):
 
 
 def test_tagmode_remains(micropythonify, faketimermanager, monkeypatch):
-    class MyFakePlaylistDb(FakePlaylistDb):
-        def __init__(self, tracklist=[b'test/path.mp3']):
-            super().__init__(tracklist)
-
-        def getSetting(self, key: bytes | str):
-            if key == 'tagmode':
-                return 'tagremains'
-            return None
-
-    fake_db = MyFakePlaylistDb()
+    fake_db = FakePlaylistDb([b'test/path.mp3'])
     fake_mp3 = FakeMp3Player()
-    deps = app.Dependencies(mp3player=lambda _: fake_mp3,
-                            nfcreader=lambda x: FakeNfcReader(x),
-                            buttons=lambda _: FakeButtons(),
-                            playlistdb=lambda _: fake_db)
+    deps = _makedeps(mp3player=fake_mp3, playlistdb=fake_db)
     app.PlayerApp(deps)
     with monkeypatch.context() as m:
         m.setattr(builtins, 'open', fake_open)
@@ -241,8 +283,52 @@ def test_tagmode_remains(micropythonify, faketimermanager, monkeypatch):
         assert fake_mp3.track.filename == b'test/path.mp3'
         # Remove tag to stop playback
         FakeNfcReader.tag_callback.onTagChange(None)
+        faketimermanager.testing_run_queued()
         assert fake_mp3.track is None
         # Presenting tag again should start playback again
         FakeNfcReader.tag_callback.onTagChange([23, 42, 1, 2, 3])
         assert fake_mp3.track is not None
         assert fake_mp3.track.filename == b'test/path.mp3'
+
+
+def test_led_state(micropythonify, faketimermanager, monkeypatch):
+    fake_leds = FakeLeds()
+    deps = _makedeps(leds=fake_leds)
+    app.PlayerApp(deps)
+    assert fake_leds.state == FakeLeds.IDLE
+    with monkeypatch.context() as m:
+        m.setattr(builtins, 'open', fake_open)
+        FakeNfcReader.tag_callback.onTagChange([23, 42, 1, 2, 3])
+        assert fake_leds.state == FakeLeds.PLAYING
+        FakeNfcReader.tag_callback.onTagChange(None)
+        faketimermanager.testing_run_queued()
+        assert fake_leds.state == FakeLeds.IDLE
+
+
+def test_idle_shutdown_after_start(micropythonify, faketimermanager, monkeypatch):
+    fake_hwconfig = FakeHwconfig()
+    fake_hwconfig.on_battery = True
+    deps = _makedeps(hwconfig=fake_hwconfig)
+    app.PlayerApp(deps)
+    assert fake_hwconfig.powered
+    faketimermanager.testing_run_queued()
+    assert not fake_hwconfig.powered
+
+
+def test_idle_shutdown_after_playback(micropythonify, faketimermanager, monkeypatch):
+    fake_hwconfig = FakeHwconfig()
+    fake_hwconfig.on_battery = True
+    deps = _makedeps(hwconfig=fake_hwconfig)
+    app.PlayerApp(deps)
+    assert fake_hwconfig.powered
+    with monkeypatch.context() as m:
+        m.setattr(builtins, 'open', fake_open)
+        FakeNfcReader.tag_callback.onTagChange([23, 42, 1, 2, 3])
+        faketimermanager.testing_run_queued()
+        assert fake_hwconfig.powered
+        # Stop playback
+        FakeNfcReader.tag_callback.onTagChange(None)
+        faketimermanager.testing_run_queued()
+        # Elapse idle timer
+        faketimermanager.testing_run_queued()
+        assert not fake_hwconfig.powered