From a2353daab44715ad0690534c54b3da723fe62e2e Mon Sep 17 00:00:00 2001 From: Trey Turner Date: Sat, 18 Jul 2026 14:18:52 -0500 Subject: [PATCH 1/3] fix: resolve Plexamp server from playback tags --- main.py | 8 +++++- plexamp.py | 50 +++++++++++++++++++++++++++++++++ tests/test_plexamp.py | 65 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 1 deletion(-) create mode 100644 plexamp.py create mode 100644 tests/test_plexamp.py diff --git a/main.py b/main.py index 99e27f8..3e42f93 100644 --- a/main.py +++ b/main.py @@ -5,6 +5,8 @@ import requests from adafruit_pn532.uart import PN532_UART +from plexamp import InvalidPlaybackURL, prepare_playback_url + # ---------------------------- # Helper: find PN532 serial device # ---------------------------- @@ -133,6 +135,11 @@ def connect_reader(): # Convert to local Plexamp URL local_url = full_url.replace("https://listen.plex.tv", "http://localhost:32500") local_url = local_url.replace("http://listen.plex.tv", "http://localhost:32500") + try: + local_url = prepare_playback_url(local_url) + except InvalidPlaybackURL as e: + print(f"Invalid playback URL: {e}") + continue print(f"Local Plexamp URL: {local_url}") # If same URL as before and still within active session, skip @@ -157,4 +164,3 @@ def connect_reader(): pn532 = connect_reader() - diff --git a/plexamp.py b/plexamp.py new file mode 100644 index 0000000..04f5038 --- /dev/null +++ b/plexamp.py @@ -0,0 +1,50 @@ +"""Helpers for preparing Plexamp companion API requests.""" + +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + + +PLAY_MEDIA_PATH = "/player/playback/playMedia" + + +class InvalidPlaybackURL(ValueError): + """Raised when an autoplay URL cannot identify its Plex server.""" + + +def prepare_playback_url(url): + """Add the Plex server identifier required by direct Plexamp requests. + + Plex web links normally derive ``machineIdentifier`` from the authority of + their nested ``server://`` URI. Direct companion API calls bypass that web + client preparation, so reproduce it here for playMedia requests. + """ + parsed_url = urlsplit(url) + if parsed_url.path != PLAY_MEDIA_PATH: + return url + + query_items = parse_qsl(parsed_url.query, keep_blank_values=True) + inner_uri = next( + (value for key, value in query_items if key == "uri"), + None, + ) + if inner_uri is None: + raise InvalidPlaybackURL("playMedia URL is missing its uri parameter") + + parsed_uri = urlsplit(inner_uri) + if parsed_uri.scheme != "server" or not parsed_uri.netloc: + raise InvalidPlaybackURL( + "playMedia uri must use the server:///... format" + ) + + if any(key == "machineIdentifier" for key, _ in query_items): + return url + + query_items.append(("machineIdentifier", parsed_uri.netloc)) + return urlunsplit( + ( + parsed_url.scheme, + parsed_url.netloc, + parsed_url.path, + urlencode(query_items), + parsed_url.fragment, + ) + ) diff --git a/tests/test_plexamp.py b/tests/test_plexamp.py new file mode 100644 index 0000000..1a05cc5 --- /dev/null +++ b/tests/test_plexamp.py @@ -0,0 +1,65 @@ +import unittest +from urllib.parse import parse_qs, urlencode, urlsplit + +from plexamp import InvalidPlaybackURL, prepare_playback_url + + +class PreparePlaybackURLTests(unittest.TestCase): + machine_id = "0123456789abcdef0123456789abcdef01234567" + + def playback_url(self, inner_uri, **parameters): + query = {"uri": inner_uri} + query.update(parameters) + return "http://localhost:32500/player/playback/playMedia?" + urlencode(query) + + def test_adds_machine_identifier_from_server_uri(self): + inner_uri = ( + f"server://{self.machine_id}/com.plexapp.plugins.library/" + "library/metadata/12345/children" + ) + + result = prepare_playback_url(self.playback_url(inner_uri)) + parameters = parse_qs(urlsplit(result).query) + + self.assertEqual(parameters["machineIdentifier"], [self.machine_id]) + self.assertEqual(parameters["uri"], [inner_uri]) + + def test_preserves_explicit_machine_identifier(self): + inner_uri = f"server://{self.machine_id}/library/metadata/12345" + explicit_id = "explicit-server-id" + + result = prepare_playback_url( + self.playback_url(inner_uri, machineIdentifier=explicit_id) + ) + + self.assertEqual( + result, + self.playback_url( + inner_uri, + machineIdentifier=explicit_id, + ), + ) + + def test_preserves_equivalent_percent_encoded_uri_data(self): + inner_uri = ( + f"server://{self.machine_id}/library/metadata/12345/children" + "?source=a%2Fb&title=Album One" + ) + + result = prepare_playback_url(self.playback_url(inner_uri)) + + self.assertEqual(parse_qs(urlsplit(result).query)["uri"], [inner_uri]) + + def test_rejects_missing_inner_uri(self): + url = "http://localhost:32500/player/playback/playMedia?type=music" + + with self.assertRaises(InvalidPlaybackURL): + prepare_playback_url(url) + + def test_rejects_malformed_inner_uri(self): + with self.assertRaises(InvalidPlaybackURL): + prepare_playback_url(self.playback_url("not-a-server-uri")) + + +if __name__ == "__main__": + unittest.main() From 9d04aa4f08fa00626cea9e4488514ad918db7ffc Mon Sep 17 00:00:00 2001 From: Trey Turner Date: Sat, 18 Jul 2026 14:22:06 -0500 Subject: [PATCH 2/3] fix: verify playback after NFC scans --- main.py | 14 ++- plexamp.py | 189 ++++++++++++++++++++++++++++++++- tests/test_plexamp_playback.py | 184 ++++++++++++++++++++++++++++++++ 3 files changed, 378 insertions(+), 9 deletions(-) create mode 100644 tests/test_plexamp_playback.py diff --git a/main.py b/main.py index 3e42f93..4f34db8 100644 --- a/main.py +++ b/main.py @@ -2,10 +2,9 @@ import os import glob import serial -import requests from adafruit_pn532.uart import PN532_UART -from plexamp import InvalidPlaybackURL, prepare_playback_url +from plexamp import InvalidPlaybackURL, PlexampClient, prepare_playback_url # ---------------------------- # Helper: find PN532 serial device @@ -68,6 +67,7 @@ def connect_reader(): # ---------------------------- if __name__ == "__main__": pn532 = connect_reader() + plexamp = PlexampClient() while True: try: @@ -147,15 +147,14 @@ def connect_reader(): print("Same tag & URL already active — skipping trigger.") continue - last_url = local_url - # Trigger Plexamp playback try: - response = requests.get(local_url) - if response.ok: + result = plexamp.play(local_url) + if result.success: + last_url = local_url print(f"Playback triggered! ({kind})") else: - print(f"Error triggering playback: {response.status_code}") + print(f"Failed to trigger Plexamp: {result.message}") except Exception as e: print(f"Failed to trigger Plexamp: {e}") @@ -163,4 +162,3 @@ def connect_reader(): print(f"Reader error: {e}. Reconnecting...") pn532 = connect_reader() - diff --git a/plexamp.py b/plexamp.py index 04f5038..71962f7 100644 --- a/plexamp.py +++ b/plexamp.py @@ -1,15 +1,49 @@ -"""Helpers for preparing Plexamp companion API requests.""" +"""Helpers for preparing and verifying Plexamp companion API requests.""" +import itertools +import time +import uuid +from dataclasses import dataclass +from xml.etree import ElementTree from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit +import requests + PLAY_MEDIA_PATH = "/player/playback/playMedia" +TIMELINE_PATH = "/player/timeline/poll" +ACTIVE_STATES = {"buffering", "playing"} +CLIENT_IDENTIFIER = str(uuid.uuid4()) +DEVICE_NAME = "Plexamp NFC UART" +DEFAULT_REQUEST_TIMEOUT = 5 +DEFAULT_VERIFICATION_TIMEOUT = 10 +DEFAULT_POLL_INTERVAL = 0.5 + +_COMMAND_IDS = itertools.count(1) class InvalidPlaybackURL(ValueError): """Raised when an autoplay URL cannot identify its Plex server.""" +@dataclass(frozen=True) +class PlaybackResult: + """The verified outcome of a Plexamp playback request.""" + + success: bool + message: str + + +@dataclass(frozen=True) +class _MusicTimeline: + state: str + queue_id: str + + +class _PlaybackError(RuntimeError): + """Raised for expected companion API failures.""" + + def prepare_playback_url(url): """Add the Plex server identifier required by direct Plexamp requests. @@ -48,3 +82,156 @@ def prepare_playback_url(url): parsed_url.fragment, ) ) + + +class PlexampClient: + """Send playback commands and verify their effect on Plexamp.""" + + def __init__( + self, + session=None, + request_timeout=DEFAULT_REQUEST_TIMEOUT, + verification_timeout=DEFAULT_VERIFICATION_TIMEOUT, + poll_interval=DEFAULT_POLL_INTERVAL, + monotonic=time.monotonic, + sleep=time.sleep, + ): + self.session = session or requests.Session() + self.request_timeout = request_timeout + self.verification_timeout = verification_timeout + self.poll_interval = poll_interval + self.monotonic = monotonic + self.sleep = sleep + self.headers = { + "X-Plex-Client-Identifier": CLIENT_IDENTIFIER, + "X-Plex-Device-Name": DEVICE_NAME, + } + + def play(self, url): + """Request playback and wait for Plexamp to create an active queue.""" + command_id = next(_COMMAND_IDS) + + try: + prepared_url = prepare_playback_url(url) + previous_timeline = self._music_timeline( + prepared_url, + max(command_id - 1, 0), + self.request_timeout, + ) + command_url = _replace_query_parameter( + prepared_url, + "commandID", + str(command_id), + ) + self._get(command_url, self.request_timeout, "playback") + except (InvalidPlaybackURL, _PlaybackError, ValueError) as error: + return PlaybackResult(False, str(error)) + + previous_queue_id = None + if previous_timeline.state in ACTIVE_STATES: + previous_queue_id = previous_timeline.queue_id + + deadline = self.monotonic() + self.verification_timeout + while self.monotonic() < deadline: + remaining = deadline - self.monotonic() + try: + timeline = self._music_timeline( + prepared_url, + command_id, + min(self.request_timeout, remaining), + ) + except _PlaybackError as error: + return PlaybackResult(False, str(error)) + + has_active_queue = ( + timeline.state in ACTIVE_STATES and bool(timeline.queue_id) + ) + queue_changed = ( + previous_queue_id is None + or timeline.queue_id != previous_queue_id + ) + if has_active_queue and queue_changed: + return PlaybackResult(True, "Plexamp created an active play queue") + + remaining = deadline - self.monotonic() + if remaining <= 0: + break + self.sleep(min(self.poll_interval, remaining)) + + return PlaybackResult( + False, + "Plexamp did not create an active play queue within " + f"{self.verification_timeout:g} seconds", + ) + + def _music_timeline(self, playback_url, command_id, timeout): + parsed_url = urlsplit(playback_url) + if not parsed_url.scheme or not parsed_url.netloc: + raise _PlaybackError("playback URL must include a scheme and host") + + timeline_url = urlunsplit( + ( + parsed_url.scheme, + parsed_url.netloc, + TIMELINE_PATH, + urlencode({"wait": "0", "commandID": str(command_id)}), + "", + ) + ) + response = self._get(timeline_url, timeout, "timeline") + + try: + document = ElementTree.fromstring(response.text) + except ElementTree.ParseError as error: + raise _PlaybackError(f"Plexamp returned an invalid timeline: {error}") + + for timeline in document.iter(): + if ( + timeline.tag.rsplit("}", 1)[-1] == "Timeline" + and timeline.get("type") == "music" + ): + return _MusicTimeline( + state=timeline.get("state", "").lower(), + queue_id=timeline.get("playQueueID", ""), + ) + + raise _PlaybackError("Plexamp timeline did not include music state") + + def _get(self, url, timeout, request_name): + try: + response = self.session.get( + url, + headers=self.headers, + timeout=timeout, + ) + except requests.RequestException as error: + raise _PlaybackError(f"Plexamp {request_name} request failed: {error}") + + if not 200 <= response.status_code < 300: + raise _PlaybackError( + f"Plexamp {request_name} request returned HTTP " + f"{response.status_code}" + ) + return response + + +def _replace_query_parameter(url, name, value): + parsed_url = urlsplit(url) + query_items = [ + (key, item_value) + for key, item_value in parse_qsl( + parsed_url.query, + keep_blank_values=True, + ) + if key != name + ] + query_items.append((name, value)) + return urlunsplit( + ( + parsed_url.scheme, + parsed_url.netloc, + parsed_url.path, + urlencode(query_items), + parsed_url.fragment, + ) + ) diff --git a/tests/test_plexamp_playback.py b/tests/test_plexamp_playback.py new file mode 100644 index 0000000..8c9004a --- /dev/null +++ b/tests/test_plexamp_playback.py @@ -0,0 +1,184 @@ +import unittest +from urllib.parse import parse_qs, urlencode, urlsplit + +import requests + +from plexamp import PlexampClient + + +class FakeResponse: + def __init__(self, status_code=200, text=""): + self.status_code = status_code + self.text = text + + +class FakeSession: + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + def get(self, url, **kwargs): + self.calls.append((url, kwargs)) + response = self.responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + +class FakeClock: + def __init__(self): + self.now = 0 + self.sleeps = [] + + def monotonic(self): + return self.now + + def sleep(self, seconds): + self.sleeps.append(seconds) + self.now += seconds + + +def timeline(state, queue_id=None): + queue_attribute = "" + if queue_id is not None: + queue_attribute = f' playQueueID="{queue_id}"' + return FakeResponse( + text=( + "" + f'' + "" + ) + ) + + +class PlexampClientTests(unittest.TestCase): + machine_id = "0123456789abcdef0123456789abcdef01234567" + + def playback_url(self): + inner_uri = f"server://{self.machine_id}/library/metadata/12345/children" + return "http://localhost:32500/player/playback/playMedia?" + urlencode( + {"uri": inner_uri} + ) + + def client(self, responses, verification_timeout=10): + session = FakeSession(responses) + clock = FakeClock() + client = PlexampClient( + session=session, + verification_timeout=verification_timeout, + poll_interval=0.5, + monotonic=clock.monotonic, + sleep=clock.sleep, + ) + return client, session, clock + + def test_empty_player_transition_to_populated_queue_succeeds(self): + client, session, _ = self.client([ + timeline("stopped"), + FakeResponse(), + timeline("playing", "42"), + ]) + + result = client.play(self.playback_url()) + + self.assertTrue(result.success) + self.assertEqual(len(session.calls), 3) + command_url, command_options = session.calls[1] + parameters = parse_qs(urlsplit(command_url).query) + self.assertIn("commandID", parameters) + self.assertEqual( + parameters["machineIdentifier"], + [self.machine_id], + ) + for _, options in session.calls: + self.assertTrue(options["headers"]["X-Plex-Client-Identifier"]) + self.assertEqual( + options["headers"]["X-Plex-Device-Name"], + "Plexamp NFC UART", + ) + self.assertIn("headers", command_options) + + def test_successive_commands_use_increasing_ids(self): + client, session, _ = self.client([ + timeline("stopped"), + FakeResponse(), + timeline("playing", "42"), + timeline("stopped"), + FakeResponse(), + timeline("playing", "43"), + ]) + + self.assertTrue(client.play(self.playback_url()).success) + self.assertTrue(client.play(self.playback_url()).success) + + first_id = int(parse_qs(urlsplit(session.calls[1][0]).query)["commandID"][0]) + second_id = int(parse_qs(urlsplit(session.calls[4][0]).query)["commandID"][0]) + self.assertGreater(second_id, first_id) + + def test_existing_playback_requires_queue_to_change(self): + client, session, clock = self.client([ + timeline("playing", "42"), + FakeResponse(), + timeline("playing", "42"), + timeline("buffering", "43"), + ]) + + result = client.play(self.playback_url()) + + self.assertTrue(result.success) + self.assertEqual(len(session.calls), 4) + self.assertEqual(clock.sleeps, [0.5]) + + def test_non_successful_response_fails(self): + client, session, _ = self.client([ + timeline("stopped"), + FakeResponse(status_code=503), + ]) + + result = client.play(self.playback_url()) + + self.assertFalse(result.success) + self.assertIn("HTTP 503", result.message) + self.assertEqual(len(session.calls), 2) + + def test_request_error_fails(self): + client, session, _ = self.client([ + timeline("stopped"), + requests.ConnectionError("connection refused"), + ]) + + result = client.play(self.playback_url()) + + self.assertFalse(result.success) + self.assertIn("connection refused", result.message) + self.assertEqual(len(session.calls), 2) + + def test_malformed_timeline_fails(self): + client, session, _ = self.client([ + FakeResponse(text=" Date: Sat, 18 Jul 2026 14:22:55 -0500 Subject: [PATCH 3/3] ci: run Python tests --- .github/workflows/tests.yml | 46 +++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 .github/workflows/tests.yml diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..e621f49 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,46 @@ +name: Tests + +on: + pull_request: + branches: + - main + push: + branches: + - main + +permissions: + contents: read + +concurrency: + group: tests-${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: + - "3.8" + - "3.14" + + steps: + - name: Check out repository + uses: actions/checkout@v7 + + - name: Set up Python + uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + cache: pip + cache-dependency-path: requirements.txt + + - name: Install dependencies + run: python -m pip install -r requirements.txt + + - name: Run tests + run: python -m unittest discover -s tests -v + + - name: Compile Python sources + run: python -m compileall -q main.py plexamp.py