Add channel commands and fix a lint error

This commit is contained in:
Alex Wolden 2025-06-01 20:30:19 -07:00
parent d73979f234
commit cca0ca90e9
6 changed files with 192 additions and 6 deletions

View file

@ -382,6 +382,25 @@ class CommandHandler:
data = b"\x32" + cmd.encode('utf-8')
return await self.send(data, [EventType.CLI_RESPONSE, EventType.ERROR])
async def get_channel(self, channel_idx: int) -> Event:
logger.debug(f"Getting channel info for channel {channel_idx}")
data = b"\x1f" + channel_idx.to_bytes(1, 'little')
return await self.send(data, [EventType.CHANNEL_INFO, EventType.ERROR])
async def set_channel(self, channel_idx: int, channel_name: str, channel_secret: bytes) -> Event:
logger.debug(f"Setting channel {channel_idx}: name={channel_name}")
# Pad channel name to 32 bytes
name_bytes = channel_name.encode('utf-8')[:32]
name_bytes = name_bytes.ljust(32, b'\x00')
# Ensure channel secret is exactly 16 bytes
if len(channel_secret) != 16:
raise ValueError("Channel secret must be exactly 16 bytes")
data = b"\x20" + channel_idx.to_bytes(1, 'little') + name_bytes + channel_secret
return await self.send(data, [EventType.OK, EventType.ERROR])
async def send_trace(self, auth_code: int = 0, tag: Optional[int] = None,
flags: int = 0, path: Optional[Union[str, bytes, bytearray]] = None) -> Event:
"""

View file

@ -35,6 +35,7 @@ class EventType(Enum):
RX_LOG_DATA = "rx_log_data"
TELEMETRY_RESPONSE = "telemetry_response"
CUSTOM_VARS = "custom_vars"
CHANNEL_INFO = "channel_info"
# Command response types
OK = "command_ok"

View file

@ -218,7 +218,23 @@ class MessageReader:
psplit = p.split(":")
res[psplit[0]] = psplit[1]
logger.debug(f"got custom vars : {res}")
await self.dispatcher.dispatch(Event(EventType.CUSTOM_VARS, res))
await self.dispatcher.dispatch(Event(EventType.CUSTOM_VARS, res))
elif packet_type_value == PacketType.CHANNEL_INFO.value:
logger.debug(f"received channel info response: {data.hex()}")
res = {}
res["channel_idx"] = data[1]
# Channel name is null-terminated, so find the first null byte
name_bytes = data[2:34]
null_pos = name_bytes.find(0)
if null_pos >= 0:
res["channel_name"] = name_bytes[:null_pos].decode('utf-8')
else:
res["channel_name"] = name_bytes.decode('utf-8')
res["channel_secret"] = data[34:50]
await self.dispatcher.dispatch(Event(EventType.CHANNEL_INFO, res, res))
# Push notifications
elif packet_type_value == PacketType.ADVERTISEMENT.value:
@ -404,13 +420,13 @@ class MessageReader:
"""Parse a given byte string and return as a LppFrame object."""
i = 0
data = []
lpp_data_list = []
while i < len(buf) and buf[i] != 0:
lppdata = LppData.from_bytes(buf[i:])
data.append(lppdata)
lpp_data_list.append(lppdata)
i = i + len(lppdata)
lpp = json.loads(json.dumps(LppFrame(data), default=lpp_json_encoder))
lpp = json.loads(json.dumps(LppFrame(lpp_data_list), default=lpp_json_encoder))
res["lpp"] = lpp