Add configurable ads volume control feature

src/iSponsorBlockTV/config_setup.py

@@ -35,8 +35,13 @@
 )
 MUTE_ADS_PROMPT = "Do you want to mute native YouTube ads automatically? (y/N) "
 SKIP_ADS_PROMPT = "Do you want to skip native YouTube ads automatically? (y/N) "
+ADS_VOLUME_PROMPT = "Enter ads volume level (0-100, where 0 is muted and 100 is full volume): "
 AUTOPLAY_PROMPT = "Do you want to enable autoplay? (Y/n) "
 
+# Error message constants
+INVALID_VOLUME_INPUT = "Invalid input. Please enter a number between 0 and 100."
+VOLUME_RANGE_ERROR = "Volume must be between 0 and 100. Please try again."
+
 
 def get_yn_input(prompt):
     while choice := input(prompt):
@@ -46,6 +51,28 @@ def get_yn_input(prompt):
     return None
 
 
+def get_volume_input(prompt):
+    """
+    Prompts user for volume input and validates range (0-100)
+    Returns: Valid integer between 0-100
+    Raises: Continues prompting until valid input received
+    """
+    while True:
+        try:
+            user_input = input(prompt).strip()
+            if not user_input:
+                print(INVALID_VOLUME_INPUT)
+                continue
+
+            volume = int(user_input)
+            if 0 <= volume <= 100:
+                return volume
+            else:
+                print(VOLUME_RANGE_ERROR)
+        except ValueError:
+            print(INVALID_VOLUME_INPUT)
+
+
 async def create_web_session(use_proxy):
     return aiohttp.ClientSession(trust_env=use_proxy)
 
@@ -201,6 +228,12 @@ def main(config, debug: bool) -> None:
     choice = get_yn_input(MUTE_ADS_PROMPT)
     config.mute_ads = choice == "y"
 
+    # Ads volume configuration logic
+    if config.mute_ads:
+        config.ads_volume = 0
+    else:
+        config.ads_volume = get_volume_input(ADS_VOLUME_PROMPT)
+
     choice = get_yn_input(SKIP_ADS_PROMPT)
     config.skip_ads = choice == "y"
 

src/iSponsorBlockTV/helpers.py

@@ -45,6 +45,7 @@ def __init__(self, data_dir):
         self.auto_play = True
         self.join_name = "iSponsorBlockTV"
         self.use_proxy = False
+        self.ads_volume = 100  # Default ads volume to 100 (full volume)
         self.__load()
 
     def validate(self):
@@ -79,7 +80,7 @@ def __load(self):
                 for i in config:
                     if i not in config_file_blacklist_keys:
                         setattr(self, i, config[i])
-        except FileNotFoundError:
+        except (FileNotFoundError, json.JSONDecodeError):
             print("Could not load config file")
             # Create data directory if it doesn't exist (if we're not running in docker)
             if not os.path.exists(self.data_dir):

src/iSponsorBlockTV/ytlounge.py

