Merge branch 'master' of github.com:Bas-Korver/personal_wildlife_stream

src/api/api/routers/v1/internal_streams.py

@@ -1,3 +1,4 @@
+import asyncio
 from dataclasses import dataclass
 from typing import Annotated
 from uuid import UUID
@@ -6,6 +7,7 @@
 from litestar.background_tasks import BackgroundTask
 from litestar import Controller, get, post, MediaType
 from litestar.contrib.sqlalchemy.repository import SQLAlchemyAsyncRepository
+from litestar.datastructures import State
 from litestar.di import Provide
 from litestar.params import Body
 from models.animal import Animal
@@ -60,6 +62,12 @@ async def provide_stream_animals_repository(
     )
 
 
+async def save_current_stream(r, stream_id: UUID, delay: int = 300) -> None:
+    await asyncio.sleep(delay)
+
+    await r.set("current_stream_id", str(stream_id))
+
+
 async def store_animals(
     animals_repository: AnimalRepository,
     stream_animals_repository: StreamAnimalRepository,
@@ -142,6 +150,24 @@ async def get_stream(
 
         return stream
 
+    @post("/streams/{stream_id:uuid}/current")
+    async def set_current_stream(
+        self, state: State, streams_repository: StreamRepository, stream_id: UUID
+    ) -> Response:
+        stream = await streams_repository.get(
+            item_id=stream_id,
+        )
+
+        return Response(
+            f"Set current shown stream to {stream.id}.",
+            status_code=200,
+            background=BackgroundTask(
+                save_current_stream,
+                r=state.r,
+                stream_id=stream.id,
+            ),
+        )
+
     @post(
         "/streams/{stream_id:uuid}/animals",
         dependencies={

src/api/api/routers/v1/streams.py

@@ -68,13 +68,15 @@ class StreamsController(Controller):
     dependencies = {"streams_repository": Provide(provide_streams_repository)}
 
     @get("/current")
-    async def get_current_stream(self, db_session: AsyncSession) -> Redirect:
-        # TODO: Get current stream id, instead of first.
-        current_stream_id = (
-            (await db_session.execute(select(Stream.id))).scalars().first()
+    async def get_current_stream(
+        self, state: State, streams_repository: StreamRepository
+    ) -> Redirect:
+        current_stream_id = await state.r.get("current_stream_id")
+        current_stream = await streams_repository.get(
+            item_id=current_stream_id,
         )
 
-        return Redirect(path=f"/v1/streams/{current_stream_id}")
+        return Redirect(path=f"/v1/streams/{current_stream.id}")
 
     @get("/{stream_id:uuid}")
     async def get_stream(

src/video_streamer/video_streamer/modules/file_streamer.py

@@ -5,6 +5,8 @@
 from core import settings
 from db import RedisConnection
 from modules import make_logger
+import requests
+import structlog
 
 # Global variables.
 VIDEO_ITERATION_DELAY = 8.5  # TODO: Test with delay make configurable.
@@ -26,6 +28,7 @@ def start_stream_file(event):
     output_file_path = settings.SAVE_PATH / "stream.ts"
     video_file = r.json().get(video_key, ".video_path")
     video_path = settings.SAVE_PATH / video_file
+    stream_id = video_path.parents[0].name
     subprocess.run(
         [
             "ffmpeg",
@@ -145,6 +148,11 @@ def start_stream_file(event):
 
         logger.debug(f"Time needed to add new file: {time.time() - start_time}")
 
+        # Send request to internal API of current stream being shown.
+        requests.post(
+            f"http://localhost:8003/v1/internal-streams/streams/{stream_id}/current"  # TODO: Make URL configurable, and stream_id correct.
+        )
+
         # Remove intermediate files.
         pathlib.Path(intermediate_file_path).unlink()
         pathlib.Path()