refactor(python): use 3.11 builtins for UTC and TimeoutError

examples/throughput_server/throughput_server.py

@@ -104,7 +104,7 @@ async def handle(self) -> None:
                     self._buffer += data
                     await self._process_buffer()
 
-                except asyncio.TimeoutError:
+                except TimeoutError:
                     # No data, continue loop (allows checking self.running)
                     continue
                 except ConnectionResetError:

openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/calendar.py

@@ -11,7 +11,7 @@
 set_timeline_color("PythonTL", "#4287f5")
 print(get_timeline("PythonTL")) #=> {'name': 'PythonTL', 'color': '#4287f5', 'execute': True, 'shard': 0, 'scope': 'DEFAULT', 'updated_at': 1737129508391137136}
 
-now = datetime.datetime.now(datetime.timezone.utc)
+now = datetime.datetime.now(datetime.UTC)
 start = now + datetime.timedelta(hours=1)
 stop = start + datetime.timedelta(hours=1)
 act1 = create_timeline_activity("PythonTL", kind="reserve", start=start, stop=stop)
@@ -45,8 +45,8 @@
 #   "events"=>[{"time"=>1737128761, "event"=>"created"}], "recurring"=>{}}
 
 # Update act2 with a custom title and notes
-new_start = datetime.datetime.fromtimestamp(act2['start'], tz=datetime.timezone.utc)
-new_stop = datetime.datetime.fromtimestamp(act2['stop'], tz=datetime.timezone.utc) + datetime.timedelta(minutes=30)
+new_start = datetime.datetime.fromtimestamp(act2['start'], tz=datetime.UTC)
+new_stop = datetime.datetime.fromtimestamp(act2['stop'], tz=datetime.UTC) + datetime.timedelta(minutes=30)
 updated = update_timeline_activity("PythonTL",
     id=act2['start'],
     kind="COMMAND", start=new_start, stop=new_stop,

openc3-cosmos-init/plugins/packages/openc3-cosmos-demo/targets/INST2/procedures/streaming.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timezone, timedelta
+from datetime import UTC, datetime, timedelta
 from openc3.script.web_socket_api import StreamingWebSocketApi
 
 
@@ -17,7 +17,7 @@
         print(api.read())
 
 # Warning this saves all data to RAM. Do not use for large queries
-now = datetime.now(timezone.utc)
+now = datetime.now(UTC)
 data = StreamingWebSocketApi.read_all(
     items=[
         "DECOM__TLM__INST__HEALTH_STATUS__TEMP1__CONVERTED",

openc3/python/examples/cosmos_web_socket_example.py

@@ -1,5 +1,5 @@
 import os
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 
 
 # Example Use
@@ -43,7 +43,7 @@
 
 
 # Warning this saves all data to RAM. Do not use for large queries
-now = datetime.now(timezone.utc)
+now = datetime.now(UTC)
 data = StreamingWebSocketApi.read_all(
     items=[
         "DECOM__TLM__INST__HEALTH_STATUS__TEMP1__CONVERTED",
@@ -55,7 +55,7 @@
 print(data)
 
 
-now = datetime.now(timezone.utc)
+now = datetime.now(UTC)
 with MessagesWebSocketApi(
     history_count=0,
     start_time=to_nsec_from_epoch(now - timedelta(seconds=3600)),

openc3/python/openc3/microservices/host_interface_microservice.py

@@ -425,7 +425,7 @@ async def _send_status(self, send):
             self._status_ping.clear()
             line = (json.dumps({"type": "status", "status": self._status_snapshot()}) + "\n").encode()
             await send.write_all(line)
-            with contextlib.suppress(asyncio.TimeoutError):
+            with contextlib.suppress(TimeoutError):
                 await asyncio.wait_for(self._status_ping.wait(), timeout=STATUS_INTERVAL)
 
     def _status_snapshot(self):

openc3/python/test/microservices/test_bridge_microservice.py

@@ -134,7 +134,7 @@ async def test_read_request_has_a_timeout(self):
         recv = FakeRecv()
         with (
             patch("openc3.microservices.bridge_microservice.REQUEST_TIMEOUT", 0.01),
-            self.assertRaises(asyncio.TimeoutError),
+            self.assertRaises(TimeoutError),
         ):
             await self.service._read_request(recv)
 

openc3/python/test/microservices/test_interface_microservice.py

@@ -216,7 +216,7 @@ def test_creates_an_interface_updates_status_and_starts_cmd_thread(self):
     # def test_preserves_existing_packet_counts(self):
     #     # Initialize the telemetry topic with a non-zero RECEIVED_COUNT
     #     for _, packet in System.telemetry.packets("INST").items():
-    #         packet.received_time = datetime.now(timezone.utc)
+    #         packet.received_time = datetime.now(UTC)
     #         packet.received_count = 10
     #         TelemetryTopic.write_packet(packet, scope="DEFAULT")
     #     im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT")

openc3/python/test/test_helper.py

@@ -90,7 +90,7 @@ def setup_system(targets=None):
                     json.dumps(packet.as_json()),
                 )
                 packet = System.telemetry.packet(target_name, packet_name)
-                # packet.received_time = datetime.now(timezone.utc)
+                # packet.received_time = datetime.now(UTC)
                 json_hash = {}
                 for item in packet.sorted_items:
                     # Initialize all items to None like TargetModel::update_store does in Ruby

openc3/python/tools/ty_report.py.lock

@@ -1,3 +1,3 @@
 version = 1
-revision = 3
-requires-python = ">=3.10"
+revision = 5
+requires-python = ">=3.11"

openc3/test/integration/tsdb/helpers/questdb_writer.py

@@ -43,7 +43,7 @@
 import os
 import sys
 import time
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 # Add openc3 python path for imports
 # Path: openc3/test/integration/tsdb/helpers -> openc3/python
@@ -78,7 +78,7 @@ def ts_to_iso(ts_ns: int) -> str:
     """Convert nanosecond timestamp to ISO string for QuestDB queries."""
     ts_us = ts_ns // 1000
     ts_s = ts_us / 1_000_000
-    return datetime.fromtimestamp(ts_s, tz=timezone.utc).strftime(
+    return datetime.fromtimestamp(ts_s, tz=UTC).strftime(
         "%Y-%m-%dT%H:%M:%S.%fZ"
     )
 

openc3/test/integration/tsdb/insert_test_data.py

@@ -18,7 +18,7 @@
 import math
 import random
 import sys
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 
 import psycopg
 
@@ -134,11 +134,11 @@ def main():
             if earliest:
                 # Ensure timezone-aware
                 if earliest.tzinfo is None:
-                    earliest = earliest.replace(tzinfo=timezone.utc)
+                    earliest = earliest.replace(tzinfo=UTC)
                 print(f"Earliest existing timestamp: {earliest.isoformat()}")
                 start_time = earliest - timedelta(seconds=args.interval)
             else:
-                start_time = datetime.now(timezone.utc)
+                start_time = datetime.now(UTC)
                 print(f"Table empty, starting from now: {start_time.isoformat()}")
 
             end_time = start_time - timedelta(hours=args.hours)

openc3/test/integration/tsdb/python/test_tsdb_roundtrip.py

@@ -28,7 +28,7 @@
 import sys
 import time
 import pytest
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from unittest.mock import patch
 
 # Add openc3 python path for imports
@@ -54,7 +54,7 @@ def ts_to_iso(ts_ns: int) -> str:
     """Convert nanosecond timestamp to ISO string, truncating to microseconds to match QuestDB storage."""
     ts_us = ts_ns // 1000
     ts_s = ts_us / 1_000_000
-    return datetime.fromtimestamp(ts_s, tz=timezone.utc).strftime(
+    return datetime.fromtimestamp(ts_s, tz=UTC).strftime(
         "%Y-%m-%dT%H:%M:%S.%fZ"
     )
 
@@ -652,7 +652,7 @@ def run_timestamp_test(
                 ), f"Expected ISO 8601 format at index {i}, got {actual}"
                 # Verify the formatted string can be parsed back to approximately the same time
                 parsed = datetime.strptime(actual, "%Y-%m-%dT%H:%M:%S.%fZ")
-                parsed_ts = parsed.replace(tzinfo=timezone.utc).timestamp()
+                parsed_ts = parsed.replace(tzinfo=UTC).timestamp()
                 assert (
                     abs(parsed_ts - expected_ts_seconds) < 0.001
                 ), f"Timestamp formatted mismatch at index {i}: expected {expected_ts_seconds}, got {actual}"