mirror/python
1
0
Fork 0
mirror of https://github.com/meshtastic/python.git synced 2026-01-09 16:27:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

Support requesting different telemetry types

Browse Source
This commit is contained in:
Ian McEwen
2024-10-13 20:35:11 -07:00
parent 05731128fa
commit 7088b90514
2 changed files with 44 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
meshtastic/__main__.py
View File

@@ -475,13 +475,22 @@ def onConnected(interface):
else:
channelIndex = mt_config.channel_index or 0
if checkChannel(interface, channelIndex):
telemMap = {
"device": "device_metrics",
"environment": "environment_metrics",
"air_quality": "air_quality_metrics",
"airquality": "air_quality_metrics",
"power": "power_metrics",
}
telemType = telemMap.get(args.request_telemetry, "device_metrics")
print(
f"Sending telemetry request to {args.dest} on channelIndex:{channelIndex} (this could take a while)"
f"Sending {telemType} telemetry request to {args.dest} on channelIndex:{channelIndex} (this could take a while)"
)
interface.sendTelemetry(
destinationId=args.dest,
wantResponse=True,
channelIndex=channelIndex,
telemetryType=telemType,
)
if args.request_position:
@@ -1592,10 +1601,14 @@ def addRemoteActionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPar
group.add_argument(
"--request-telemetry",
help="Request telemetry from a node. "
help="Request telemetry from a node. With an argument, requests that specific type of telemetry. "
"You need to pass the destination ID as argument with '--dest'. "
"For repeaters, the nodeNum is required.",
action="store_true",
action="store",
nargs="?",
default=None,
const="device",
metavar="TYPE",
)
group.add_argument(

50
meshtastic/mesh_interface.py
View File

@@ -606,32 +606,38 @@ class MeshInterface: # pylint: disable=R0902
destinationId: Union[int, str] = BROADCAST_ADDR,
wantResponse: bool = False,
channelIndex: int = 0,
telemetryType: str = "device_metrics"
):
"""Send telemetry and optionally ask for a response"""
r = telemetry_pb2.Telemetry()
if self.nodes is not None:
node = next(
n for n in self.nodes.values() if n["num"] == self.localNode.nodeNum
)
if node is not None:
metrics = node.get("deviceMetrics")
if metrics:
batteryLevel = metrics.get("batteryLevel")
if batteryLevel is not None:
r.device_metrics.battery_level = batteryLevel
voltage = metrics.get("voltage")
if voltage is not None:
r.device_metrics.voltage = voltage
channel_utilization = metrics.get("channelUtilization")
if channel_utilization is not None:
r.device_metrics.channel_utilization = channel_utilization
air_util_tx = metrics.get("airUtilTx")
if air_util_tx is not None:
r.device_metrics.air_util_tx = air_util_tx
uptime_seconds = metrics.get("uptimeSeconds")
if uptime_seconds is not None:
r.device_metrics.uptime_seconds = uptime_seconds
if telemetryType == "environment_metrics":
r.environment_metrics.CopyFrom(telemetry_pb2.EnvironmentMetrics())
elif telemetryType == "air_quality_metrics":
r.air_quality_metrics.CopyFrom(telemetry_pb2.AirQualityMetrics())
elif telemetryType == "power_metrics":
r.power_metrics.CopyFrom(telemetry_pb2.PowerMetrics())
else: # fall through to device metrics
if self.nodes is not None:
node = self.nodesByNum.get(self.localNode.nodeNum)
if node is not None:
metrics = node.get("deviceMetrics")
if metrics:
batteryLevel = metrics.get("batteryLevel")
if batteryLevel is not None:
r.device_metrics.battery_level = batteryLevel
voltage = metrics.get("voltage")
if voltage is not None:
r.device_metrics.voltage = voltage
channel_utilization = metrics.get("channelUtilization")
if channel_utilization is not None:
r.device_metrics.channel_utilization = channel_utilization
air_util_tx = metrics.get("airUtilTx")
if air_util_tx is not None:
r.device_metrics.air_util_tx = air_util_tx
uptime_seconds = metrics.get("uptimeSeconds")
if uptime_seconds is not None:
r.device_metrics.uptime_seconds = uptime_seconds
if wantResponse:
onResponse = self.onResponseTelemetry