Accept audio submitted as a URL

evaluation_function/audio_processing.py

@@ -16,6 +16,10 @@
 import contextlib
 import io
 import os
+import shutil
+import tempfile
+import urllib.request
+from urllib.parse import urlparse
 
 # basic_pitch (and its librosa / numba / onnxruntime dependencies) is a very
 # heavy import - tens of seconds on a cold Lambda. It is imported lazily inside
@@ -46,6 +50,65 @@
 # File extensions we treat as "already MIDI, no transcription needed"
 MIDI_EXTENSIONS = [".mid", ".midi"]
 
+# How long to wait for an audio download before giving up.
+DOWNLOAD_TIMEOUT_SECONDS = 30
+
+
+# Helpers for reading a local path or a URL
+# ---------------------------------------------------------------------
+def file_extension(path):
+    """
+    Return the lowercase file extension of a local path or a URL.
+
+    Audio submitted from the platform arrives as a URL to an object in a
+    bucket, and those URLs are usually presigned, so splitting the raw
+    string would read the extension as ".wav?X-Amz-Signature=...".
+    Taking the URL path first drops the query string and the fragment.
+    Local paths are unaffected -- they parse as a path and nothing else.
+    """
+    return os.path.splitext(urlparse(str(path)).path)[1].lower()
+
+
+def file_name(path):
+    """
+    Return the base name of a local path or a URL, without any query
+    string. Used for messages shown to the student, so a presigned URL
+    does not print its whole signature back at them.
+    """
+    return os.path.basename(urlparse(str(path)).path)
+
+
+def is_url(value):
+    """
+    Return True if value is an http(s) URL rather than a local file path.
+    """
+    return isinstance(value, str) and urlparse(value).scheme in ("http", "https")
+
+
+@contextlib.contextmanager
+def local_audio_file(audio_path):
+    """
+    Yield a local file path for audio_path.
+
+    Basic Pitch reads audio with librosa, which needs a file on disk, so
+    a URL has to be downloaded first. A local path is handed straight
+    back. A downloaded file is removed once transcription is done.
+    """
+    if not is_url(audio_path):
+        yield audio_path
+        return
+
+    handle, temp_path = tempfile.mkstemp(suffix=file_extension(audio_path))
+    os.close(handle)
+    try:
+        with urllib.request.urlopen(
+            audio_path, timeout=DOWNLOAD_TIMEOUT_SECONDS
+        ) as response, open(temp_path, "wb") as temp_file:
+            shutil.copyfileobj(response, temp_file)
+        yield temp_path
+    finally:
+        os.remove(temp_path)
+
 
 # Helper function to check if the response is audio
 # ---------------------------------------------------------------------
@@ -55,19 +118,19 @@ def is_audio_input(response):
     through the AMT pipeline before it can be compared.
 
     response can be:
-      - a file path string (checked by extension)
+      - a file path or URL string (checked by extension)
       - a dict that already has a "notes" key (already MIDI, skip AMT)
     """
     # Case 1: response is already a notes dictionary, no AMT needed
     if isinstance(response, dict) and "notes" in response:
         return False
 
-    # Case 2: response is a file path string, check its extension
+    # Case 2: response is a file path or URL string, check its extension
     if isinstance(response, str):
-        file_extension = os.path.splitext(response)[1].lower()
-        if file_extension in AUDIO_EXTENSIONS:
+        extension = file_extension(response)
+        if extension in AUDIO_EXTENSIONS:
             return True
-        if file_extension in MIDI_EXTENSIONS:
+        if extension in MIDI_EXTENSIONS:
             return False
 
     # Default: if we cannot tell, assume it is not audio
@@ -323,9 +386,14 @@ def transcription_pipeline(audio_path, model, apply_postprocessing=True):
     Full production pipeline: run Basic Pitch, optionally apply post-processing, 
     and return notes ready to hand to compare_MIDI.py.
 
+    audio_path may be a local path or a URL; a URL is downloaded to a
+    temporary file first. runtime_seconds covers transcription only, not
+    the download.
+
     Returns (compare_midi_input, predicted_notes, runtime_seconds).
     """
-    predicted_notes, predicted_midi, runtime_seconds = transcribe_audio(audio_path, model)
+    with local_audio_file(audio_path) as local_path:
+        predicted_notes, predicted_midi, runtime_seconds = transcribe_audio(local_path, model)
 
     if apply_postprocessing:
         predicted_notes = postprocess_predictions(predicted_notes)

evaluation_function/audio_processing_test.py

