PowerStress WIP

meshtastic/__init__.py

@@ -97,6 +97,7 @@ def onConnection(interface, topic=pub.AUTO_TOPIC): # called when we (re)connect
     remote_hardware_pb2,
     storeforward_pb2,
     telemetry_pb2,
+    powermon_pb2
 )
 from . import (
     util,
@@ -229,6 +230,9 @@ def _receiveInfoUpdate(iface, asDict):
     portnums_pb2.PortNum.TRACEROUTE_APP: KnownProtocol(
         "traceroute", mesh_pb2.RouteDiscovery
     ),
+    portnums_pb2.PortNum.POWERSTRESS_APP: KnownProtocol(
+        "powerstress", powermon_pb2.PowerStressMessage
+    ),
     portnums_pb2.PortNum.WAYPOINT_APP: KnownProtocol("waypoint", mesh_pb2.Waypoint),
     portnums_pb2.PortNum.PAXCOUNTER_APP: KnownProtocol("paxcounter", paxcount_pb2.Paxcount),
     portnums_pb2.PortNum.STORE_FORWARD_APP: KnownProtocol("storeforward", storeforward_pb2.StoreAndForward),

meshtastic/__main__.py

@@ -11,6 +11,7 @@
 import platform
 import sys
 import time
+from typing import Optional
 
 import pyqrcode # type: ignore[import-untyped]
 import yaml
@@ -25,9 +26,11 @@
 from meshtastic.version import get_active_version
 from meshtastic.ble_interface import BLEInterface
 from meshtastic.mesh_interface import MeshInterface
-from meshtastic.powermon import RidenPowerSupply, PPK2PowerSupply, SimPowerSupply
+from meshtastic.powermon import RidenPowerSupply, PPK2PowerSupply, SimPowerSupply, PowerStress, PowerMeter
 from meshtastic.slog import LogSet
 
+meter: Optional[PowerMeter] = None
+
 def onReceive(packet, interface):
     """Callback invoked when a packet arrives"""
     args = mt_config.args
@@ -849,6 +852,15 @@ def setSimpleConfig(modem_preset):
             qr = pyqrcode.create(url)
             print(qr.terminal())
 
+        if args.slog_out:
+            # Setup loggers
+            global meter
+            LogSet(interface, args.slog_out if args.slog_out != 'default' else None, meter)
+
+        if args.power_stress:
+            stress = PowerStress(interface)
+            stress.run()
+
         if args.listen:
             closeNow = False
 
@@ -989,6 +1001,7 @@ def export_config(interface):
 def create_power_meter():
     """Setup the power meter."""
 
+    global meter
     args = mt_config.args
     meter = None # assume no power meter
     if args.power_riden:
@@ -1009,7 +1022,6 @@ def create_power_meter():
 
         if args.power_wait:
             input("Powered on, press enter to continue...")
-    return meter
 
 def common():
     """Shared code for all of our command line wrappers."""
@@ -1029,7 +1041,7 @@ def common():
             meshtastic.util.support_info()
             meshtastic.util.our_exit("", 0)
 
-        meter = create_power_meter()
+        create_power_meter()
 
         if args.ch_index is not None:
             channelIndex = int(args.ch_index)
@@ -1120,11 +1132,6 @@ def common():
             # We assume client is fully connected now
             onConnected(client)
 
-            log_set = None
-            if args.slog_out:
-                # Setup loggers
-                log_set = LogSet(client, args.slog_out if args.slog_out != 'default' else None, meter)
-
             have_tunnel = platform.system() == "Linux"
             if (
                 args.noproto or args.reply or (have_tunnel and args.tunnel) or args.listen
@@ -1135,9 +1142,6 @@ def common():
                 except KeyboardInterrupt:
                     logging.info("Exiting due to keyboard interrupt")
 
-            if log_set:
-                log_set.close()
-
         # don't call exit, background threads might be running still
         # sys.exit(0)
 

meshtastic/powermon/__init__.py

@@ -4,3 +4,4 @@
 from .ppk2 import PPK2PowerSupply
 from .riden import RidenPowerSupply
 from .sim import SimPowerSupply
+from .stress import PowerStress

meshtastic/powermon/stress.py

@@ -0,0 +1,78 @@
+"""Power stress testing support.
+"""
+import logging
+import time
+
+from pubsub import pub  # type: ignore[import-untyped]
+
+from meshtastic.protobuf import portnums_pb2
+from meshtastic.protobuf.powermon_pb2 import PowerStressMessage
+
+
+def onPowerStressResponse(packet, interface):
+    """Delete me? FIXME"""
+    logging.debug(f"packet:{packet} interface:{interface}")
+    # interface.gotResponse = True
+
+
+class PowerStressClient:
+    """
+    The client stub for talking to the firmware PowerStress module.
+    """
+
+    def __init__(self, iface, node_id = None):
+        """
+        Create a new PowerStressClient instance.
+
+        iface is the already open MeshInterface instance
+        """
+        self.iface = iface
+
+        if not node_id:
+            node_id = iface.myInfo.my_node_num
+
+        self.node_id = node_id            
+        # No need to subscribe - because we
+        # pub.subscribe(onGPIOreceive, "meshtastic.receive.powerstress")
+
+    def sendPowerStress(
+        self, cmd: PowerStressMessage.Opcode.ValueType, num_seconds: float = 0.0, onResponse=None
+    ):
+        r = PowerStressMessage()
+        r.cmd = cmd
+        r.num_seconds = num_seconds
+
+        return self.iface.sendData(
+            r,
+            self.node_id,
+            portnums_pb2.POWERSTRESS_APP,
+            wantAck=True,
+            wantResponse=False,
+            onResponse=onResponse,
+            onResponseAckPermitted=True
+        )
+
+class PowerStress:
+    """Walk the UUT through a set of power states so we can capture repeatable power consumption measurements."""
+
+    def __init__(self, iface):
+        self.client = PowerStressClient(iface)
+
+
+    def run(self):
+        """Run the power stress test."""
+        # Send the power stress command
+        gotAck = False
+
+        def onResponse(packet, interface):
+            nonlocal gotAck
+            gotAck = True
+
+        logging.info("Starting power stress test, attempting to contact UUT...")   
+        self.client.sendPowerStress(PowerStressMessage.PRINT_INFO, onResponse=onResponse)
+
+        # Wait for the response
+        while not gotAck:
+            time.sleep(0.1)
+
+        logging.info("Power stress test complete.")

meshtastic/protobuf/portnums_pb2.pyi

@@ -171,6 +171,10 @@ class _PortNumEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._EnumTy
     """
     Provides unencrypted information about a node for consumption by a map via MQTT
     """
+    POWERSTRESS_APP: _PortNum.ValueType  # 74
+    """
+    PowerStress based monitoring support (for automated power consumption testing)
+    """
     PRIVATE_APP: _PortNum.ValueType  # 256
     """
     Private applications should use portnums >= 256.
@@ -352,6 +356,10 @@ MAP_REPORT_APP: PortNum.ValueType  # 73
 """
 Provides unencrypted information about a node for consumption by a map via MQTT
 """
+POWERSTRESS_APP: PortNum.ValueType  # 74
+"""
+PowerStress based monitoring support (for automated power consumption testing)
+"""
 PRIVATE_APP: PortNum.ValueType  # 256
 """
 Private applications should use portnums >= 256.

meshtastic/protobuf/powermon_pb2.pyi

@@ -95,3 +95,127 @@ class PowerMon(google.protobuf.message.Message):
     ) -> None: ...
 
 global___PowerMon = PowerMon
+
+@typing.final
+class PowerStressMessage(google.protobuf.message.Message):
+    """
+    PowerStress testing support via the C++ PowerStress module
+    """
+
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    class _Opcode:
+        ValueType = typing.NewType("ValueType", builtins.int)
+        V: typing_extensions.TypeAlias = ValueType
+
+    class _OpcodeEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[PowerStressMessage._Opcode.ValueType], builtins.type):
+        DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor
+        UNSET: PowerStressMessage._Opcode.ValueType  # 0
+        """
+        Unset/unused
+        """
+        PRINT_INFO: PowerStressMessage._Opcode.ValueType  # 1
+        """Print board version slog and send an ack that we are alive and ready to process commands"""
+        FORCE_QUIET: PowerStressMessage._Opcode.ValueType  # 2
+        """Try to turn off all automatic processing of packets, screen, sleeping, etc (to make it easier to measure in isolation)"""
+        END_QUIET: PowerStressMessage._Opcode.ValueType  # 3
+        """Stop powerstress processing - probably by just rebooting the board"""
+        SCREEN_ON: PowerStressMessage._Opcode.ValueType  # 16
+        """Turn the screen on"""
+        SCREEN_OFF: PowerStressMessage._Opcode.ValueType  # 17
+        """Turn the screen off"""
+        CPU_IDLE: PowerStressMessage._Opcode.ValueType  # 32
+        """Let the CPU run but we assume mostly idling for num_seconds"""
+        CPU_DEEPSLEEP: PowerStressMessage._Opcode.ValueType  # 33
+        """Force deep sleep for FIXME seconds"""
+        CPU_FULLON: PowerStressMessage._Opcode.ValueType  # 34
+        """Spin the CPU as fast as possible for num_seconds"""
+        LED_ON: PowerStressMessage._Opcode.ValueType  # 48
+        """Turn the LED on for num_seconds (and leave it on - for baseline power measurement purposes)"""
+        LED_OFF: PowerStressMessage._Opcode.ValueType  # 49
+        """Force the LED off for num_seconds"""
+        LORA_OFF: PowerStressMessage._Opcode.ValueType  # 64
+        """Completely turn off the LORA radio for num_seconds"""
+        LORA_TX: PowerStressMessage._Opcode.ValueType  # 65
+        """Send Lora packets for num_seconds"""
+        LORA_RX: PowerStressMessage._Opcode.ValueType  # 66
+        """Receive Lora packets for num_seconds (node will be mostly just listening, unless an external agent is helping stress this by sending packets on the current channel)"""
+        BT_OFF: PowerStressMessage._Opcode.ValueType  # 80
+        """Turn off the BT radio for num_seconds"""
+        BT_ON: PowerStressMessage._Opcode.ValueType  # 81
+        """Turn on the BT radio for num_seconds"""
+        WIFI_OFF: PowerStressMessage._Opcode.ValueType  # 96
+        """Turn off the WIFI radio for num_seconds"""
+        WIFI_ON: PowerStressMessage._Opcode.ValueType  # 97
+        """Turn on the WIFI radio for num_seconds"""
+        GPS_OFF: PowerStressMessage._Opcode.ValueType  # 112
+        """Turn off the GPS radio for num_seconds"""
+        GPS_ON: PowerStressMessage._Opcode.ValueType  # 113
+        """Turn on the GPS radio for num_seconds"""
+
+    class Opcode(_Opcode, metaclass=_OpcodeEnumTypeWrapper):
+        """
+        What operation would we like the UUT to perform.
+        note: senders should probably set want_response in their request packets, so that they can know when the state
+        machine has started processing their request
+        """
+
+    UNSET: PowerStressMessage.Opcode.ValueType  # 0
+    """
+    Unset/unused
+    """
+    PRINT_INFO: PowerStressMessage.Opcode.ValueType  # 1
+    """Print board version slog and send an ack that we are alive and ready to process commands"""
+    FORCE_QUIET: PowerStressMessage.Opcode.ValueType  # 2
+    """Try to turn off all automatic processing of packets, screen, sleeping, etc (to make it easier to measure in isolation)"""
+    END_QUIET: PowerStressMessage.Opcode.ValueType  # 3
+    """Stop powerstress processing - probably by just rebooting the board"""
+    SCREEN_ON: PowerStressMessage.Opcode.ValueType  # 16
+    """Turn the screen on"""
+    SCREEN_OFF: PowerStressMessage.Opcode.ValueType  # 17
+    """Turn the screen off"""
+    CPU_IDLE: PowerStressMessage.Opcode.ValueType  # 32
+    """Let the CPU run but we assume mostly idling for num_seconds"""
+    CPU_DEEPSLEEP: PowerStressMessage.Opcode.ValueType  # 33
+    """Force deep sleep for FIXME seconds"""
+    CPU_FULLON: PowerStressMessage.Opcode.ValueType  # 34
+    """Spin the CPU as fast as possible for num_seconds"""
+    LED_ON: PowerStressMessage.Opcode.ValueType  # 48
+    """Turn the LED on for num_seconds (and leave it on - for baseline power measurement purposes)"""
+    LED_OFF: PowerStressMessage.Opcode.ValueType  # 49
+    """Force the LED off for num_seconds"""
+    LORA_OFF: PowerStressMessage.Opcode.ValueType  # 64
+    """Completely turn off the LORA radio for num_seconds"""
+    LORA_TX: PowerStressMessage.Opcode.ValueType  # 65
+    """Send Lora packets for num_seconds"""
+    LORA_RX: PowerStressMessage.Opcode.ValueType  # 66
+    """Receive Lora packets for num_seconds (node will be mostly just listening, unless an external agent is helping stress this by sending packets on the current channel)"""
+    BT_OFF: PowerStressMessage.Opcode.ValueType  # 80
+    """Turn off the BT radio for num_seconds"""
+    BT_ON: PowerStressMessage.Opcode.ValueType  # 81
+    """Turn on the BT radio for num_seconds"""
+    WIFI_OFF: PowerStressMessage.Opcode.ValueType  # 96
+    """Turn off the WIFI radio for num_seconds"""
+    WIFI_ON: PowerStressMessage.Opcode.ValueType  # 97
+    """Turn on the WIFI radio for num_seconds"""
+    GPS_OFF: PowerStressMessage.Opcode.ValueType  # 112
+    """Turn off the GPS radio for num_seconds"""
+    GPS_ON: PowerStressMessage.Opcode.ValueType  # 113
+    """Turn on the GPS radio for num_seconds"""
+
+    CMD_FIELD_NUMBER: builtins.int
+    NUM_SECONDS_FIELD_NUMBER: builtins.int
+    cmd: global___PowerStressMessage.Opcode.ValueType
+    """
+    What type of HardwareMessage is this?
+    """
+    num_seconds: builtins.float
+    def __init__(
+        self,
+        *,
+        cmd: global___PowerStressMessage.Opcode.ValueType = ...,
+        num_seconds: builtins.float = ...,
+    ) -> None: ...
+    def ClearField(self, field_name: typing.Literal["cmd", b"cmd", "num_seconds", b"num_seconds"]) -> None: ...
+
+global___PowerStressMessage = PowerStressMessage