fixes typing and node info

Author: abestroka

src/opentrons_resources.py

@@ -12,9 +12,9 @@ def __init__(self):
         Initialize the Opentrons Resources manager
         """
         self.client = ResourceClient("http://localhost:8003")
-        self.deck_slots: Dict[str, Any] #map slot numbers to resources
-        self.pipette_slots: Dict[str, Any]
-        self.node_name: str
+        self.deck_slots: Dict[str, Any] = {}#map slot numbers to resources
+        self.pipette_slots: Dict[str, Any] = {}
+        self.node_name: str = ""
 
         # dicts for mapping OT2 labware IDs to MADSci resources
         self.labware_id_to_resource: Dict[str, Any] = {}

src/ot2_rest_node.py

@@ -38,26 +38,26 @@ def startup_handler(self) -> None:
         # temp_dir = Path.home() / ".madsci" / ".ot2_temp"
         # temp_dir.mkdir(exist_ok=True)
         # self.protocols_folder_path = str(
-        #     temp_dir / self.node_definition.node_name / "protocols/"
+        #     temp_dir / self.node_info.node_name / "protocols/"
         # )
 
         OT2_TEMP_DIR.mkdir(parents=True, exist_ok=True)
         self.protocols_folder_path = str(
-            OT2_TEMP_DIR / self.node_definition.node_name / "protocols/"
+            OT2_TEMP_DIR / self.node_info.node_name / "protocols/"
         )
         # Create templates
         self._create_ot2_templates()
 
         # Create deck instance
         self.deck = self.resource_client.create_resource_from_template(
             template_name="ot2_deck",
-            resource_name=f"ot2_{self.node_definition.node_name}_deck",
+            resource_name=f"ot2_{self.node_info.node_name}_deck",
             add_to_database=True,
         )
 
         # Create 12 deck slots (1-11 standard, 12 is trash)
         for i in range(1, 13):
-            slot_name = f"ot2_{self.node_definition.node_name}_deck_slot_{i}"
+            slot_name = f"ot2_{self.node_info.node_name}_deck_slot_{i}"
             template_name = "ot2_trash_slot" if i == 12 else "ot2_deck_slot"
 
             slot = self.resource_client.create_resource_from_template(
@@ -76,7 +76,7 @@ def startup_handler(self) -> None:
         for mount in ["left", "right"]:
             mount_slot = self.resource_client.create_resource_from_template(
                 template_name="ot2_pipette_mount",
-                resource_name=f"ot2_{self.node_definition.node_name}_{mount}_mount",
+                resource_name=f"ot2_{self.node_info.node_name}_{mount}_mount",
                 add_to_database=True,
             )
             self.pipette_slots[mount] = mount_slot
@@ -117,7 +117,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 deck container. Holds 11 deck slots plus trash bin.",
             required_overrides=["resource_name"],
             tags=["ot2", "deck", "container"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -139,7 +139,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 deck slot. Standard SBS-compatible position.",
             required_overrides=["resource_name"],
             tags=["ot2", "deck", "slot"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -160,7 +160,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 trash bin slot.",
             required_overrides=["resource_name"],
             tags=["ot2", "trash", "stack"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -181,7 +181,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 pipette mount slot.",
             required_overrides=["resource_name"],
             tags=["ot2", "pipette", "mount", "slot"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -208,7 +208,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 P20 Single-Channel pipette (1-20 µL).",
             required_overrides=["resource_name"],
             tags=["ot2", "pipette", "p20", "single-channel", "pool"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -241,7 +241,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 P300 Single-Channel pipette (20-300 µL).",
             required_overrides=["resource_name"],
             tags=["ot2", "pipette", "p300", "single-channel", "pool"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -277,7 +277,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 P1000 Single-Channel pipette (100-1000 µL).",
             required_overrides=["resource_name"],
             tags=["ot2", "pipette", "p1000", "single-channel", "pool"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -308,7 +308,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 P20 8-Channel pipette (1-20 µL).",
             required_overrides=["resource_name"],
             tags=["ot2", "pipette", "p20", "8-channel", "multi-channel", "pool"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -345,7 +345,7 @@ def _create_ot2_templates(self) -> None:
             description="Template for OT2 P300 8-Channel pipette (20-300 µL).",
             required_overrides=["resource_name"],
             tags=["ot2", "pipette", "p300", "8-channel", "multi-channel", "pool"],
-            created_by=self.node_definition.node_id,
+            created_by=self.node_info.node_id,
             version="1.0.0",
         )
 
@@ -480,7 +480,7 @@ def execute(self, protocol_path, payload=None, resource_config=None):
             protocol_id, run_id = self.ot2_interface.transfer(protocol_file_path)
             self.logger.log(
                 "OT2 "
-                + self.node_definition.node_name
+                + self.node_info.node_name
                 + " protocol transfer successful"
             )
 
@@ -499,10 +499,10 @@ def execute(self, protocol_path, payload=None, resource_config=None):
 
             resp = self.ot2_interface.execute(run_id)
 
-            # log_dir = Path.home() / ".madsci" / ".ot2_temp" / self.node_definition.node_name / "logs"
+            # log_dir = Path.home() / ".madsci" / ".ot2_temp" / self.node_info.node_name / "logs"
             # log_dir.mkdir(parents=True, exist_ok=True)
             # self.last_log_path = log_dir / f"run_{run_id}_log.json"
-            log_dir = OT2_TEMP_DIR / self.node_definition.node_name / "logs"
+            log_dir = OT2_TEMP_DIR / self.node_info.node_name / "logs"
             log_dir.mkdir(parents=True, exist_ok=True)
             self.last_log_path = log_dir / f"run_{run_id}_log.json"
 
@@ -520,39 +520,39 @@ def execute(self, protocol_path, payload=None, resource_config=None):
                 # poll_OT2_until_run_completion()
                 self.logger.log(
                     "OT2 "
-                    + self.node_definition.node_name
+                    + self.node_info.node_name
                     + " succeeded in executing a protocol"
                 )
                 response_msg = (
                     "OT2 "
-                    + self.node_definition.node_name
+                    + self.node_info.node_name
                     + " successfully IDLE running a protocol"
                 )
                 return "succeeded", response_msg, run_id
 
             elif resp["data"]["status"] == "stopped":
                 self.logger.log(
                     "OT2 "
-                    + self.node_definition.node_name
+                    + self.node_info.node_name
                     + " stopped while executing a protocol"
                 )
                 response_msg = (
                     "OT2 "
-                    + self.node_definition.node_name
+                    + self.node_info.node_name
                     + " successfully IDLE after stopping a protocol"
                 )
                 return "stopped", response_msg, run_id
 
             else:
                 self.logger.log(
                     "OT2 "
-                    + self.node_definition.node_name
+                    + self.node_info.node_name
                     + " failed in executing a protocol"
                 )
                 self.logger.log(resp["data"])
                 response_msg = (
                     "OT2 "
-                    + self.node_definition.node_name
+                    + self.node_info.node_name
                     + " failed running a protocol\n"
                     + str(resp["data"])
                 )