diff --git a/app/device_manager.py b/app/device_manager.py index 7cb273c..a9330c1 100644 --- a/app/device_manager.py +++ b/app/device_manager.py @@ -49,6 +49,7 @@ class DeviceManager: self._self_info = None self._subscriptions = [] # active event subscriptions self._channel_secrets = {} # {channel_idx: secret_hex} for pkt_payload + self._max_channels = 8 # updated from device_info at connect self._pending_echo = None # {'timestamp': float, 'channel_idx': int, 'msg_id': int, 'pkt_payload': str|None} self._echo_lock = threading.Lock() @@ -167,6 +168,16 @@ class DeviceManager: self_info=json.dumps(self._self_info, default=str) ) + # Fetch device_info for max_channels + try: + dev_info_event = await self.mc.commands.send_device_query() + if dev_info_event and hasattr(dev_info_event, 'payload'): + dev_info = dev_info_event.payload or {} + self._max_channels = dev_info.get('max_channels', 8) + logger.info(f"Device max_channels: {self._max_channels}") + except Exception as e: + logger.warning(f"Could not fetch device_info: {e}") + logger.info(f"Connected to device: {self._device_name} " f"(key: {self._self_info.get('public_key', '?')[:8]}...)") @@ -196,7 +207,7 @@ class DeviceManager: async def _load_channel_secrets(self): """Load channel secrets from device for pkt_payload computation.""" try: - for idx in range(8): # MeshCore supports channels 0-7 + for idx in range(self._max_channels): event = await self.mc.commands.get_channel(idx) if event: data = getattr(event, 'payload', None) or {} diff --git a/app/main.py b/app/main.py index df65432..15def31 100644 --- a/app/main.py +++ b/app/main.py @@ -258,7 +258,7 @@ def _execute_console_command(args: list) -> str: elif cmd == 'channels': lines = [] - for i in range(8): + for i in range(device_manager._max_channels): ch = device_manager.get_channel_info(i) if ch and ch.get('name'): lines.append(f" [{i}] {ch['name']}") diff --git a/app/meshcore/cli.py b/app/meshcore/cli.py index 6bde8c7..4595c40 100644 --- a/app/meshcore/cli.py +++ b/app/meshcore/cli.py @@ -284,7 +284,7 @@ def get_channels() -> Tuple[bool, List[Dict]]: try: dm = _get_dm() channels = [] - for idx in range(8): + for idx in range(dm._max_channels): info = dm.get_channel_info(idx) if info and info.get('name'): channels.append({ @@ -302,8 +302,8 @@ def add_channel(name: str) -> Tuple[bool, str, Optional[str]]: """Add a new channel.""" try: dm = _get_dm() - # Find first free slot (1-7, slot 0 is Public) - for idx in range(1, 8): + # Find first free slot (1+, slot 0 is Public) + for idx in range(1, dm._max_channels): info = dm.get_channel_info(idx) if not info or not info.get('name'): result = dm.set_channel(idx, name)