implement changes for dynamic power setting (#28)

Implements power control for the radios, both radios independent.

Reviewed-on: #28
Co-authored-by: pstruebi <struebin.patrick@gmail.com>
Co-committed-by: pstruebi <struebin.patrick@gmail.com>

src/auracast/auracast_config.py

@@ -1,5 +1,11 @@
 from typing import List
-from pydantic import BaseModel
+from pydantic import BaseModel, field_validator
+
+# Discrete TX power levels (dBm) supported by the Nordic SoftDevice Controller
+# for the nRF radio PA. The HCI controller will clamp requested values to the
+# nearest supported step. The maximum is bounded by CONFIG_BT_CTLR_TX_PWR_*
+# in the hci_uart firmware (currently +8 dBm).
+TX_POWER_VALID = [8, 7, 6, 5, 4, 3, 2, 0, -4, -8, -12, -16, -20]
 
 # Define some base to hold the relevant parameters
 class AuracastQoSConfig(BaseModel):
@@ -35,6 +41,19 @@ class AuracastGlobalConfig(BaseModel):
     # so receivers may render earlier than the presentation delay for lower latency.
     immediate_rendering: bool = False
     assisted_listening_stream: bool = False
+    # Bluetooth advertising TX power for this radio in dBm (per advertising set).
+    # Sent through HCI_LE_Set_Extended_Advertising_Parameters; the SDC clamps to
+    # nearest supported hardware step and propagates to primary/secondary adv,
+    # the periodic advertising train and the BIS ISO PDUs.
+    advertising_tx_power: int = 8
+
+    @field_validator('advertising_tx_power')
+    @classmethod
+    def _snap_tx_power(cls, v: int) -> int:
+        # Snap to the nearest supported discrete step in TX_POWER_VALID.
+        if v in TX_POWER_VALID:
+            return v
+        return min(TX_POWER_VALID, key=lambda s: abs(s - v))
 
 # "Audio input. "
 # "'device' -> use the host's default sound input device, "

src/auracast/multicast.py

@@ -49,6 +49,7 @@ import bumble.transport
 import bumble.utils
 from bumble.device import Host, AdvertisingChannelMap
 from bumble.audio import io as audio_io
+from bumble.vendor.zephyr.hci import HCI_Write_Tx_Power_Level_Command
 
 from auracast import auracast_config
 from auracast.utils.read_lc3_file import read_lc3_file
@@ -519,7 +520,17 @@ async def init_broadcast(
                 advertising_sid=i,
                 primary_advertising_phy=hci.Phy.LE_1M, # 2m phy config throws error - because for primary advertising channels, 1mbit is only supported
                 secondary_advertising_phy=hci.Phy.LE_1M, # this is the secondary advertising beeing send on non advertising channels (extendend advertising)
-                #advertising_tx_power= # tx power in dbm (max 20)
+                # Pass NO_PREFERENCE (0x7F) here for two reasons:
+                # 1. The Nordic SoftDevice Controller ignores this field for
+                #    advertising sets and always returns the compile-time
+                #    CONFIG_BT_CTLR_TX_PWR_* value. The real TX power is
+                #    applied via the Zephyr VS Write_Tx_Power_Level command
+                #    issued right after create_advertising_set() returns.
+                # 2. Bumble's HCI metadata declares this field as 1-byte
+                #    *unsigned* (a bumble bug — the BT spec defines it as
+                #    signed int8), so negative values would raise
+                #    "bytes must be in range(0, 256)" at serialization.
+                advertising_tx_power=hci.HCI_LE_Set_Extended_Advertising_Parameters_Command.TX_POWER_NO_PREFERENCE,
                 #secondary_advertising_max_skip=10,
             ),
             advertising_data=(
@@ -536,6 +547,48 @@ async def init_broadcast(
             auto_start=True,
         )
         bigs[f'big{i}']['advertising_set'] = advertising_set
+        # NOTE: selected_tx_power below reflects the SDC's compile-time max
+        # (LE_Set_Ext_Adv_Params was sent with NO_PREFERENCE). The actual
+        # transmit power is set by the VS Write_Tx_Power_Level call below.
+        logging.debug(
+            'LE_Set_Ext_Adv_Params reports controller fallback TX power: %+d dBm (handle=%d)',
+            getattr(advertising_set, 'selected_tx_power', 0),
+            i,
+        )
+
+        # The Nordic SoftDevice Controller does not honor the per-set
+        # advertising_tx_power passed in HCI_LE_Set_Extended_Advertising_Parameters
+        # (it returns the compile-time CONFIG_BT_CTLR_TX_PWR_* value regardless).
+        # Apply the requested level via the Zephyr Vendor-Specific HCI command
+        # Write_Tx_Power_Level (opcode 0xFC0E), which the SDC honors per
+        # advertising handle. The SDC clamps the value to the nearest supported
+        # hardware step (max bounded by CONFIG_BT_CTLR_TX_PWR_PLUS_8).
+        try:
+            adv_handle = getattr(advertising_set, 'advertising_handle', i)
+            response = await device.send_command(
+                HCI_Write_Tx_Power_Level_Command(
+                    handle_type=HCI_Write_Tx_Power_Level_Command.TX_POWER_HANDLE_TYPE_ADV,
+                    connection_handle=adv_handle,
+                    tx_power_level=global_config.advertising_tx_power,
+                )
+            )
+            rp = getattr(response, 'return_parameters', None)
+            status = getattr(rp, 'status', 0xFF) if rp is not None else 0xFF
+            selected = getattr(rp, 'selected_tx_power_level', None) if rp is not None else None
+            if status == 0 and selected is not None:
+                logging.info(
+                    'Advertising TX power (VS Write_Tx_Power_Level): requested=%+d dBm, controller selected=%+d dBm (handle=%d)',
+                    global_config.advertising_tx_power,
+                    selected,
+                    adv_handle,
+                )
+            else:
+                logging.warning(
+                    'VS Write_Tx_Power_Level failed: status=0x%02X handle=%d requested=%+d dBm',
+                    status, adv_handle, global_config.advertising_tx_power,
+                )
+        except Exception as e:
+            logging.warning('VS Write_Tx_Power_Level not supported by controller: %s', e)
 
         logging.info('Start Periodic Advertising')
         await advertising_set.start_periodic()

src/auracast/server/multicast_frontend.py

@@ -96,6 +96,36 @@ QOS_PRESET_MAP = {
     "Robust": auracast_config.AuracastQosRobust(),
 }
 
+# Discrete advertising TX power steps in dBm supported by the Nordic SDC radio
+# PA. Sent through HCI_LE_Set_Extended_Advertising_Parameters; the controller
+# clamps to the nearest hardware step.
+TX_POWER_OPTIONS = [8, 7, 6, 5, 4, 3, 2, 0, -4, -8, -12, -16, -20]
+TX_POWER_DEFAULT = 8
+
+
+def _coerce_tx_power(value, default: int = TX_POWER_DEFAULT) -> int:
+    try:
+        v = int(value)
+    except (TypeError, ValueError):
+        return default
+    if v in TX_POWER_OPTIONS:
+        return v
+    return min(TX_POWER_OPTIONS, key=lambda s: abs(s - v))
+
+
+def _tx_power_selectbox(label: str, key: str, default: int, disabled: bool, help_text: str | None = None) -> int:
+    snapped = _coerce_tx_power(default)
+    idx = TX_POWER_OPTIONS.index(snapped)
+    return st.selectbox(
+        label,
+        TX_POWER_OPTIONS,
+        index=idx,
+        key=key,
+        format_func=lambda v: f"{v:+d} dBm",
+        disabled=disabled,
+        help=help_text or "Bluetooth advertising TX power for this radio. Higher values increase range; lower values reduce interference and power draw.",
+    )
+
 # Try loading persisted settings from backend
 saved_settings = {}
 try:
@@ -394,6 +424,22 @@ if audio_mode == "Demo":
             disabled=is_streaming,
             help="Fast: 2 retransmissions, lower latency. Robust: 4 retransmissions, better reliability."
         )
+    # Per-radio TX power for Demo (independent for R1 and R2)
+    col_tx_r1, col_tx_r2 = st.columns(2, gap="small")
+    with col_tx_r1:
+        tx_power_r1 = _tx_power_selectbox(
+            "TX Power (R1)",
+            key="demo_tx_power_r1",
+            default=saved_settings.get('advertising_tx_power', TX_POWER_DEFAULT),
+            disabled=is_streaming,
+        )
+    with col_tx_r2:
+        tx_power_r2 = _tx_power_selectbox(
+            "TX Power (R2)",
+            key="demo_tx_power_r2",
+            default=saved_settings.get('secondary', {}).get('advertising_tx_power', TX_POWER_DEFAULT),
+            disabled=is_streaming,
+        )
     #st.info(f"Demo mode selected: {demo_selected} (Streams: {demo_stream_map[demo_selected]['streams']}, Rate: {demo_stream_map[demo_selected]['rate']} Hz)")
     quality = None  # Not used in demo mode
 else:
@@ -490,6 +536,13 @@ else:
                     help="Fast: 2 retransmissions, lower latency. Robust: 4 retransmissions, better reliability."
                 )
 
