meshcore-dev/meshcore_py
1
0
Fork 0
mirror of https://github.com/meshcore-dev/meshcore_py.git synced 2026-04-20 22:13:49 +00:00
Code Issues Projects Releases Packages Wiki Activity

custom vars

Browse source
This commit is contained in:
Florent de Lamotte 2025-05-05 15:03:45 +02:00
parent a39af17566
commit 27f87075d2
3 changed files with 25 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/meshcore/commands.py
View file

@ -313,6 +313,16 @@ class CommandHandler:
data = b"\x27\x00\x00\x00" + dst_bytes
return await self.send(data, [EventType.MSG_SENT, EventType.ERROR])
async def get_custom_vars(self) -> Event:
logger.debug(f"Asking for custom vars")
data = b"\x28"
return await self.send(data, [EventType.CUSTOM_VARS, EventType.ERROR])
async def set_custom_var(self, key, value) -> Event:
logger.debug(f"Setting custom var {key} to {value}")
data = b"\x29" + key.encode("utf-8") + ":" + value.encode("utf-8")
return await self.send(data, [EventType.OK, EventType.ERROR])
async def send_cli(self, cmd) -> Event:
logger.debug(f"Sending CLI command: {cmd}")
data = b"\x32" + cmd.encode('utf-8')

3
src/meshcore/events.py
View file

@ -34,6 +34,7 @@ class EventType(Enum):
TRACE_DATA = "trace_data"
RX_LOG_DATA = "rx_log_data"
TELEMETRY_RESPONSE = "telemetry_response"
CUSTOM_VARS = "custom_vars"
# Command response types
OK = "command_ok"
@ -183,4 +184,4 @@ class EventDispatcher:
except asyncio.TimeoutError:
return None
finally:
subscription.unsubscribe()
subscription.unsubscribe()

14
src/meshcore/reader.py
View file

@ -201,6 +201,17 @@ class MessageReader:
res["response"] = data[1:].decode()
await self.dispatcher.dispatch(Event(EventType.CLI_RESPONSE, res))
elif packet_type_value == PacketType.CUSTOM_VARS.value:
logger.debug(f"received custom vars response: {data.hex()}")
res = {}
rawdata = data[1:].decode()
pairs = rawdata.split(",")
for p in pairs :
psplit = p.split(":")
res[psplit[0]] = psplit[1]
logger.debug(f"got custom vars : {res}")
await self.dispatcher.dispatch(Event(EventType.CUSTOM_VARS, res))
# Push notifications
elif packet_type_value == PacketType.ADVERTISEMENT.value:
logger.debug("Advertisement received")
@ -374,4 +385,5 @@ class MessageReader:
else:
logger.debug(f"Unhandled data received {data}")
logger.debug(f"Unhandled packet type: {packet_type_value}")
logger.debug(f"Unhandled packet type: {packet_type_value}")