Add ad_volume option to lower volume during ads instead of muting

config.json.template

@@ -11,6 +11,7 @@
     ],
     "skip_count_tracking": true,
     "mute_ads": true,
+    "ad_volume": null,
     "skip_ads": true,
     "minimum_skip_length": 1,
     "auto_play": true,

src/iSponsorBlockTV/config_setup.py

@@ -35,6 +35,10 @@
     " UUID will be sent? (Y/n) "
 )
 MUTE_ADS_PROMPT = "Do you want to mute native YouTube ads automatically? (y/N) "
+AD_VOLUME_PROMPT = (
+    "Enter the volume to set during ads (0-100), or leave empty to mute instead."
+    " Useful if muting an ad is too abrupt: "
+)
 SKIP_ADS_PROMPT = "Do you want to skip native YouTube ads automatically? (y/N) "
 AUTOPLAY_PROMPT = "Do you want to enable autoplay? (Y/n) "
 ENTER_SPONSORBLOCK_API_PROMPT = f"Enter SponsorBlock API URL (default: {SponsorBlock_api}): "
@@ -203,6 +207,17 @@ def main(config, debug: bool) -> None:
     choice = get_yn_input(MUTE_ADS_PROMPT)
     config.mute_ads = choice == "y"
 
+    if config.mute_ads:
+        while True:
+            ad_volume = input(AD_VOLUME_PROMPT).strip()
+            if not ad_volume:
+                config.ad_volume = None
+                break
+            if ad_volume.isdigit() and int(ad_volume) <= 100:
+                config.ad_volume = int(ad_volume)
+                break
+            print("You entered an invalid volume, try again.")
+
     choice = get_yn_input(SKIP_ADS_PROMPT)
     config.skip_ads = choice == "y"
 

src/iSponsorBlockTV/helpers.py

@@ -49,6 +49,7 @@ def __init__(self, data_dir):
         self.channel_whitelist = []
         self.skip_count_tracking = True
         self.mute_ads = False
+        self.ad_volume = None  # None = mute ads instead of lowering the volume
         self.skip_ads = False
         self.minimum_skip_length = 1
         self.auto_play = True
@@ -78,6 +79,10 @@ def validate(self):
         self.devices = [Device(i) for i in self.devices]
         if not self.apikey and self.channel_whitelist:
             raise ValueError("No youtube API key found and channel whitelist is not empty")
+        if self.ad_volume is not None and not (
+            isinstance(self.ad_volume, int) and 0 <= self.ad_volume <= 100
+        ):
+            raise ValueError("ad_volume must be a number between 0 and 100")
         if self.skip_categories is None:
             self.skip_categories = ["sponsor"]
             print("No categories found, using default: sponsor")

src/iSponsorBlockTV/setup_wizard.py

@@ -938,11 +938,40 @@ def compose(self) -> ComposeResult:
                 id="mute-ads-switch",
                 label="Enable muting ads",
             )
+        yield Label(
+            (
+                "Muting an ad cuts the sound dead. Set a volume below to lower the"
+                " volume during ads instead of muting; the previous volume is"
+                " restored when the ad ends."
+            ),
+            classes="subtitle",
+        )
+        yield Input(
+            placeholder="Ad volume 0-100 (empty to mute instead)",
+            id="ad-volume-input",
+            value="" if self.config.ad_volume is None else str(self.config.ad_volume),
+            validators=[
+                Function(
+                    lambda user_input: (
+                        not user_input or (user_input.isdigit() and int(user_input) <= 100)
+                    ),
+                    "Please enter a volume between 0 and 100",
+                )
+            ],
+        )
 
     @on(Checkbox.Changed, "#mute-ads-switch")
     def changed_mute(self, event: Checkbox.Changed):
         self.config.mute_ads = event.checkbox.value
 
+    @on(Input.Changed, "#ad-volume-input")
+    def changed_ad_volume(self, event: Input.Changed):
+        value = event.input.value.strip()
+        if value.isdigit() and int(value) <= 100:
+            self.config.ad_volume = int(value)
+        else:
+            self.config.ad_volume = None
+
     @on(Checkbox.Changed, "#skip-ads-switch")
     def changed_skip(self, event: Checkbox.Changed):
         self.config.skip_ads = event.checkbox.value

src/iSponsorBlockTV/ytlounge.py