+            tx_power_r1 = _tx_power_selectbox(
+                "TX Power (R1)",
+                key="analog_tx_power_r1",
+                default=saved_settings.get('advertising_tx_power', TX_POWER_DEFAULT),
+                disabled=is_streaming,
+            )
+
             col_r1_name, col_r1_lang = st.columns([2, 1])
             with col_r1_name:
                 stream_name1 = st.text_input(
@@ -658,6 +711,13 @@ else:
                         help="Fast: 2 retransmissions, lower latency. Robust: 4 retransmissions, better reliability."
                     )
 
+                tx_power_r2 = _tx_power_selectbox(
+                    "TX Power (R2)",
+                    key="analog_tx_power_r2",
+                    default=saved_settings.get('secondary', {}).get('advertising_tx_power', TX_POWER_DEFAULT),
+                    disabled=is_streaming,
+                )
+
                 col_r2_name, col_r2_lang = st.columns([2, 1])
                 with col_r2_name:
                     stream_name2 = st.text_input(
@@ -713,6 +773,7 @@ else:
                     'immediate_rendering': immediate_rendering2,
                     'presentation_delay_ms': presentation_delay_ms2,
                     'qos_preset': qos_preset2,
+                    'tx_power': tx_power_r2,
                     'analog_gain_db_left': analog_gain_db_left,
                     'analog_gain_db_right': analog_gain_db_right,
                 }
