chore(hardware): typing fixups

Author: sfoster1

hardware/opentrons_hardware/firmware_bindings/messages/binary_message_definitions.py

@@ -1,7 +1,7 @@
 """Definition of Binary messages."""
 from dataclasses import dataclass, field, fields, MISSING
 from functools import lru_cache
-from typing import Type, Union, Optional
+from typing import Type, Union, Optional, cast
 from typing_extensions import get_args
 
 from ..binary_constants import BinaryMessageId, LightTransitionType, LightAnimationType
@@ -28,7 +28,7 @@ class Echo(utils.BinarySerializable):
     )
     length: utils.UInt16Field = field(default_factory=lambda: utils.UInt16Field(0))
     message: utils.BinaryFieldBase[bytes] = field(
-        default_factory=utils.BinaryFieldBase(bytes())
+        default_factory=lambda: utils.BinaryFieldBase(bytes())
     )
 
 
@@ -348,7 +348,7 @@ class WriteEEPromRequest(utils.BinarySerializable):
         default_factory=lambda: utils.UInt16Field(0)
     )
     data_length: utils.UInt16Field = field(default_factory=lambda: utils.UInt16Field(0))
-    data: EepromDataField = field(default_factory=EepromDataField(bytes()))
+    data: EepromDataField = field(default_factory=lambda: EepromDataField(bytes()))
 
 
 @dataclass
@@ -377,7 +377,7 @@ class ReadEEPromResponse(utils.BinarySerializable):
         default_factory=lambda: utils.UInt16Field(0)
     )
     data_length: utils.UInt16Field = field(default_factory=lambda: utils.UInt16Field(0))
-    data: EepromDataField = field(default_factory=EepromDataField(bytes()))
+    data: EepromDataField = field(default_factory=lambda: EepromDataField(bytes()))
 
 
 @dataclass
@@ -396,7 +396,7 @@ class AddLightActionRequest(utils.BinarySerializable):
         default_factory=lambda: utils.UInt16Field(0)
     )
     transition_type: LightTransitionTypeField = field(
-        default_factory=LightTransitionTypeField(LightTransitionType.linear)
+        default_factory=lambda: LightTransitionTypeField(LightTransitionType.linear)
     )
     red: utils.UInt8Field = field(default_factory=lambda: utils.UInt8Field(0))
     green: utils.UInt8Field = field(default_factory=lambda: utils.UInt8Field(0))
@@ -526,11 +526,11 @@ def get_binary_definition(
                 message_type_field.default is not MISSING
                 and message_type_field.default.value == message_id
             ):
-                return i
+                return cast(Type[BinaryMessageDefinition], i)
             if (
                 message_type_field.default_factory is not MISSING
                 and message_type_field.default_factory().value == message_id
             ):
-                return i
+                return cast(Type[BinaryMessageDefinition], i)
     log.error("No binary message definition found.")
     return None

hardware/opentrons_hardware/firmware_bindings/messages/payloads.py

@@ -57,7 +57,7 @@ def __eq__(self, other: object) -> bool:
     # to work around this in binary_serializable.build() and can_comm.prompt_payload
     # we ignore the message_index when constructing args and then set the value manually after
     message_index: utils.UInt32Field = field(
-        init=False, default_factory=lambda: utils.UInt32Field(None)
+        init=False, default_factory=lambda: utils.UInt32Field(0)
     )
 
 

hardware/opentrons_hardware/instruments/serial_utils.py

@@ -13,4 +13,4 @@ def ensure_serial_length(
 
 def model_versionstring_from_int(model: int) -> str:
     """Format the encoded model from an int into a dotted version string."""
-    return f"{model//10}.{model%10}"
+    return f"{model // 10}.{model % 10}"

hardware/opentrons_hardware/scripts/load.py

@@ -72,7 +72,10 @@ async def run(
         """Run a probe and return detected devices."""
         try:
             messenger.add_listener(self)
-            await asyncio.wait([self._stimulate(messenger)], timeout=response_timeout_s)
+            await asyncio.wait(
+                [asyncio.create_task(self._stimulate(messenger))],
+                timeout=response_timeout_s,
+            )
         except asyncio.TimeoutError:
             pass
         finally:
@@ -189,7 +192,7 @@ async def run(self, run_for: float) -> LoadResults:
         )
         try:
             self._messenger.add_listener(self)
-            await asyncio.wait([self._load_bus()], timeout=run_for)
+            await asyncio.wait([asyncio.create_task(self._load_bus())], timeout=run_for)
         except asyncio.TimeoutError:
             pass
         except KeyboardInterrupt: