fixup! core: services: recoder_extractor: Add first version

Author: patrickelectric

core/services/recorder_extractor/main.py

@@ -11,6 +11,7 @@
 
 from urllib.parse import quote
 
+from aiocache import cached
 from commonwealth.utils.apis import GenericErrorHandlingRoute, PrettyJSONResponse
 from commonwealth.utils.general import file_is_open_async
 from commonwealth.utils.logs import InterceptHandler, init_logger
@@ -26,6 +27,9 @@
 RECORDER_DIR = Path("/usr/blueos/userdata/recorder")
 PORT = 9150
 
+# Prevent thumbnails from being generated while MCAP extraction is running
+thumbnail_lock = asyncio.Lock()
+
 logging.basicConfig(handlers=[InterceptHandler()], level=logging.DEBUG)
 init_logger(SERVICE_NAME)
 logger.info("Starting Recorder Extractor service")
@@ -86,21 +90,31 @@ def parse_duration_ns(discover_output: str) -> int:
     return duration_ns
 
 
-def build_thumbnail_bytes(path: Path) -> bytes:
+@cached
+async def build_thumbnail_bytes(path: Path) -> bytes:
     """
-    Extract a single JPEG frame from the recording using a raw gst-launch pipeline.
+    Extract a single JPEG frame from the recording using a raw gst-launch pipeline (ASYNC).
 
     Seek to the middle of the file, scale to 320x180, and encode as JPEG. If any step fails,
     propagate an HTTP 500 so callers can fall back.
     """
     # 1) Discover duration (nanoseconds) using gst-discoverer
     discover_cmd = ["gst-discoverer-1.0", f"file://{path}"]
-    discover = subprocess.run(discover_cmd, capture_output=True, text=True, check=False)
-    if discover.returncode != 0:
-        logger.error("gst-discoverer-1.0 failed for %s: %s", path, discover.stderr.strip())
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to inspect recording.")
+    discover_proc = await asyncio.create_subprocess_exec(
+        *discover_cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        text=True,
+    )
+    stdout, stderr = await discover_proc.communicate()
+    if discover_proc.returncode != 0:
+        logger.error(f"gst-discoverer-1.0 failed for {path}: {stderr.strip()}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to inspect recording.",
+        )
 
-    duration_ns = parse_duration_ns(discover.stdout)
+    duration_ns = parse_duration_ns(stdout)
     target_ns = duration_ns // 2 if duration_ns > 0 else 0
     target_sec = target_ns / 1_000_000_000
 
@@ -125,13 +139,21 @@ def build_thumbnail_bytes(path: Path) -> bytes:
         f"Thumbnail target: duration_ns={duration_ns} target_ns={target_ns} target_sec={target_sec:.3f} file={path}"
     )
     logger.info(f"Thumbnail command: {' '.join(play_cmd)}")
-    result = subprocess.run(play_cmd, capture_output=True, check=False)
-    if result.returncode != 0 or not result.stdout:
-        stderr = result.stderr.decode("utf-8", "ignore")
-        logger.error(f"gst-play-1.0 failed for {path} (code={result.returncode}): {stderr}")
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to generate thumbnail.")
+    play_proc = await asyncio.create_subprocess_exec(
+        *play_cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    out_bytes, err_bytes = await play_proc.communicate()
+    if play_proc.returncode != 0 or not out_bytes:
+        stderr = err_bytes.decode("utf-8", "ignore")
+        logger.error(f"gst-play-1.0 failed for {path} (code={play_proc.returncode}): {stderr}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to generate thumbnail.",
+        )
 
-    return result.stdout
+    return out_bytes
 
 
 async def extract_mcap_recordings() -> None:
@@ -159,27 +181,21 @@ async def extract_mcap_recordings() -> None:
                     str(output_dir),
                 ]
                 logger.info(f"Extracting MCAP video to {output_dir} with command: {' '.join(command)}")
-                process = await asyncio.create_subprocess_exec(
-                    *command,
-                    stdout=asyncio.subprocess.PIPE,
-                    stderr=asyncio.subprocess.PIPE,
-                )
-                stdout, stderr = await process.communicate()
+                async with thumbnail_lock:
+                    process = await asyncio.create_subprocess_exec(
+                        *command,
+                        stdout=asyncio.subprocess.PIPE,
+                        stderr=asyncio.subprocess.PIPE,
+                    )
+                    stdout, stderr = await process.communicate()
                 if process.returncode != 0:
                     logger.error(
-                        "MCAP extract failed for %s (code=%s): %s",
-                        mcap_path,
-                        process.returncode,
-                        stderr.decode("utf-8", "ignore"),
+                        f"MCAP extract failed for {mcap_path} (code={process.returncode}): {stderr.decode('utf-8', 'ignore')}",
                     )
                 else:
-                    logger.info(
-                        "MCAP extract completed for %s: %s",
-                        mcap_path,
-                        stdout.decode("utf-8", "ignore").strip(),
-                    )
+                    logger.info(f"MCAP extract completed for {mcap_path}: {stdout.decode('utf-8', 'ignore').strip()}")
         except Exception as exception:
-            logger.exception("MCAP extraction loop failed: %s", exception)
+            logger.exception(f"MCAP extraction loop failed: {exception}")
 
 
 def to_http_exception(endpoint: Callable[..., Any]) -> Callable[..., Any]:
@@ -245,7 +261,8 @@ async def list_recordings() -> List[RecordingFile]:
 @to_http_exception
 async def get_recording_thumbnail(filename: str) -> StreamingResponse:
     path = resolve_recording(filename)
-    thumbnail_bytes = await asyncio.to_thread(build_thumbnail_bytes, path)
+    async with thumbnail_lock:
+        thumbnail_bytes = await build_thumbnail_bytes(path)
     return StreamingResponse(BytesIO(thumbnail_bytes), media_type="image/jpeg")
 
 
@@ -260,7 +277,7 @@ async def delete_recording(filename: str) -> None:
     try:
         path.unlink()
     except Exception as exception:
-        logger.exception("Failed to delete recording %s", filename)
+        logger.exception(f"Failed to delete recording {filename}")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="Failed to delete recording.",

core/services/recorder_extractor/pyproject.toml

@@ -4,6 +4,7 @@ version = "0.1.0"
 description = "Serve and manage recorded MP4 files with thumbnails and streaming."
 requires-python = ">=3.11"
 dependencies = [
+    "aiocache==0.12.2",
     "commonwealth==0.1.0",
     "fastapi==0.105.0",
     "fastapi-versioning==0.9.1",