@@ -55,6 +55,9 @@ def __init__(
         self.auth.lounge_id_token = None
         self.api_helper = api_helper
         self.volume_state = {}
+        self.ad_volume = None
+        self.content_volume = None
+        self.volume_lowered = False
         self.playback_speed = 1.0
         self.subscribe_task = None
         self.subscribe_task_watchdog = None
@@ -68,6 +71,7 @@ def __init__(
         self.last_event_time = 0
         if config:
             self.mute_ads = config.mute_ads
+            self.ad_volume = config.ad_volume
             self.skip_ads = config.skip_ads
             self.auto_play = config.auto_play
         self._command_mutex = asyncio.Lock()
@@ -188,6 +192,7 @@ async def _process_event(self, event_type: str, args: List[Any]):
         # when unmuting (even if they already have it)
         elif event_type == "onVolumeChanged":
             self.volume_state = args[0]
+            self._remember_content_volume(args[0].get("volume"))
         # Gets segments for the next video before it starts playing
         elif event_type == "autoplayUpNext":
             if len(args) > 0 and (vid_id := args[0]["videoId"]):  # if video id is not empty
@@ -222,6 +227,10 @@ async def _process_event(self, event_type: str, args: List[Any]):
                         self._sid = None
                         self._gsession = None  # Force disconnect
                         return
+            # The device only reports its volume when something changes it, so
+            # ask now: without it the first ad has no volume to come back to.
+            if self.ad_volume is not None and self.content_volume is None:
+                create_task(self.get_volume())
 
         elif event_type == "onSubtitlesTrackChanged":
             if self.shorts_disconnected:
@@ -247,6 +256,10 @@ async def _process_event(self, event_type: str, args: List[Any]):
     async def set_volume(self, volume: int) -> None:
         await self._command("setVolume", {"volume": volume})
 
+    # Ask the device for its volume, answered with an onVolumeChanged event
+    async def get_volume(self) -> bool:
+        return await self._command("getVolume")
+
     async def mute(self, mute: bool, override: bool = False) -> None:
         """
         Mute or unmute the device (if the device already
@@ -258,6 +271,19 @@ async def mute(self, mute: bool, override: bool = False) -> None:
 
         TODO: Only works if the device is subscribed to the lounge
         """
+        # Falls through to the muted flag while the device volume is still
+        # unknown, as there would be nothing to restore once the ad ends.
+        if self.ad_volume is not None:
+            if mute and self.content_volume is not None:
+                await self._duck_volume(True)
+                return
+            if not mute and self.volume_lowered:
+                await self._duck_volume(False)
+                return
+            # Playback resumes on every onStateChange: with nothing lowered and
+            # nothing muted, resending volume_state could replay a stale echo.
+            if not mute and self.volume_state.get("muted") != "true":
+                return
         if mute:
             mute_str = "true"
         else:
@@ -270,6 +296,52 @@ async def mute(self, mute: bool, override: bool = False) -> None:
                 {"volume": self.volume_state.get("volume", 100), "muted": mute_str},
             )
 
+    async def _duck_volume(self, mute: bool) -> None:
+        """
+        Silence an ad by lowering the volume to `ad_volume` and restore the
+        previous volume afterwards, leaving the muted flag alone.
+
+        Used instead of muting when `ad_volume` is set, for when cutting the
+        sound dead is more jarring than the ad and a barely audible floor is
+        preferred. Only called by `mute`, which is what checks there is a volume
+        to lower and to restore.
+
+        The volume restored is `content_volume`, kept up to date by
+        `_remember_content_volume`, never a snapshot taken when the ad starts:
+        back to back ads let the device's echo of the lowered volume land
+        between two ads, and a snapshot would then save the ad volume as the
+        one to come back to.
+
+        :param bool mute: True when an ad starts, False when it ends
+        """
+        if mute:
+            if self.volume_lowered:
+                return
+            self.volume_lowered = True
+            volume = self.ad_volume
+        else:
+            self.volume_lowered = False
+            volume = self.content_volume
+        self.volume_state["volume"] = volume
+        self.volume_state["muted"] = "false"
+        self.logger.info("Setting volume to %s", volume)
+        # Clears a muted flag left by an ad seen before the volume was known
+        await self._command("setVolume", {"volume": volume, "muted": "false"})
+
+    def _remember_content_volume(self, volume) -> None:
+        """
+        Keep the volume to restore after an ad from what the device reports.
+
+        The device echoes every setVolume, late and in any order relative to
+        ad events, so a report is the user's volume only when no ad is lowering
+        it and it differs from `ad_volume`. A user volume exactly equal to
+        `ad_volume` is therefore never learnt.
+        """
+        if self.ad_volume is None or self.volume_lowered or volume is None:
+            return
+        if int(volume) != self.ad_volume:
+            self.content_volume = volume
+
     async def play_video(self, video_id: str) -> bool:
         return await self._command("setPlaylist", {"videoId": video_id})
 

tests/test_ad_volume.py

@@ -0,0 +1,144 @@
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from iSponsorBlockTV.ytlounge import YtLoungeApi
+
+AD_VOLUME = 4
+
+
+class FakeLogger:
+    def __getattr__(self, name):
+        return lambda *args, **kwargs: None
+
+
+@pytest.fixture
+def lounge():
+    config = SimpleNamespace(
+        join_name="test",
+        mute_ads=True,
+        ad_volume=AD_VOLUME,
+        skip_ads=True,
+        auto_play=True,
+    )
+    api = YtLoungeApi("screen", config, api_helper=None, logger=FakeLogger())
+    api.sent = []
+
+    async def fake_command(command, command_parameters=None):
+        api.sent.append((command, command_parameters))
+        return True
+
+    api._command = fake_command
+    return api
+
+
+async def event(api, event_type, payload):
+    await api._process_event(event_type, [payload])
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+
+async def ad_starts(api):
+    await event(
+        api, "onAdStateChange", {"adState": "1", "currentTime": "0", "isSkipEnabled": "false"}
+    )
+
+
+async def ad_ends(api):
+    await event(api, "onStateChange", {"state": "1", "currentTime": "10", "duration": "100"})
+
+
+async def device_reports(api, volume):
+    await event(api, "onVolumeChanged", {"volume": str(volume), "muted": "false"})
+
+
+def volumes_sent(api):
+    return [params["volume"] for command, params in api.sent if command == "setVolume"]
+
+
+@pytest.mark.asyncio
+async def test_ad_is_lowered_then_user_volume_restored(lounge):
+    await device_reports(lounge, 100)
+    await ad_starts(lounge)
+    await ad_ends(lounge)
+    assert volumes_sent(lounge) == [AD_VOLUME, "100"]
+
+
+@pytest.mark.asyncio
+async def test_late_echo_of_ad_volume_does_not_become_restore_point(lounge):
+    # Sequence seen on 2026-09-27 23:17:12: two ads back to back, the device's
+    # echo of the first lowering lands after the restore and before the second ad.
+    await device_reports(lounge, 100)
+    await ad_starts(lounge)
+    await ad_ends(lounge)
+    await device_reports(lounge, AD_VOLUME)
+    await ad_starts(lounge)
+    await ad_ends(lounge)
+    assert volumes_sent(lounge)[-1] == "100"
+
+
+@pytest.mark.asyncio
+async def test_volume_changed_by_user_between_ads_is_restored(lounge):
+    await device_reports(lounge, 100)
+    await ad_starts(lounge)
+    await ad_ends(lounge)
+    await device_reports(lounge, 60)
+    await ad_starts(lounge)
+    await ad_ends(lounge)
+    assert volumes_sent(lounge)[-1] == "60"
+
+
+@pytest.mark.asyncio
+async def test_repeated_ad_start_events_do_not_move_restore_point(lounge):
+    await device_reports(lounge, 100)
+    await ad_starts(lounge)
+    await device_reports(lounge, AD_VOLUME)
+    await ad_starts(lounge)
+    await ad_ends(lounge)
+    assert volumes_sent(lounge)[-1] == "100"
+
+
+@pytest.mark.asyncio
+async def test_unknown_volume_falls_back_to_muted_flag(lounge):
+    await ad_starts(lounge)
+    assert lounge.sent[-1] == ("setVolume", {"volume": 100, "muted": "true"})
+
+
+@pytest.mark.asyncio
+async def test_device_already_at_ad_volume_on_connect_is_not_a_restore_point(lounge):
+    # Left at the ad volume by a crash mid-ad: restoring to it would stick forever.
+    await device_reports(lounge, AD_VOLUME)
+    await ad_starts(lounge)
+    assert lounge.sent[-1][1].get("muted") == "true"
+
+
+@pytest.mark.asyncio
+async def test_playback_resuming_outside_an_ad_does_not_resend_a_stale_volume(lounge):
+    await device_reports(lounge, 100)
+    await ad_starts(lounge)
+    await ad_ends(lounge)
+    await device_reports(lounge, AD_VOLUME)
+    sent_before = len(lounge.sent)
+    await ad_ends(lounge)
+    assert lounge.sent[sent_before:] == []
+
+
+@pytest.mark.asyncio
+async def test_muted_fallback_is_undone_once_volume_becomes_known(lounge):
+    await ad_starts(lounge)
+    await event(lounge, "onVolumeChanged", {"volume": "100", "muted": "true"})
+    await ad_ends(lounge)
+    assert lounge.sent[-1] == ("setVolume", {"volume": "100", "muted": "false"})
+
+
+@pytest.mark.asyncio
+async def test_lowering_clears_a_muted_flag_left_by_the_fallback(lounge):
+    await ad_starts(lounge)
+    await event(lounge, "onVolumeChanged", {"volume": "100", "muted": "true"})
+    await ad_starts(lounge)
+    await ad_ends(lounge)
+    assert lounge.sent[-2:] == [
+        ("setVolume", {"volume": AD_VOLUME, "muted": "false"}),
+        ("setVolume", {"volume": "100", "muted": "false"}),
+    ]