@@ -729,6 +790,7 @@ else:
             'immediate_rendering': immediate_rendering1,
             'presentation_delay_ms': presentation_delay_ms1,
             'qos_preset': qos_preset1,
+            'tx_power': tx_power_r1,
             'stereo_mode': stereo_enabled,
             'analog_gain_db_left': analog_gain_db_left,
             'analog_gain_db_right': analog_gain_db_right,
@@ -907,7 +969,14 @@ else:
                     disabled=is_streaming,
                     help="Quality of Service preset for Radio 1"
                 )
-            
+
+            r1_tx_power = _tx_power_selectbox(
+                "TX Power (R1)",
+                key="dante_tx_power_r1",
+                default=saved_r1_config.get('advertising_tx_power', saved_settings.get('advertising_tx_power', TX_POWER_DEFAULT)),
+                disabled=is_streaming,
+            )
+
             # Per-stream configuration for Radio 1
             if dante_stereo_enabled:
                 st.write("**Stereo Stream Configuration (Radio 1)**")
@@ -1188,7 +1257,14 @@ else:
                         disabled=is_streaming,
                         help="Quality of Service preset for Radio 2"
                     )
-                
+
+                r2_tx_power = _tx_power_selectbox(
+                    "TX Power (R2)",
+                    key="dante_tx_power_r2",
+                    default=saved_r2_config.get('advertising_tx_power', saved_settings.get('secondary', {}).get('advertising_tx_power', TX_POWER_DEFAULT)),
+                    disabled=is_streaming,
+                )
+
                 # Per-stream configuration for Radio 2
                 st.write("**Stream Configuration (Radio 2)**")
                 r2_streams = []
@@ -1304,6 +1380,7 @@ else:
                 r2_immediate_rendering = False
                 r2_presentation_delay_ms = 40
                 r2_qos_preset = 'Fast'
+                r2_tx_power = TX_POWER_DEFAULT
         
         # Validate unique input devices for Network - Dante mode
         if audio_mode == "Network - Dante":
@@ -1335,6 +1412,7 @@ else:
             'immediate_rendering': r1_immediate_rendering,
             'presentation_delay_ms': r1_presentation_delay_ms,
             'qos_preset': r1_qos_preset,
