Overhaul time management

Author: AngellusMortis

HISTORY.rst

@@ -8,6 +8,7 @@ History
 * Pulls `tune_time` from `wallClockRenderTime`
 * Adds `primary_hls` and `seconary_hls`
 * Adds quality selection
+* Overhauls time/datetime management
 
 0.2.4 (2021-08-15)
 ------------------

sxm/client.py

@@ -407,15 +407,20 @@ async def get_now_playing(self, channel: XMChannel) -> Union[Dict[str, Any], Non
             SXM channel to look up live channel data for
         """
 
+        now = time.time()
+        now_dt = datetime.datetime.fromtimestamp(now).replace(
+            tzinfo=datetime.timezone.utc
+        )
+
         params = {
             "assetGUID": channel.guid,
             "ccRequestType": "AUDIO_VIDEO",
             "channelId": channel.id,
             "hls_output_mode": "custom",
             "marker_mode": "all_separate_cue_points",
             "result-template": "web",
-            "time": str(int(round(time.time() * 1000.0))),
-            "timestamp": datetime.datetime.utcnow().isoformat("T") + "Z",
+            "time": str(int(round(now * 1000.0))),
+            "timestamp": now_dt.isoformat("T") + "Z",
         }
 
         return await self._get("tune/now-playing-live", params)
@@ -428,7 +433,7 @@ async def close_session(self):
     def reset_session(self) -> None:
         """Resets session used by client"""
 
-        self._session_start = time.time()
+        self._session_start = time.monotonic()
         self._session = httpx.AsyncClient()
         self._session.headers.update({"User-Agent": self._ua["string"]})
 
@@ -511,7 +516,7 @@ async def _request(
         method = method.upper()
 
         if authenticate:
-            now = time.time()
+            now = time.monotonic()
             if (now - self._session_start) > SESSION_MAX_LIFE:
                 self._log.info("Session exceed max time, reseting")
                 await self.close_session()
@@ -584,7 +589,7 @@ async def _get_playlist_url(
             self._log.info(f"No channel for {channel_id}")
             return None
 
-        now = time.time()
+        now = time.monotonic()
 
         if use_cache and channel.id in self._playlists:
             if (
@@ -651,7 +656,7 @@ async def _get_playlist_url(
         playlist = await self._get_playlist_variant_url(live_channel.primary_hls.url)
         if playlist is not None:
             self._playlists[channel.id] = playlist
-            self.last_renew = time.time()
+            self.last_renew = time.monotonic()
 
             if self.update_handler is not None:
                 self.update_handler(live_channel_raw)

sxm/models.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
-import datetime
-import time
+from datetime import datetime, timedelta, timezone
 from enum import Enum
 from typing import List, Optional, Tuple, Union
 
@@ -32,6 +31,16 @@
 LIVE_PRIMARY_HLS = "https://siriusxm-priprodlive.akamaized.net"
 
 
+def parse_xm_datetime(dt_string: str):
+    dt_string = dt_string.replace("+0000", "")
+    dt = datetime.fromisoformat(dt_string)
+    return dt.replace(tzinfo=timezone.utc)
+
+
+def parse_xm_timestamp(timestamp: int):
+    return datetime.utcfromtimestamp(timestamp / 1000).replace(tzinfo=timezone.utc)
+
+
 class QualitySize(str, Enum):
     SMALL_64k = "SMALL"
     MEDIUM_128k = "MEDIUM"
@@ -96,15 +105,15 @@ def from_dict(data: dict) -> XMCategory:
 
 class XMMarker(BaseModel):
     guid: str
-    time: int
-    duration: int
+    time: datetime
+    duration: timedelta
 
     @staticmethod
     def from_dict(data: dict) -> XMMarker:
         return XMMarker(
             guid=data["assetGUID"],
-            time=data["time"],
-            duration=data["duration"],
+            time=parse_xm_timestamp(data["time"]),
+            duration=timedelta(seconds=data["duration"]),
         )
 
 
@@ -162,8 +171,8 @@ class XMEpisodeMarker(XMMarker):
     def from_dict(data: dict) -> XMEpisodeMarker:
         return XMEpisodeMarker(
             guid=data["assetGUID"],
-            time=data["time"],
-            duration=data["duration"],
+            time=parse_xm_timestamp(data["time"]),
+            duration=timedelta(seconds=data["duration"]),
             episode=XMEpisode.from_dict(data["episode"]),
         )
 
@@ -250,23 +259,20 @@ def from_dict(data: dict) -> XMCutMarker:
 
         return XMCutMarker(
             guid=data["assetGUID"],
-            time=data["time"],
-            duration=data["duration"],
+            time=parse_xm_timestamp(data["time"]),
+            duration=timedelta(seconds=data["duration"]),
             cut=cut,
         )
 
 
 class XMPosition(BaseModel):
-    timestamp: datetime.datetime
+    timestamp: datetime
     position: str
 
     @staticmethod
     def from_dict(data: dict) -> XMPosition:
-        dt_string = data["timestamp"].replace("+0000", "")
-        dt = datetime.datetime.fromisoformat(dt_string)
-
         return XMPosition(
-            timestamp=dt.replace(tzinfo=datetime.timezone.utc),
+            timestamp=parse_xm_datetime(data["timestamp"]),
             position=data["position"],
         )
 
@@ -352,7 +358,7 @@ class XMLiveChannel(BaseModel):
     custom_hls_infos: List[XMHLSInfo]
     episode_markers: List[XMEpisodeMarker]
     cut_markers: List[XMCutMarker]
-    tune_time: Optional[int] = None
+    tune_time: Optional[datetime] = None
     # ... plus many unused
 
     _stream_quality: QualitySize = PrivateAttr(QualitySize.LARGE_256k)
@@ -374,10 +380,10 @@ def from_dict(data: dict) -> XMLiveChannel:
         )
 
         return XMLiveChannel(
-            id=data["channelId"],
+            id=data["moduleResponse"]["liveChannelData"]["channelId"],
             hls_infos=hls_infos,
             custom_hls_infos=custom_hls_infos,
-            tune_time=data["wallClockRenderTime"],
+            tune_time=parse_xm_datetime(data["wallClockRenderTime"]),
             episode_markers=episode_markers,
             cut_markers=cut_markers,
         )
@@ -475,7 +481,7 @@ def sort_markers(markers: List[XMMarker]) -> List[XMMarker]:
         return sorted(markers, key=lambda x: x.time)
 
     def _latest_marker(
-        self, marker_attr: str, now: Optional[int] = None
+        self, marker_attr: str, now: Optional[datetime] = None
     ) -> Union[XMMarker, None]:
         """Returns the latest `XMMarker` based on type relative to now"""
 
@@ -484,7 +490,7 @@ def _latest_marker(
             return None
 
         if now is None:
-            now = int(time.time() * 1000)
+            now = datetime.now(timezone.utc)
 
         latest = None
         for marker in markers:
@@ -495,7 +501,7 @@ def _latest_marker(
         return latest
 
     def get_latest_episode(
-        self, now: Optional[int] = None
+        self, now: Optional[datetime] = None
     ) -> Union[XMEpisodeMarker, None]:
         """Returns the latest :class:`XMEpisodeMarker` based
         on type relative to now
@@ -507,7 +513,9 @@ def get_latest_episode(
         """
         return self._latest_marker("episode_markers", now)  # type: ignore
 
-    def get_latest_cut(self, now: Optional[int] = None) -> Union[XMCutMarker, None]:
+    def get_latest_cut(
+        self, now: Optional[datetime] = None
+    ) -> Union[XMCutMarker, None]:
         """Returns the latest :class:`XMCutMarker` based
         on type relative to now
 

tests/test_models.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 from unittest.mock import patch
 
 from sxm.models import XMLiveChannel
@@ -46,14 +47,20 @@ def test_channels(sxm_client):
 def test_live_channel(sxm_client):
     data = sxm_client.get_now_playing("octane")
 
-    channel = XMLiveChannel.from_dict(
-        data["moduleList"]["modules"][0]["moduleResponse"]["liveChannelData"]
-    )
+    channel = XMLiveChannel.from_dict(data["moduleList"]["modules"][0])
 
     assert channel.id == "octane"
 
-    with patch("sxm.models.time") as mock_time:
-        mock_time.time.return_value = 1630000000
+    with patch("sxm.models.datetime") as mock_time:
+        mock_time.now.return_value = datetime(
+            year=2021,
+            month=8,
+            day=26,
+            hour=17,
+            minute=40,
+            second=40,
+            tzinfo=timezone.utc,
+        )
         cut = channel.get_latest_cut()
 
     assert cut.cut.title == "Ten Thousand Fists"