@@ -0,0 +1,115 @@
+"""
+audio_processing_test.py
+========================
+Unit tests for reading audio input. The platform sends uploaded
+recordings as URLs to a bucket, so these cover URLs as well as the
+local paths used in development.
+
+Sections
+--------
+1. Tests for file_extension, file_name and is_url
+2. Tests for is_audio_input
+3. Tests for local_audio_file
+"""
+
+
+import os
+import tempfile
+import threading
+import unittest
+from functools import partial
+from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
+
+from .audio_processing import (
+    file_extension,
+    file_name,
+    is_url,
+    is_audio_input,
+    local_audio_file,
+)
+
+
+# A presigned bucket URL: the extension is followed by a query string.
+SIGNED_URL = "https://bucket.example.com/uploads/rec.wav?X-Amz-Signature=abc&expires=1"
+
+
+# 1. Tests for file_extension, file_name and is_url
+# ------------------------------------------------------------------------------
+class TestPathHelpers(unittest.TestCase):
+
+    def test_extension_of_local_path(self):
+        assert file_extension("/uploads/rec.wav") == ".wav"
+
+    def test_extension_ignores_query_string(self):
+        assert file_extension(SIGNED_URL) == ".wav"
+
+    def test_extension_ignores_fragment(self):
+        assert file_extension("https://x.example.com/rec.mp3#part2") == ".mp3"
+
+    def test_extension_is_lowercased(self):
+        assert file_extension("/uploads/REC.WAV") == ".wav"
+
+    def test_name_drops_the_signature(self):
+        assert file_name(SIGNED_URL) == "rec.wav"
+
+    def test_is_url_only_for_http(self):
+        assert is_url("https://x.example.com/rec.wav") is True
+        assert is_url("http://x.example.com/rec.wav") is True
+        assert is_url("/uploads/rec.wav") is False
+        assert is_url({"notes": []}) is False
+
+
+# 2. Tests for is_audio_input
+# ------------------------------------------------------------------------------
+class TestIsAudioInput(unittest.TestCase):
+
+    def test_signed_url_is_recognised_as_audio(self):
+        assert is_audio_input(SIGNED_URL) is True
+
+    def test_local_audio_path_is_recognised(self):
+        assert is_audio_input("/uploads/rec.flac") is True
+
+    def test_notes_dict_is_not_audio(self):
+        assert is_audio_input({"notes": []}) is False
+
+    def test_midi_url_is_not_audio(self):
+        assert is_audio_input("https://x.example.com/rec.mid?sig=abc") is False
+
+
+# 3. Tests for local_audio_file
+# ------------------------------------------------------------------------------
+class QuietHandler(SimpleHTTPRequestHandler):
+    def log_message(self, *args):
+        pass
+
+
+class TestLocalAudioFile(unittest.TestCase):
+
+    CONTENT = b"stand-in for audio bytes"
+
+    def test_local_path_is_handed_back_unchanged(self):
+        with local_audio_file("/uploads/rec.wav") as path:
+            assert path == "/uploads/rec.wav"
+
+    def test_url_is_downloaded_and_removed_afterwards(self):
+        with tempfile.TemporaryDirectory() as served_directory:
+            with open(os.path.join(served_directory, "rec.wav"), "wb") as f:
+                f.write(self.CONTENT)
+
+            handler = partial(QuietHandler, directory=served_directory)
+            server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
+            threading.Thread(target=server.serve_forever, daemon=True).start()
+
+            try:
+                url = f"http://127.0.0.1:{server.server_port}/rec.wav?X-Amz-Signature=abc"
+                with local_audio_file(url) as path:
+                    # Basic Pitch picks its decoder by extension, so the
+                    # temporary file has to keep the .wav suffix.
+                    assert path.endswith(".wav")
+                    with open(path, "rb") as f:
+                        assert f.read() == self.CONTENT
+                    downloaded_path = path
+
+                assert not os.path.exists(downloaded_path)
+            finally:
+                server.shutdown()

evaluation_function/preview.py

@@ -13,12 +13,11 @@
 """
 
 import json
-import os
 from typing import Any
 
 from lf_toolkit.preview import Result, Params, Preview
 
-from .audio_processing import AUDIO_EXTENSIONS
+from .audio_processing import AUDIO_EXTENSIONS, file_extension, file_name
 from .compare_MIDI import PITCH_CLASS_NAMES
 
 # Fixed messages, named so that tests can assert which case was hit without
@@ -68,9 +67,8 @@ def preview_function(response: Any, params: Params) -> Result:
         # An audio recording is reported as such. Transcribing it here would
         # take seconds, which is far too slow while the student is working.
         if isinstance(response, str):
-            extension = os.path.splitext(response)[1].lower()
-            if extension in AUDIO_EXTENSIONS:
-                name = os.path.basename(response)
+            if file_extension(response) in AUDIO_EXTENSIONS:
+                name = file_name(response)
                 return Result(preview=Preview(
                     feedback=f"Audio recording {name}, transcribed on submission."
                 ))

evaluation_function/preview_test.py

@@ -72,6 +72,14 @@ def test_audio_path_is_described_without_transcribing(self):
         assert "audio" in feedback.lower()
         assert "practice.wav" in feedback
 
+    def test_signed_url_is_described_without_the_signature(self):
+        feedback = feedback_for(
+            "https://bucket.example.com/uploads/practice.wav?X-Amz-Signature=abc"
+        )
+        assert "audio" in feedback.lower()
+        assert "practice.wav" in feedback
+        assert "X-Amz-Signature" not in feedback
+
     def test_unreadable_submission_is_reported(self):
         assert feedback_for("this is not MIDI at all") == UNREADABLE_MESSAGE