+            'tx_power': r1_tx_power,
             'dante_stereo_mode': dante_stereo_enabled,
             'dante_stereo_left': dante_left_channel,
             'dante_stereo_right': dante_right_channel,
@@ -1350,6 +1428,7 @@ else:
             'immediate_rendering': r2_immediate_rendering if radio2_enabled else False,
             'presentation_delay_ms': r2_presentation_delay_ms if radio2_enabled else 40000,
             'qos_preset': r2_qos_preset if radio2_enabled else 'Fast',
+            'tx_power': r2_tx_power if radio2_enabled else TX_POWER_DEFAULT,
         } if radio2_enabled else None
         
     if audio_mode in ("USB", "Network"):
@@ -1406,6 +1485,13 @@ else:
                 help="Fast: 2 retransmissions, lower latency. Robust: 4 retransmissions, better reliability."
             )
 
+        tx_power = _tx_power_selectbox(
+            "TX Power",
+            key="usb_tx_power",
+            default=saved_settings.get('advertising_tx_power', TX_POWER_DEFAULT),
+            disabled=is_streaming,
+        )
+
         stream_name = st.text_input(
             "Channel Name",
             value=default_name,
@@ -1559,6 +1645,7 @@ if start_stream:
             immediate_rendering=immediate_rendering,
             presentation_delay_us=int(presentation_delay_ms * 1000),
             qos_config=QOS_PRESET_MAP[qos_preset],
+            advertising_tx_power=tx_power_r1,
             bigs=bigs1
         )
         config2 = None
@@ -1571,6 +1658,7 @@ if start_stream:
                 immediate_rendering=immediate_rendering,
                 presentation_delay_us=int(presentation_delay_ms * 1000),
                 qos_config=QOS_PRESET_MAP[qos_preset],
+                advertising_tx_power=tx_power_r2,
                 bigs=bigs2
             )
 
@@ -1614,6 +1702,7 @@ if start_stream:
                 immediate_rendering=bool(cfg['immediate_rendering']),
                 presentation_delay_us=int(cfg['presentation_delay_ms'] * 1000),
                 qos_config=QOS_PRESET_MAP[cfg['qos_preset']],
+                advertising_tx_power=int(cfg.get('tx_power', TX_POWER_DEFAULT)),
                 analog_gain_db_left=cfg.get('analog_gain_db_left', 0.0),
                 analog_gain_db_right=cfg.get('analog_gain_db_right', 0.0),
                 bigs=[
@@ -1701,6 +1790,7 @@ if start_stream:
                 immediate_rendering=bool(radio_cfg['immediate_rendering']),
                 presentation_delay_us=int(radio_cfg['presentation_delay_ms'] * 1000),
                 qos_config=QOS_PRESET_MAP[radio_cfg['qos_preset']],
+                advertising_tx_power=int(radio_cfg.get('tx_power', TX_POWER_DEFAULT)),
                 bigs=bigs
             )
 
@@ -1736,6 +1826,7 @@ if start_stream:
             immediate_rendering=immediate_rendering,
             presentation_delay_us=int(presentation_delay_ms * 1000),
             qos_config=QOS_PRESET_MAP[qos_preset],
