fix(python): Do math on monotonic clocks only (#19060)

## Overview

A lot of Python code (in various projects) was doing stuff like:

```python
start = time.now()
do_stuff()
end = time.now()
duration = end - start
```

That's not quite right because `time.now()` can be affected by things
like daylight savings time, leap seconds, and NTP adjustments. If we're
doing math on timestamps, it needs to come from a monotonic clock. So
this switches all (or at least most) of those places to use
`time.monotonic()` instead.

## Changelog

Basically a global Ctrl+F for `time.time()`. I replaced it with
`time.monotonic()` when it was obvious to me that it was only used in
the local scope and that math was being done on it. So this excludes
things like `time()` calls that were getting saved in CSV reports.

Author: SyntaxColoring

abr-testing/abr_testing/tools/test_modules.py

@@ -16,8 +16,8 @@
 async def tc_test_1(module: str, path_to_file: str) -> None:
     """Thermocycler Test 1 Open and Close Lid."""
     duration = int(input("How long to run this test for? (in seconds): "))
-    start = time.time()
-    while time.time() - start < duration:
+    start = time.monotonic()
+    while time.monotonic() - start < duration:
         try:
             await (tc_open_lid(module, path_to_file))
         except asyncio.TimeoutError:
@@ -34,8 +34,8 @@ async def hs_test_1(module: str, path_to_file: str) -> None:
     """Heater Shaker Test 1. (Home and Shake)."""
     duration = int(input("How long to run this test for? (in seconds): "))
     rpm = input("Target RPM (200-3000): ")
-    start = time.time()
-    while time.time() - start < duration:
+    start = time.monotonic()
+    while time.monotonic() - start < duration:
         try:
             await (hs_test_home(module, path_to_file))
         except asyncio.TimeoutError:

analyses-snapshot-testing/citools/generate_analyses.py

@@ -125,8 +125,8 @@ def start_containers(image_name: str, num_containers: int, timeout: int = 60) ->
         containers.append(container)
 
     # Wait for containers to be ready
-    start_time = time.time()
-    while time.time() - start_time < timeout:
+    start_time = time.monotonic()
+    while time.monotonic() - start_time < timeout:
         all_ready = True
         for container in containers:
             exit_code, _ = container.exec_run(f"ls -al {CONTAINER_LABWARE}")
@@ -172,7 +172,7 @@ def analyze(protocol: TargetProtocol, container: docker.models.containers.Contai
     # Build the command with all relevant file paths
     all_files = [str(protocol.container_protocol_file)] + [str(lw) for lw in labware_files]
     command = f"python -I -m opentrons.cli analyze --json-output {protocol.container_analysis_file} " + " ".join(all_files)
-    start_time = time.time()
+    start_time = time.monotonic()
     result = None
     exit_code = None
     console.print(f"Beginning analysis of {protocol.host_protocol_file.name}")
@@ -191,7 +191,7 @@ def analyze(protocol: TargetProtocol, container: docker.models.containers.Contai
         protocol.set_analysis()
         return False
     finally:
-        protocol.set_analysis_execution_time(time.time() - start_time)
+        protocol.set_analysis_execution_time(time.monotonic() - start_time)
         console.print(f"Analysis of {protocol.host_protocol_file.name} completed in {protocol.analysis_execution_time:.2f} seconds.")
 
 
@@ -238,7 +238,7 @@ def get_container_instances(protocol_len: int) -> int:
 
 def generate_analyses_from_test(tag: str, protocols: List[Protocol]) -> List[TargetProtocol]:
     """Generate analyses from the tests."""
-    start_time = time.time()
+    start_time = time.monotonic()
     protocols_to_process: List[TargetProtocol] = []
     for test_protocol in protocols:
         host_protocol_file = Path(test_protocol.file_path)
@@ -256,6 +256,6 @@ def generate_analyses_from_test(tag: str, protocols: List[Protocol]) -> List[Tar
         )
     instance_count = get_container_instances(len(protocols_to_process))
     analyze_against_image(tag, protocols_to_process, instance_count)
-    end_time = time.time()
+    end_time = time.monotonic()
     console.print(f"Clock time to generate analyses: {end_time - start_time:.2f} seconds.")
     return protocols_to_process

api/src/opentrons/drivers/smoothie_drivers/driver_3_0.py

@@ -12,7 +12,7 @@
 import contextlib
 import logging
 from os import environ
-from time import time
+from time import monotonic
 from typing import Any, Dict, Optional, Union, List, Tuple, cast, AsyncIterator
 
 from math import isclose
@@ -1901,12 +1901,12 @@ async def update_firmware(
         # if loop:
         #    kwargs["loop"] = loop
         log.info(update_cmd)
-        before = time()
+        before = monotonic()
         proc = await asyncio.create_subprocess_shell(update_cmd, **kwargs)
-        created = time()
+        created = monotonic()
         log.info(f"created lpc21isp subproc in {created-before}")
         out_b, err_b = await proc.communicate()
-        done = time()
+        done = monotonic()
         log.info(f"ran lpc21isp subproc in {done-created}")
         if proc.returncode != 0:
             log.error(

g-code-testing/g_code_parsing/g_code_engine.py

@@ -108,10 +108,10 @@ def _emulate(self) -> Iterator[ThreadManager]:
             feature_flags=HardwareFeatureFlags.build_from_ff(),
         )
         # Wait for modules to be present
-        wait_begins = time.time()
+        wait_begins = time.monotonic()
         while len(emulator.attached_modules) != len(modules):
             time.sleep(0.1)
-            if (time.time() - wait_begins) > 30:
+            if (time.monotonic() - wait_begins) > 30:
                 proc.kill()
                 proc.join()
                 raise RuntimeError(

hardware-testing/hardware_testing/drivers/mark10/mark10_fg.py

@@ -1,7 +1,7 @@
 """Mark10 Force Gauge Driver."""
 from serial import Serial  # type: ignore[import]
 from abc import ABC, abstractmethod
-from time import time
+from time import monotonic
 from typing import Tuple
 
 
@@ -96,8 +96,8 @@ def disconnect(self) -> None:
     def read_force(self, timeout: float = 1.0) -> float:
         """Get Force in Newtons."""
         self._force_guage.write("?\r\n".encode("utf-8"))
-        start_time = time()
-        while time() < start_time + timeout:
+        start_time = monotonic()
+        while monotonic() < start_time + timeout:
             # return "12.3 N"
             line = self._force_guage.readline().decode("utf-8").strip()
             try:

hardware-testing/hardware_testing/drivers/mitutoyo_digimatic_indicator.py

@@ -66,9 +66,9 @@ def read(self) -> float:
 
     def read_stable(self, timeout: float = 5) -> float:
         """Reads dial indicator with stable reading."""
-        then = time.time()
+        then = time.monotonic()
         values = [self.read(), self.read(), self.read(), self.read(), self.read()]
-        while (time.time() - then) < timeout:
+        while (time.monotonic() - then) < timeout:
             if numpy.allclose(values, list(reversed(values))):
                 return values[-1]
             values = values[1:] + [self.read()]
@@ -79,8 +79,8 @@ def read_stable(self, timeout: float = 5) -> float:
     print("Mitutoyo ABSOLUTE Digimatic Indicator")
     gauge = Mitutoyo_Digimatic_Indicator(port="/dev/ttyUSB0")
     gauge.connect()
-    start_time = time.time()
+    start_time = time.monotonic()
     while True:
-        elapsed_time = round(time.time() - start_time, 3)
+        elapsed_time = round(time.monotonic() - start_time, 3)
         distance = gauge.read()
         print("Time: {} Distance: {}".format(elapsed_time, distance))

hardware-testing/hardware_testing/production_qc/ninety_six_assembly_qc_ot3/test_droplets.py

@@ -1,6 +1,6 @@
 """Test Droplets."""
 from asyncio import sleep
-from time import time
+from time import monotonic
 from typing import List, Union, Tuple, Optional, Dict, Literal
 
 from opentrons.hardware_control.ot3api import OT3API
@@ -110,7 +110,7 @@ async def aspirate_and_wait(
     await api.aspirate(OT3Mount.LEFT, volume)
     await api.move_to(OT3Mount.LEFT, reservoir + Point(z=HOVER_HEIGHT_MM))
 
-    start_time = time()
+    start_time = monotonic()
     for i in range(seconds):
         print(f"waiting {i + 1}/{seconds}")
         if i == 0 or i == seconds - 1:
@@ -123,7 +123,7 @@ async def aspirate_and_wait(
         result = ui.get_user_answer("look good")
     else:
         result = True
-    duration_seconds = time() - start_time
+    duration_seconds = monotonic() - start_time
     print(f"waited for {duration_seconds} seconds")
 
     await api.move_to(

hardware-testing/hardware_testing/production_qc/robot_assembly_qc_ot3/test_peripherals.py

@@ -4,7 +4,7 @@
 from subprocess import run as run_subprocess, Popen, CalledProcessError
 from typing import List, Union, Optional, Dict
 from urllib.request import urlopen
-from time import time
+from time import monotonic
 from typing import Tuple
 
 from opentrons_hardware.hardware_control.rear_panel_settings import set_ui_color
@@ -223,19 +223,19 @@ def _get_user_confirmation(question: str) -> bool:
     ui.print_header("DOOR SWITCH")
     door_timeout_seconds = 10
     print("CLOSE the front door")
-    start_time_seconds = time()
+    start_time_seconds = monotonic()
     while not api.is_simulator and api.door_state != DoorState.CLOSED:
         await asyncio.sleep(0.1)
-        if time() - start_time_seconds > door_timeout_seconds:
+        if monotonic() - start_time_seconds > door_timeout_seconds:
             ui.print_error("timed out waiting for door to close")
             break
     print(api.door_state)
     is_closed = api.door_state == DoorState.CLOSED
     print("OPEN the front door")
-    start_time_seconds = time()
+    start_time_seconds = monotonic()
     while not api.is_simulator and api.door_state != DoorState.OPEN:
         await asyncio.sleep(0.1)
-        if time() - start_time_seconds > door_timeout_seconds:
+        if monotonic() - start_time_seconds > door_timeout_seconds:
             ui.print_error("timed out waiting for door to open")
             break
     print(api.door_state)

hardware-testing/hardware_testing/production_qc/stress_test_qc_ot3.py

@@ -131,7 +131,7 @@ def _create_csv_and_get_callbacks(sn: str) -> Tuple[CSVProperties, CSVCallbacks]
     file_name = data.create_file_name(test_name=test_name, run_id=run_id, tag=sn)
     csv_display_name = os.path.join(run_path, file_name)
     print(f"CSV: {csv_display_name}")
-    start_time = time.time()
+    start_time = time.monotonic()
 
     def _append_csv_data(
         data_list: List[Any],
@@ -142,7 +142,7 @@ def _append_csv_data(
         # every line in the CSV file begins with the elapsed seconds
         if not first_row_value_included:
             if first_row_value is None:
-                first_row_value = str(round(time.time() - start_time, 2))
+                first_row_value = str(round(time.monotonic() - start_time, 2))
             data_list = [first_row_value] + data_list
         data_str = ",".join([str(d) for d in data_list])
         if line_number is None:

hardware-testing/hardware_testing/scripts/faster_plunger_lifetime_test.py

@@ -101,7 +101,7 @@ async def position_check() -> bool:
         }
         writer = csv.DictWriter(csvfile, test_data)
         writer.writeheader()
-        start_time = time.time()
+        start_time = time.monotonic()
         try:
             currents = list(CURRENTS_SPEEDS.keys())
             for cycle in range(1, args.cycles + 1):
@@ -133,7 +133,7 @@ async def position_check() -> bool:
                             api, mount, bottom, speed=speed, motor_current=current
                         )
                         down_passed = await position_check()
-                        test_data["time_sec"] = time.time() - start_time
+                        test_data["time_sec"] = time.monotonic() - start_time
                         test_data["cycle"] = cycle
                         test_data["position"] = "bottom"
                         test_data["position_check"] = down_passed
@@ -191,7 +191,7 @@ async def position_check() -> bool:
                             api, mount, 0, speed=speed, motor_current=current
                         )
                         up_passed = await position_check()
-                        test_data["time_sec"] = time.time() - start_time
+                        test_data["time_sec"] = time.monotonic() - start_time
                         test_data["cycle"] = cycle
                         test_data["position"] = "top"
                         test_data["position_check"] = up_passed