feat(api): add /info and /stream/* endpoints

api/libretime_api/core/models/preference.py

@@ -1,4 +1,45 @@
+from enum import Enum
+
 from django.db import models
+from pydantic import BaseModel
+
+
+class SitePreferences(BaseModel):
+    station_name: str
+
+
+class MessageFormatKind(int, Enum):
+    ARTIST_TITLE = 0
+    SHOW_ARTIST_TITLE = 1
+    RADIO_SHOW = 2
+
+
+class StreamPreferences(BaseModel):
+    input_fade_transition: float
+    message_format: MessageFormatKind
+    message_offline: str
+    # input_auto_switch_off: bool
+    # input_auto_switch_on: bool
+    # input_main_user: str
+    # input_main_password: str
+    # replay_gain_enabled: bool
+    # replay_gain_offset: float
+    # track_fade_in: float
+    # track_fade_out: float
+    # track_fade_transition: float
+
+
+class StreamState(BaseModel):
+    input_main_connected: bool
+    input_main_streaming: bool
+    input_show_connected: bool
+    input_show_streaming: bool
+    schedule_streaming: bool
+
+
+class SitePreferenceManager(models.Manager):
+    def get_queryset(self):
+        return super().get_queryset().filter(user__isnull=True)
 
 
 class Preference(models.Model):
@@ -22,6 +63,36 @@ class Preference(models.Model):
         db_column="valstr",
     )
 
+    objects = models.Manager()
+    site = SitePreferenceManager()
+
+    @classmethod
+    def get_site_preferences(cls) -> SitePreferences:
+        entries = dict(cls.site.values_list("key", "value"))
+        return SitePreferences(
+            station_name=entries.get("station_name") or "LibreTime",
+        )
+
+    @classmethod
+    def get_stream_preferences(cls) -> StreamPreferences:
+        entries = dict(cls.site.values_list("key", "value"))
+        return StreamPreferences(
+            input_fade_transition=float(entries.get("default_transition_fade") or 0.0),
+            message_format=int(entries.get("stream_label_format") or 0),
+            message_offline=entries.get("off_air_meta") or "Offline",
+        )
+
+    @classmethod
+    def get_stream_state(cls) -> StreamState:
+        entries = dict(cls.site.values_list("key", "value"))
+        return StreamState(
+            input_main_connected=entries.get("master_dj") == "true",
+            input_main_streaming=entries.get("master_dj_switch") == "on",
+            input_show_connected=entries.get("live_dj") == "true",
+            input_show_streaming=entries.get("live_dj_switch") == "on",
+            schedule_streaming=entries.get("scheduled_play_switch") == "on",
+        )
+
     class Meta:
         managed = False
         db_table = "cc_pref"

api/libretime_api/core/router.py

@@ -3,10 +3,12 @@ from rest_framework import routers
 
 from .views import (
     CeleryTaskViewSet,
+    InfoView,
     LoginAttemptViewSet,
     PreferenceViewSet,
     ServiceRegisterViewSet,
-    StreamSettingViewSet,
+    StreamPreferencesView,
+    StreamStateView,
     ThirdPartyTrackReferenceViewSet,
     UserTokenViewSet,
     UserViewSet,
@@ -17,7 +19,6 @@ router = routers.DefaultRouter(trailing_slash=False)
 router.register("login-attempts", LoginAttemptViewSet)
 router.register("preferences", PreferenceViewSet)
 router.register("service-registers", ServiceRegisterViewSet)
-router.register("stream-settings", StreamSettingViewSet)
 router.register("users", UserViewSet)
 router.register("user-tokens", UserTokenViewSet)
 router.register("celery-tasks", CeleryTaskViewSet)
@@ -25,5 +26,8 @@ router.register("third-party-track-references", ThirdPartyTrackReferenceViewSet)
 
 urls = [
     *router.urls,
+    path("info", InfoView.as_view()),
     path("version", VersionView.as_view()),
+    path("stream/preferences", StreamPreferencesView.as_view()),
+    path("stream/state", StreamStateView.as_view()),
 ]

api/libretime_api/core/serializers/__init__.py

@@ -1,6 +1,7 @@
 from .auth import LoginAttemptSerializer, UserTokenSerializer
-from .info import VersionSerializer
-from .preference import PreferenceSerializer, StreamSettingSerializer
+from .info import InfoSerializer, VersionSerializer
+from .preference import PreferenceSerializer
 from .service import ServiceRegisterSerializer
+from .stream import StreamPreferencesSerializer, StreamStateSerializer
 from .user import UserSerializer
 from .worker import CeleryTaskSerializer, ThirdPartyTrackReferenceSerializer

api/libretime_api/core/serializers/info.py

@@ -3,4 +3,9 @@ from rest_framework import serializers
 
 # pylint: disable=abstract-method
 class VersionSerializer(serializers.Serializer):
-    api_version = serializers.CharField()
+    api_version = serializers.CharField(read_only=True)
+
+
+# pylint: disable=abstract-method
+class InfoSerializer(serializers.Serializer):
+    station_name = serializers.CharField(read_only=True)

api/libretime_api/core/serializers/preference.py

@@ -1,15 +1,9 @@
 from rest_framework import serializers
 
-from ..models import Preference, StreamSetting
+from ..models import Preference
 
 
 class PreferenceSerializer(serializers.ModelSerializer):
     class Meta:
         model = Preference
         fields = "__all__"
-
-
-class StreamSettingSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = StreamSetting
-        fields = "__all__"

api/libretime_api/core/serializers/stream.py

@@ -0,0 +1,17 @@
+from rest_framework import serializers
+
+
+# pylint: disable=abstract-method
+class StreamPreferencesSerializer(serializers.Serializer):
+    input_fade_transition = serializers.FloatField(read_only=True)
+    message_format = serializers.IntegerField(read_only=True)
+    message_offline = serializers.CharField(read_only=True)
+
+
+# pylint: disable=abstract-method
+class StreamStateSerializer(serializers.Serializer):
+    input_main_connected = serializers.BooleanField(read_only=True)
+    input_main_streaming = serializers.BooleanField(read_only=True)
+    input_show_connected = serializers.BooleanField(read_only=True)
+    input_show_streaming = serializers.BooleanField(read_only=True)
+    schedule_streaming = serializers.BooleanField(read_only=True)

api/libretime_api/core/tests/models/test_preference.py

@@ -0,0 +1,31 @@
+from libretime_api.core.models.preference import Preference
+
+
+# pylint: disable=invalid-name,unused-argument
+def test_preference_get_site_preferences(db):
+    result = Preference.get_site_preferences()
+    assert result.dict() == {
+        "station_name": "LibreTime",
+    }
+
+
+# pylint: disable=invalid-name,unused-argument
+def test_preference_get_stream_preferences(db):
+    result = Preference.get_stream_preferences()
+    assert result.dict() == {
+        "input_fade_transition": 0.0,
+        "message_format": 0,
+        "message_offline": "LibreTime - offline",
+    }
+
+
+# pylint: disable=invalid-name,unused-argument
+def test_preference_get_stream_state(db):
+    result = Preference.get_stream_state()
+    assert result.dict() == {
+        "input_main_connected": False,
+        "input_main_streaming": False,
+        "input_show_connected": False,
+        "input_show_streaming": False,
+        "schedule_streaming": True,
+    }

api/libretime_api/core/tests/views/test_info.py

@@ -0,0 +1,19 @@
+from rest_framework.test import APIClient
+
+
+# pylint: disable=invalid-name,unused-argument
+def test_version_get(db, api_client: APIClient):
+    response = api_client.get("/api/v2/version")
+    assert response.status_code == 200
+    assert response.json() == {
+        "api_version": "2.0.0",
+    }
+
+
+# pylint: disable=invalid-name,unused-argument
+def test_info_get(db, api_client: APIClient):
+    response = api_client.get("/api/v2/info")
+    assert response.status_code == 200
+    assert response.json() == {
+        "station_name": "LibreTime",
+    }

api/libretime_api/core/tests/views/test_stream.py

@@ -0,0 +1,25 @@
+from rest_framework.test import APIClient
+
+
+# pylint: disable=invalid-name,unused-argument
+def test_stream_preferences_get(db, api_client: APIClient):
+    response = api_client.get("/api/v2/stream/preferences")
+    assert response.status_code == 200
+    assert response.json() == {
+        "input_fade_transition": 0.0,
+        "message_format": 0,
+        "message_offline": "LibreTime - offline",
+    }
+
+
+# pylint: disable=invalid-name,unused-argument
+def test_stream_state_get(db, api_client: APIClient):
+    response = api_client.get("/api/v2/stream/state")
+    assert response.status_code == 200
+    assert response.json() == {
+        "input_main_connected": False,
+        "input_main_streaming": False,
+        "input_show_connected": False,
+        "input_show_streaming": False,
+        "schedule_streaming": True,
+    }

api/libretime_api/core/views/__init__.py

@@ -1,6 +1,7 @@
 from .auth import LoginAttemptViewSet, UserTokenViewSet
-from .info import VersionView
-from .preference import PreferenceViewSet, StreamSettingViewSet
+from .info import InfoView, VersionView
+from .preference import PreferenceViewSet
 from .service import ServiceRegisterViewSet
+from .stream import StreamPreferencesView, StreamStateView
 from .user import UserViewSet
 from .worker import CeleryTaskViewSet, ThirdPartyTrackReferenceViewSet

api/libretime_api/core/views/info.py

@@ -3,7 +3,8 @@ from rest_framework.permissions import AllowAny
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
-from ..serializers import VersionSerializer
+from ..models import Preference
+from ..serializers import InfoSerializer, VersionSerializer
 
 
 class VersionView(APIView):
@@ -12,3 +13,18 @@ class VersionView(APIView):
 
     def get(self, request):
         return Response({"api_version": settings.API_VERSION})
+
+
+class InfoView(APIView):
+    permission_classes = [AllowAny]
+    serializer_class = InfoSerializer
+
+    def get(self, request):
+        data = Preference.get_site_preferences()
+        return Response(
+            data.dict(
+                include={
+                    "station_name",
+                }
+            )
+        )

api/libretime_api/core/views/preference.py

@@ -1,16 +1,10 @@
 from rest_framework import viewsets
 
-from ..models import Preference, StreamSetting
-from ..serializers import PreferenceSerializer, StreamSettingSerializer
+from ..models import Preference
+from ..serializers import PreferenceSerializer
 
 
 class PreferenceViewSet(viewsets.ModelViewSet):
     queryset = Preference.objects.all()
     serializer_class = PreferenceSerializer
     model_permission_name = "preference"
-
-
-class StreamSettingViewSet(viewsets.ModelViewSet):
-    queryset = StreamSetting.objects.all()
-    serializer_class = StreamSettingSerializer
-    model_permission_name = "streamsetting"

api/libretime_api/core/views/stream.py

@@ -0,0 +1,44 @@
+from rest_framework import views
+from rest_framework.response import Response
+
+from ...permissions import IsSystemTokenOrUser
+from ..models import Preference
+from ..serializers import StreamPreferencesSerializer, StreamStateSerializer
+
+
+class StreamPreferencesView(views.APIView):
+    permission_classes = [IsSystemTokenOrUser]
+    serializer_class = StreamPreferencesSerializer
+    model_permission_name = "streamsetting"
+
+    def get(self, request):
+        data = Preference.get_stream_preferences()
+        return Response(
+            data.dict(
+                include={
+                    "input_fade_transition",
+                    "message_format",
+                    "message_offline",
+                }
+            )
+        )
+
+
+class StreamStateView(views.APIView):
+    permission_classes = [IsSystemTokenOrUser]
+    serializer_class = StreamStateSerializer
+    model_permission_name = "streamsetting"
+
+    def get(self, request):
+        data = Preference.get_stream_state()
+        return Response(
+            data.dict(
+                include={
+                    "input_main_connected",
+                    "input_main_streaming",
+                    "input_show_connected",
+                    "input_show_streaming",
+                    "schedule_streaming",
+                }
+            )
+        )