branding; bugs; tests

Author: aw-was-here

nowplaying/inputs/remote.py

@@ -16,6 +16,7 @@
     import nowplaying.config
     from PySide6.QtCore import QSettings
 
+
 class Plugin(InputPlugin):  # pylint: disable=too-many-instance-attributes
     ''' handler for NowPlaying '''
 
@@ -26,8 +27,8 @@ def __init__(self,
         self.displayname = "Remote"
         # Set default path
         default_path = pathlib.Path(
-                QStandardPaths.standardLocations(
-                    QStandardPaths.CacheLocation)[0]).joinpath('remotedb').joinpath("remote.db")
+            QStandardPaths.standardLocations(
+                QStandardPaths.CacheLocation)[0]).joinpath('remotedb').joinpath("remote.db")
 
         # Use configured path if available, otherwise use default
         if self.config and self.config.cparser.value('remote/remotedb'):

nowplaying/inputs/stagelinq.py

@@ -34,32 +34,45 @@ class DeckInfo():
     playing: bool = False
 
     def __post_init__(self):
+        """Set updated timestamp if not provided."""
         if not self.updated:
             self.updated = datetime.datetime.now(tz=datetime.timezone.utc)
 
     def __lt__(self, other: "DeckInfo") -> bool:
+        """Compare DeckInfo instances by updated timestamp for sorting."""
         return self.updated < other.updated
 
     def copy(self) -> "DeckInfo":
+        """Create a copy of this DeckInfo instance."""
         return dataclasses.replace(self)
 
     def same_content(self, other: "DeckInfo") -> bool:
+        """Check if this deck has the same track and artist as another deck."""
         if self.track == other.track and self.artist == other.artist:
             return True
         return False
 
 
 class StagelinqHandler():
-    """ stagelinq server """
+    """ StagelinQ server """
 
     def __init__(self, event: asyncio.Event):
+        """Initialize the StagelinQ handler.
+        
+        Args:
+            event: Asyncio event used to signal shutdown
+        """
         self.event = event
         self.device: AsyncDevice | None = None
         self.loop_task: asyncio.Task | None = None
         self.decks: dict[int, DeckInfo] = {}
 
     async def get_device(self):
-        """ find a device """
+        """Discover and connect to a StagelinQ device.
+        
+        Continuously searches for StagelinQ devices until one is found
+        or the event is set to stop searching.
+        """
         config = DiscoveryConfig(discovery_timeout=3.0)
         while not self.event.is_set() and self.device is None:
             try:
@@ -80,6 +93,12 @@ async def get_device(self):
                 logging.exception("Discovery error: %s", e)
 
     async def loop(self):
+        """Main loop that maintains connection to StagelinQ device and processes updates.
+        
+        This method handles device discovery, connection management, and subscribes
+        to track state updates from all decks. It automatically reconnects if
+        the connection is lost.
+        """
 
         # Connect to device with retry loop
         while not self.event.is_set():
@@ -139,27 +158,46 @@ async def loop(self):
                 await asyncio.sleep(5)
 
     def process_state_update(self, temp_decks: dict[int, DeckInfo], state: "State"):
-        """Process a state update and update deck information."""
+        """Process a state update and update deck information.
+        
+        Args:
+            temp_decks: Dictionary of deck information being built during this update cycle
+            state: StagelinQ state object containing updated deck information
+        """
         deck_num = next((i for i in range(1, 5) if f"Deck{i}" in state.name), None)
         if deck_num is None:
             return
 
+        # Initialize deck if it doesn't exist
+        if deck_num not in temp_decks:
+            temp_decks[deck_num] = DeckInfo(updated=datetime.datetime.now(tz=datetime.timezone.utc))
+
         # Update deck information based on state type using typed values
         if "ArtistName" in state.name:
             temp_decks[deck_num].artist = state.get_typed_value() or ""
         elif "SongName" in state.name:
             temp_decks[deck_num].track = state.get_typed_value() or ""
         elif "CurrentBPM" in state.name:
             # BPM values are already properly typed as float
-            temp_decks[deck_num].bpm = state.get_typed_value() or 0.0
+            temp_decks[deck_num].bpm = state.get_typed_value()
         elif "PlayState" in state.name:
             # Boolean states are already properly typed
             temp_decks[deck_num].playing = state.get_typed_value()
 
     def update_current_tracks(self, temp_decks: dict[int, DeckInfo]):
+        """Update the current deck states with new information.
+        
+        Args:
+            temp_decks: Dictionary of updated deck information from the latest state updates
+        """
         this_update = datetime.datetime.now(tz=datetime.timezone.utc)
         for deck_num in range(1, 5):
-            if not temp_decks[deck_num].playing and self.decks[deck_num]:
+            # If deck doesn't exist in temp_decks, remove it from self.decks
+            if deck_num not in temp_decks:
+                self.decks.pop(deck_num, None)
+                continue
+                
+            if not temp_decks[deck_num].playing and deck_num in self.decks:
                 del self.decks[deck_num]
             elif self.decks.get(deck_num) is None:
                 self.decks[deck_num] = temp_decks[deck_num].copy()
