open and close actions added, not yet working

Author: caseystone

src/bmg_driver.py

@@ -0,0 +1,69 @@
+import ctypes
+import pythoncom
+import comtypes.client
+
+class BmgCom:
+   def __init__(self, control_name=None):
+       pythoncom.CoInitialize()
+       self.com = comtypes.client.CreateObject("BMG_ActiveX.BMGRemoteControl")
+       if control_name:
+           self.open(control_name)
+
+   def open(self, control_name):
+       ep = ctypes.c_char_p(control_name.encode('ascii'))
+       res = self.com.OpenConnection(ep)
+       if res:
+           raise Exception(f"OpenConnection failed: {res}")
+
+   def version(self):
+       version = self.com.GetVersion()
+       return version
+
+   def dummy(self):
+       self.exec('Dummy')
+
+   def status(self):
+       item = ctypes.c_char_p(b"Status")
+       status = self.com.GetInfo(item)
+       return status.strip() if isinstance(status, str) else 'unknown'
+
+   def error(self):
+       item = ctypes.c_char_p(b"Error")
+       status = self.com.GetInfo(item)
+       return status.strip() if isinstance(status, str) else 'unknown'
+
+   def init(self):
+       self.exec('Init')
+
+   def plate_in(self):
+       self.exec('PlateIn')
+
+   def plate_out(self):
+       self.exec('PlateOut')
+
+   def run(self, protocol_name, ):
+       self.exec('Run', "RNA", "C:\\Program Files (x86)\\BMG\\CLARIOstar\\User\\Definit", "C:\\Program Files (x86)\\BMG\\CLARIOstar\\User\\Data", 0, 1, 2, "C:\\Program Files (x86)\\BMG\\CLARIOstar\\User\\Data", "test2.txt")
+
+   def isBusy(self):
+       return self.status() == 'Busy'
+
+   def exec(self, cmd, *args):
+       args = tuple((cmd, *args))
+       print(args)
+       res = self.com.ExecuteAndWait(args)
+       if res:
+           raise Exception(f"command {cmd} failed: {res}")
+
+if __name__ == '__main__':
+   com = BmgCom("CLARIOstar")
+   print(f"BMG LABTECH Remote Control Version: {com.version()}")
+   com.dummy()
+   print(f"Status: {com.status()}")
+
+   # TESTING  command
+   #com.plate_out()
+   com.plate_in()
+   #com.exec("Run", )
+   #com.init()
+#    print(f"Status: {com.status()}")
+#    print(com.run())

src/bmg_rest_node.py

@@ -0,0 +1,125 @@
+"""
+REST-based node that interfaces with WEI and provides a simple Sleep(t) function
+"""
+
+import time
+from typing import Annotated
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+
+from fastapi.datastructures import UploadFile
+from starlette.datastructures import State
+
+# # from typing_extensions import Annotated
+from wei.modules.rest_module import RESTModule
+from wei.types.module_types import (
+    LocalFileModuleActionResult,
+    ModuleAction,
+    ModuleActionArg,
+    ModuleState,
+    ModuleStatus,
+    ValueModuleActionResult,
+)
+from wei.types.step_types import (
+    ActionRequest,
+    StepFileResponse,
+    StepResponse,
+    StepStatus,
+)
+
+from bmg_driver import BmgCom  # import the bmg driver
+
+rest_module = RESTModule(
+    name="bmg_module",
+    verson="0.0.1",
+    description="A REST node to control the BMG VANTAstar microplate reader",
+    model="bmg",
+)
+rest_module.arg_parser.add_argument(
+    "--output_path", type=str, help="data ourtput directory path for bmg data", default="C:\\Program Files (x86)\\BMG\\CLARIOstar\\User\\Data"
+)
+
+@rest_module.startup()
+def bmg_startup(state: State):
+    """BMG startup handler"""
+
+    try:
+        # Connect to the BMG
+        state.bmg = BmgCom("CLARIOstar")
+
+        # TODO: Set module status
+        # state.status = check_state(state)
+        # print(state.status)
+        # QUESTION: Why does state.status[ModuleStatus.READY] = True work for the pf400 but not for my bmg module
+        # state.status = state.bmg.status()
+        # print(state.status)
+        # state.status[ModuleStatus.READY] = True if state.bmg.status() == "Ready" else False
+        # print(state.status[ModuleStatus.READY])
+        # print(state.bmg.status())
+
+    except Exception:
+        print("Something went wrong")
+        raise
+
+    else:
+        print("BMG online")
+
+
+# @rest_module.state_handler()
+# def check_state(state: State) -> ModuleState:
+#     """Updates the BMG state"""
+
+#     bmg_state = state.bmg.status()
+
+#     if bmg_state == "Ready":
+#         return ModuleStatus.READY
+#     else:
+#         return ModuleStatus.ERROR
+
+
+# OPEN TRAY ACTION 
+@rest_module.action(
+    name="open", description="Open the bmg plate tray"
+)
+def open(
+    state: State,
+    action: ActionRequest,
+) -> StepResponse:
+    """Opens the BMG plate tray"""
+    state.bmg.plate_out()
+    return StepResponse.step_succeeded()
+
+
+# CLOSE TRAY ACTION
+@rest_module.action(
+    name="close", description="Close the BMG plate tray"
+)
+def close(
+    state: State,
+    action: ActionRequest,
+) -> StepResponse:
+    """Closes the BMG plate tray"""
+    state.bmg.plate_in()
+    return StepResponse.step_succeeded()
+
+
+# RUN ASSAY ACTION
+@rest_module.action(
+    name="run_assay", description="run an assay on the BMG VANTAstar plate reader"
+)
+def run_assay(
+    state: State,
+    action: ActionRequest,
+    assay_name: 
+) -> StepFileResponse: 
+    """Runs an assay on the BMG plate reader"""
+    state.bmg.run()
+
+
+rest_module.start()
+
+
+
+
+
+