@@ -36,12 +36,21 @@ def __init__(
         self.auto_play = True
         self.watchdog_running = False
         self.last_event_time = 0
+        self.ads_volume = 100
         if config:
             self.mute_ads = config.mute_ads
             self.skip_ads = config.skip_ads
             self.auto_play = config.auto_play
+            self.ads_volume = getattr(config, "ads_volume", 100)
         self._command_mutex = asyncio.Lock()
 
+    def should_handle_ads(self) -> bool:
+        """
+        Determine if ad handling should be active.
+        Returns True if either mute_ads is enabled OR ads_volume is less than 100.
+        """
+        return self.mute_ads or self.ads_volume < 100
+
     # Ensures that we still are subscribed to the lounge
     async def _watchdog(self):
         """
@@ -114,28 +123,29 @@ def _process_event(self, event_type: str, args: List[Any]):
             data = args[0]
             # print(data)
             # Unmute when the video starts playing
-            if self.mute_ads and data["state"] == "1":
-                create_task(self.mute(False, override=True))
+            if self.should_handle_ads() and data["state"] == "1":
+                create_task(self.handle_ads_end())
         elif event_type == "nowPlaying":
             data = args[0]
-            # Unmute when the video starts playing
-            if self.mute_ads and data.get("state", "0") == "1":
-                self.logger.info("Ad has ended, unmuting")
-                create_task(self.mute(False, override=True))
+            # Handle ad end when video starts playing
+            if self.should_handle_ads() and data.get("state", "0") == "1":
+                create_task(self.handle_ads_end())
         elif event_type == "onAdStateChange":
             data = args[0]
             if data["adState"] == "0" and data["currentTime"] != "0":  # Ad is not playing
-                self.logger.info("Ad has ended, unmuting")
-                create_task(self.mute(False, override=True))
-            elif (
-                self.skip_ads and data["isSkipEnabled"] == "true"
-            ):  # YouTube uses strings for booleans
-                self.logger.info("Ad can be skipped, skipping")
-                create_task(self.skip_ad())
-                create_task(self.mute(False, override=True))
-            elif self.mute_ads:  # Seen multiple other adStates, assuming they are all ads
-                self.logger.info("Ad has started, muting")
-                create_task(self.mute(True, override=True))
+                # Ad has ended - handle volume restoration or unmuting
+                if self.should_handle_ads():
+                    create_task(self.handle_ads_end())
+            elif data["adState"] != "0":  # Ad is playing
+                if (
+                    self.skip_ads and data["isSkipEnabled"] == "true"
+                ):  # YouTube uses strings for booleans
+                    self.logger.info("Ad can be skipped, skipping")
+                    create_task(self.skip_ad())
+                    if self.should_handle_ads():
+                        create_task(self.handle_ads_end())
+                elif self.should_handle_ads():  # Ad is playing - handle muting or volume control
+                    create_task(self.handle_ads_start())
         # Manages volume, useful since YouTube wants to know the volume
         # when unmuting (even if they already have it)
         elif event_type == "onVolumeChanged":
@@ -159,10 +169,10 @@ def _process_event(self, event_type: str, args: List[Any]):
             ):  # YouTube uses strings for booleans
                 self.logger.info("Ad can be skipped, skipping")
                 create_task(self.skip_ad())
-                create_task(self.mute(False, override=True))
-            elif self.mute_ads:  # Seen multiple other adStates, assuming they are all ads
-                self.logger.info("Ad has started, muting")
-                create_task(self.mute(True, override=True))
+                if self.should_handle_ads():
+                    create_task(self.handle_ads_end())
+            elif self.should_handle_ads():  # Handle ad start for muting or volume control
+                create_task(self.handle_ads_start())
 
         elif event_type == "loungeStatus":
             data = args[0]
@@ -199,6 +209,41 @@ def _process_event(self, event_type: str, args: List[Any]):
     async def set_volume(self, volume: int) -> None:
         await self._command("setVolume", {"volume": volume})
 
+    async def handle_ads_start(self) -> None:
+        """
+        Handle ad start: either mute the ad or set configured ads volume.
+        """
+        try:
+            if self.mute_ads:
+                # Mute the ad completely
+                self.logger.info("Ad has started, muting")
+                await self.mute(True, override=True)
+            elif self.ads_volume < 100:
+                # Set configured ads volume (only if ads_volume < 100)
+                self.logger.info(f"Ad has started, setting volume to {self.ads_volume}%")
+                await self.set_volume(self.ads_volume)
+        except Exception as e:
+            self.logger.error(f"Failed to handle ad start: {e}")
+            # Don't re-raise the exception to avoid disrupting ad handling
+
+    async def handle_ads_end(self) -> None:
+        """
+        Handle ad end: restore original volume or unmute.
+        """
+        try:
+            if self.mute_ads:
+                # Unmute when ad ends
+                self.logger.info("Ad has ended, unmuting")
+                await self.mute(False, override=True)
+            elif self.ads_volume < 100:
+                # Restore original volume when ad ends (only if we changed it)
+                original_volume = self.volume_state.get("volume") or 100
+                self.logger.info(f"Ad has ended, restoring volume to {original_volume}%")
+                await self.set_volume(original_volume)
+        except Exception as e:
+            self.logger.error(f"Failed to handle ad end: {e}")
+            # Don't re-raise the exception to avoid disrupting ad handling
+
     async def mute(self, mute: bool, override: bool = False) -> None:
         """
         Mute or unmute the device (if the device already