@@ -169,15 +207,25 @@ def update_current_tracks(self, temp_decks: dict[int, DeckInfo]):
                 self.decks[deck_num].updated = this_update
 
     async def start(self):
+        """Start the StagelinQ handler by creating the main loop task."""
         self.loop_task = asyncio.create_task(self.loop())
 
     async def stop(self):
+        """Stop the StagelinQ handler and cancel the main loop task."""
         self.event.set()
-        logging.info("Shutting down Stagelinq")
+        logging.info("Shutting down StagelinQ")
         if self.loop_task:
             self.loop_task.cancel()
 
     async def get_track(self, mixmode: str) -> DeckInfo | None:
+        """Get the currently playing track based on the specified mix mode.
+        
+        Args:
+            mixmode: Either "newest" or "oldest" to determine which deck to return
+            
+        Returns:
+            DeckInfo for the selected deck, or None if no decks are playing
+        """
         sorted_decks = sorted(self.decks.values(), key=lambda deck: deck.updated)
 
         if not sorted_decks:
@@ -194,7 +242,7 @@ def __init__(self,
                  config: "nowplaying.config.ConfigFile | None" = None,
                  qsettings: "QWidget | None" = None):
         super().__init__(config=config, qsettings=qsettings)
-        self.displayname = "Stagelinq"
+        self.displayname = "StagelinQ"
         self.url: str | None = None
         self.mixmode = "newest"
         self.testmode = False
@@ -204,67 +252,84 @@ def __init__(self,
 #### Additional UI method
 
     def desc_settingsui(self, qwidget: "QWidget") -> None:  # pylint: disable=no-self-use
-        ''' description of this input '''
-        qwidget.setText('Denon Stagelinq compatible equipment')
+        """Set the description text for the settings UI."""
+        qwidget.setText('Denon StagelinQ compatible equipment')
 
 #### Autoinstallation methods ####
 
     def install(self) -> bool:  # pylint: disable=no-self-use
-        ''' if a fresh install, run this '''
+        """Install method for fresh installations. Not required for StagelinQ."""
         return False
 
 #### Mix Mode menu item methods
 
     def validmixmodes(self) -> list[str]:  # pylint: disable=no-self-use
-        ''' tell ui valid mixmodes '''
+        """Return the list of valid mix modes for the UI."""
         return ['newest', 'oldest']
 
     def setmixmode(self, mixmode: str) -> str:  # pylint: disable=no-self-use, unused-argument
-        ''' handle user switching the mix mode: TBD '''
+        """Set the mix mode for determining which deck to use.
+        
+        Args:
+            mixmode: Either "newest" or "oldest"
+            
+        Returns:
+            The validated mix mode that was set
+        """
         if mixmode not in ['newest', 'oldest']:
             mixmode = self.config.cparser.value('stagelinq/mixmode')
 
         self.config.cparser.setValue('stagelinq/mixmode', mixmode)
         return mixmode
 
     def getmixmode(self) -> str:  # pylint: disable=no-self-use
-        ''' return what the current mixmode is set to '''
+        """Get the current mix mode setting."""
         return self.config.cparser.value('stagelinq/mixmode')
 
 #### Data feed methods
 
     async def getplayingtrack(self) -> TrackMetadata | None:
-        ''' Get the currently playing track '''
+        """Get the currently playing track metadata.
+        
+        Returns:
+            Dictionary containing track metadata (artist, track, bpm) or None if no handler
+        """
         if not self.handler:
             return None
 
         deck = await self.handler.get_track(mixmode=self.mixmode)
         metadata: TrackMetadata = {}
         if not deck:
             return metadata
-        if deck.artist:
+        if deck.artist is not None:
             metadata["artist"] = deck.artist
-        if deck.track:
+        if deck.track is not None:
             metadata["track"] = deck.track
-        if deck.bpm:
+        if deck.bpm is not None:
             metadata["bpm"] = str(deck.bpm)
         return metadata
 
     async def getrandomtrack(self, playlist: str) -> str | None:
-        ''' Get a file associated with a playlist, crate, whatever '''
+        """Get a random track from a playlist.
+        
+        Args:
+            playlist: Name of the playlist (not implemented for StagelinQ)
+            
+        Raises:
+            NotImplementedError: This method is not supported for StagelinQ
+        """
         raise NotImplementedError
 
 
 #### Control methods
 
     async def start(self) -> None:
-        ''' any initialization before actual polling starts '''
+        """Initialize the StagelinQ handler and start listening for devices."""
         self.handler = StagelinqHandler(event=self.event)
         await self.handler.start()
 
     async def stop(self) -> None:
-        ''' stopping either the entire program or just this
-            input '''
+        """Stop the StagelinQ handler and clean up resources."""
+        self.event.set()
         if self.handler:
-            self.event.set()
             await self.handler.stop()