+            advertising_tx_power=tx_power,
             bigs=[
                 auracast_config.AuracastBigConfig(
                     code=(stream_passwort.strip() or None),

src/auracast/server/multicast_server.py

@@ -585,6 +585,7 @@ async def init_radio(transport: str, conf: auracast_config.AuracastConfigGroup,
             'analog_stereo_mode': getattr(conf.bigs[0], 'analog_stereo_mode', False) if conf.bigs else False,
             'analog_gain_db_left': getattr(conf, 'analog_gain_db_left', 0.0),
             'analog_gain_db_right': getattr(conf, 'analog_gain_db_right', 0.0),
+            'advertising_tx_power': getattr(conf, 'advertising_tx_power', 8),
             'stream_password': (conf.bigs[0].code if conf.bigs and getattr(conf.bigs[0], 'code', None) else None),
             'big_ids': [getattr(big, 'id', DEFAULT_BIG_ID) for big in conf.bigs],
             'big_random_addresses': [getattr(big, 'random_address', DEFAULT_RANDOM_ADDRESS) for big in conf.bigs],
@@ -755,11 +756,12 @@ async def _autostart_from_settings():
         big_ids = settings.get('big_ids') or []
         big_addrs = settings.get('big_random_addresses') or []
         stream_password = settings.get('stream_password')
+        tx_power = int(settings.get('advertising_tx_power', 8))
         original_ts = settings.get('timestamp')
         previously_streaming = bool(settings.get('is_streaming'))
 
         log.info(
-            "[AUTOSTART][PRIMARY] loaded settings: previously_streaming=%s audio_mode=%s rate=%s octets=%s pres_delay=%s qos_preset=%s immediate_rendering=%s assisted_listening_stream=%s demo_sources=%s",
+            "[AUTOSTART][PRIMARY] loaded settings: previously_streaming=%s audio_mode=%s rate=%s octets=%s pres_delay=%s qos_preset=%s immediate_rendering=%s assisted_listening_stream=%s tx_power=%+d dBm demo_sources=%s",
             previously_streaming,
             audio_mode,
             rate,
@@ -768,6 +770,7 @@ async def _autostart_from_settings():
             saved_qos_preset,
             immediate_rendering,
             assisted_listening_stream,
+            tx_power,
             (settings.get('demo_sources') or []),
         )
 
@@ -817,6 +820,7 @@ async def _autostart_from_settings():
                 immediate_rendering=immediate_rendering,
                 assisted_listening_stream=assisted_listening_stream,
                 presentation_delay_us=pres_delay if pres_delay is not None else 40000,
+                advertising_tx_power=tx_power,
                 bigs=bigs,
             )
             # Set num_bis for stereo mode if needed
@@ -886,6 +890,7 @@ async def _autostart_from_settings():
                     presentation_delay_us=pres_delay if pres_delay is not None else 40000,
                     analog_gain_db_left=settings.get('analog_gain_db_left', 0.0),
                     analog_gain_db_right=settings.get('analog_gain_db_right', 0.0),
+                    advertising_tx_power=tx_power,
                     bigs=bigs,
                 )
                 # Set num_bis for stereo mode if needed
@@ -921,10 +926,11 @@ async def _autostart_from_settings():
         big_ids = settings.get('big_ids') or []
         big_addrs = settings.get('big_random_addresses') or []
         stream_password = settings.get('stream_password')
+        tx_power = int(settings.get('advertising_tx_power', 8))
         original_ts = settings.get('timestamp')
         previously_streaming = bool(settings.get('is_streaming'))
         log.info(
-            "[AUTOSTART][SECONDARY] loaded settings: previously_streaming=%s audio_mode=%s rate=%s octets=%s pres_delay=%s qos_preset=%s immediate_rendering=%s assisted_listening_stream=%s demo_sources=%s",
+            "[AUTOSTART][SECONDARY] loaded settings: previously_streaming=%s audio_mode=%s rate=%s octets=%s pres_delay=%s qos_preset=%s immediate_rendering=%s assisted_listening_stream=%s tx_power=%+d dBm demo_sources=%s",
             previously_streaming,
             audio_mode,
             rate,
@@ -933,6 +939,7 @@ async def _autostart_from_settings():
             saved_qos_preset,
             immediate_rendering,
             assisted_listening_stream,
+            tx_power,
             (settings.get('demo_sources') or []),
         )
         if not previously_streaming:
@@ -972,6 +979,7 @@ async def _autostart_from_settings():
                 immediate_rendering=immediate_rendering,
                 assisted_listening_stream=assisted_listening_stream,
                 presentation_delay_us=pres_delay if pres_delay is not None else 40000,
+                advertising_tx_power=tx_power,
                 bigs=bigs,
             )
             conf.qos_config = QOS_PRESET_MAP.get(saved_qos_preset, QOS_PRESET_MAP["Fast"])
@@ -1041,6 +1049,7 @@ async def _autostart_from_settings():
                     presentation_delay_us=pres_delay if pres_delay is not None else 40000,
                     analog_gain_db_left=settings.get('analog_gain_db_left', 0.0),
                     analog_gain_db_right=settings.get('analog_gain_db_right', 0.0),
+                    advertising_tx_power=tx_power,
                     bigs=bigs,
                 )
                 conf.qos_config = QOS_PRESET_MAP.get(saved_qos_preset, QOS_PRESET_MAP["Fast"])