transition through power stress states and capture data

meshtastic-py3.12kevinh@kdesktop:~/development/meshtastic/meshtastic-python$  cd /home/kevinh/development/meshtastic/meshtastic-python ; /usr/bin/env /home/kevinh/.cache/pypoetry/virtualenvs/meshtastic-l6tP90xw-py3.12/bin/python /home/kevinh/.vscode/extensions/ms-python.debugpy-2024.6.0-linux-x64/bundled/libs/debugpy/adapter/../../debugpy/launcher 52521 -- -m meshtastic --slog --power-ppk2-meter --power-stress --power-voltage 3.3
INFO file:ppk2.py __init__ line:52 Connected to Power Profiler Kit II (PPK2)
INFO file:__main__.py create_power_meter line:1022 Setting power supply to 3.3 volts
Connected to radio
INFO file:slog.py __init__ line:183 Writing slogs to /home/kevinh/.local/share/meshtastic/slogs/20240706-123803
INFO file:stress.py syncPowerStress line:68 Sending power stress command PRINT_INFO
INFO file:stress.py run line:105 Running power stress test 48 for 5.0 seconds
INFO file:stress.py syncPowerStress line:68 Sending power stress command LED_ON
INFO file:stress.py run line:105 Running power stress test 80 for 5.0 seconds
INFO file:stress.py syncPowerStress line:68 Sending power stress command BT_OFF
INFO file:stress.py run line:105 Running power stress test 81 for 5.0 seconds
INFO file:stress.py syncPowerStress line:68 Sending power stress command BT_ON
INFO file:stress.py run line:105 Running power stress test 34 for 5.0 seconds
INFO file:stress.py syncPowerStress line:68 Sending power stress command CPU_FULLON
INFO file:stress.py run line:105 Running power stress test 32 for 5.0 seconds
INFO file:stress.py syncPowerStress line:68 Sending power stress command CPU_IDLE
INFO file:stress.py run line:105 Running power stress test 33 for 5.0 seconds
INFO file:stress.py syncPowerStress line:68 Sending power stress command CPU_DEEPSLEEP
INFO file:stress.py run line:108 Power stress test complete.
INFO file:slog.py close line:201 Closing slogs in /home/kevinh/.local/share/meshtastic/slogs/20240706-123803
WARNING file:arrow.py close line:67 Discarding empty file: /home/kevinh/.local/share/meshtastic/slogs/20240706-123803/slog.arrow
INFO file:arrow.py close line:70 Compressing log data into /home/kevinh/.local/share/meshtastic/slogs/20240706-123803/power.feather
meshtastic-py3.12kevinh@kdesktop:~/development/meshtastic/meshtastic-python$

.vscode/launch.json

@@ -204,7 +204,7 @@
       "request": "launch",
       "module": "meshtastic",
       "justMyCode": false,
-      "args": ["--slog-out", "default", "--power-ppk2-meter", "--power-stress", "--power-voltage", "3.3", "--seriallog"]
+      "args": ["--slog", "--power-ppk2-meter", "--power-stress", "--power-voltage", "3.3"]
     },
     {
       "name": "meshtastic test",

.vscode/settings.json

@@ -2,6 +2,7 @@
     "cSpell.words": [
         "bitmask",
         "boardid",
+        "DEEPSLEEP",
         "Meshtastic",
         "milliwatt",
         "portnums",

meshtastic/__main__.py

@@ -852,14 +852,16 @@ def onConnected(interface):
             qr = pyqrcode.create(url)
             print(qr.terminal())
 
+        log_set: Optional[LogSet] = None  # we need to keep a reference to the logset so it doesn't get GCed early
         if args.slog or args.power_stress:
             # Setup loggers
             global meter  # pylint: disable=global-variable-not-assigned
-            LogSet(interface, args.slog if args.slog != 'default' else None, meter)
+            log_set = LogSet(interface, args.slog if args.slog != 'default' else None, meter)
 
             if args.power_stress:
                 stress = PowerStress(interface)
                 stress.run()
+                closeNow = True # exit immediately after stress test
 
         if args.listen:
             closeNow = False
@@ -895,6 +897,8 @@ def onConnected(interface):
         # if the user didn't ask for serial debugging output, we might want to exit after we've done our operation
         if (not args.seriallog) and closeNow:
             interface.close()  # after running command then exit
+            if log_set:
+                log_set.close()
 
     except Exception as ex:
         print(f"Aborting due to: {ex}")

meshtastic/powermon/stress.py

@@ -3,7 +3,7 @@
 import logging
 import time
 
-from ..protobuf import ( portnums_pb2, powermon_pb2 )
+from ..protobuf import portnums_pb2, powermon_pb2
 
 
 def onPowerStressResponse(packet, interface):
@@ -53,6 +53,32 @@ class PowerStressClient:
             onResponseAckPermitted=True,
         )
 
+    def syncPowerStress(
+        self,
+        cmd: powermon_pb2.PowerStressMessage.Opcode.ValueType,
+        num_seconds: float = 0.0,
+    ):
+        """Send a power stress command and wait for the ack."""
+        gotAck = False
+
+        def onResponse(packet: dict):  # pylint: disable=unused-argument
+            nonlocal gotAck
+            gotAck = True
+
+        logging.info(f"Sending power stress command {powermon_pb2.PowerStressMessage.Opcode.Name(cmd)}")
+        self.sendPowerStress(cmd, onResponse=onResponse, num_seconds=num_seconds)
+
+        if num_seconds == 0.0:
+            # Wait for the response and then continue
+            while not gotAck:
+                time.sleep(0.1)
+        else:
+            # we wait a little bit longer than the time the UUT would be waiting (to make sure all of its messages are handled first)
+            time.sleep(num_seconds + 0.2)  # completely block our thread for the duration of the test
+            if not gotAck:
+                logging.error("Did not receive ack for power stress command!")
+
+
 
 class PowerStress:
     """Walk the UUT through a set of power states so we can capture repeatable power consumption measurements."""
@@ -63,19 +89,21 @@ class PowerStress:
     def run(self):
         """Run the power stress test."""
         # Send the power stress command
-        gotAck = False
 
-        def onResponse(packet: dict):  # pylint: disable=unused-argument
-            nonlocal gotAck
-            gotAck = True
+        self.client.syncPowerStress(powermon_pb2.PowerStressMessage.PRINT_INFO)
 
-        logging.info("Starting power stress test, attempting to contact UUT...")
-        self.client.sendPowerStress(
-            powermon_pb2.PowerStressMessage.PRINT_INFO, onResponse=onResponse
-        )
+        num_seconds = 5.0
+        states = [
+            powermon_pb2.PowerStressMessage.LED_ON,
+            powermon_pb2.PowerStressMessage.BT_OFF,
+            powermon_pb2.PowerStressMessage.BT_ON,
+            powermon_pb2.PowerStressMessage.CPU_FULLON,
+            powermon_pb2.PowerStressMessage.CPU_IDLE,
+            powermon_pb2.PowerStressMessage.CPU_DEEPSLEEP,
+        ]
+        for s in states:
+            s_name = powermon_pb2.PowerStressMessage.Opcode.Name(s)
+            logging.info(f"Running power stress test {s_name} for {num_seconds} seconds")
+            self.client.syncPowerStress(s, num_seconds)
 
-        # Wait for the response
-        while not gotAck:
-            time.sleep(0.1)
-
-        logging.info("Power stress test complete.")
+        logging.info("Power stress test complete.")