Has anyone integrated a Mantis with a Star instrument using a library for Venus 6?
I have a Mantis and a Star. I would like to use both instruments together.
I have an old vworks driver (GitHub - calico/vworks-mantis-plugin: Driver for controlling Formulatrix Mantis through VWorks · GitHub). I asked claude to refactor it into a CLI. I’m just going to paste it in here…
README
mantis_client.py - SOAP client wrapper using zeep that connects to http://localhost:7091/LiquidDispenser/Mantis?wsdl
(the same endpoint as the C# WCF service). Wraps all the service methods used by the original code.
mantis_cli.py - CLI entry point that mirrors the C# Program.cs logic exactly:
- help - show usage
- test - check if Mantis is in remote mode (returns exit code 1 for connected, 0 otherwise)
- clear errors - clear Mantis errors
- home - home the arms
- washall - wash all inputs
- washallchips - iterate all chips, attach/wash/detach each with correct volumes (150.4 for Low Volume, 754.0 for High
Volume) - primeall - prime all inputs with volume 250
- .dl.txt - load and execute a dispense list
- Interactive error recovery loop (R/H/C/X) on failure, matching the C# behavior
requirements.txt - just zeep>=4.2.1
venv/ - Python virtual environment with dependencies installed
To use on a new machine:
cd python
python -m venv venv
venv\Scripts\activate # Windows
pip install -r requirements.txt
python mantis_cli.py help
“”" mantis_client.py SOAP client wrapper for the Formulatrix Mantis liquid dispenser service.“”"
from enum import IntEnum
from zeep import Client
from zeep.transports import Transport
from requests import Session
WSDL_URL = “http://localhost:7091/LiquidDispenser/Mantis?wsdl”
class DispenseStatus(IntEnum):
READY = 0
READY_TO_DISPENSE = 1
BUSY = 2
ERROR = 3
class MantisClient:
def \__init_\_(self, wsdl_url=WSDL_URL):
session = Session()
transport = Transport(session=session, timeout=300, operation_timeout=300)
self.client = Client(wsdl_url, transport=transport)
self.service = self.client.service
def get_service_status(self, computer_name="localhost"):
return self.service.GetServiceStatus(computerName=computer_name)
def get_status(self):
raw = self.service.GetStatus()
return DispenseStatus(raw)
def get_status_blocking(self):
raw = self.service.GetStatusBlocking()
return DispenseStatus(raw)
def get_status_blocking_with_timeout(self, max_timeout):
raw = self.service.GetStatusBlockingWithTimeout(maxTimeout=max_timeout)
return DispenseStatus(raw)
def home_arms(self):
self.service.HomeArms()
def clear_error(self):
self.service.ClearError()
def stop(self):
self.service.Stop()
def run(self):
self.service.Run()
def wash_all(self):
self.service.WashAll()
def prime_all_with_volume(self, volume):
self.service.PrimeAllWithVolume(volume=volume)
def load_dispense_list(self, file_name):
self.service.LoadDispenseList(fileName=file_name)
def get_dispense_list_info(self):
return self.service.GetDispenseListInfo()
def get_last_dispensed_well_name(self):
return self.service.GetLastDispensedWellName()
def get_app_error_messages(self):
return self.service.GetAppErrorMessages()
def get_all_available_inputs(self):
return self.service.GetAllAvailableInputs()
def attach(self, input_number):
self.service.Attach(inputNumber=input_number)
def detach(self):
self.service.Detach()
def wash_double_with_volume(self, prime_volume, secondary_volume, rinse_volume):
self.service.WashDoubleWithVolume(
primeVolume=prime_volume,
secondaryVolume=secondary_volume,
rinseVolume=rinse_volume,
)
“”"mantis_cli.py Command-line interface for controlling the Formulatrix Mantis liquid dispenser.
Usage:
python mantis_cli.py help
python mantis_cli.py test
python mantis_cli.py home
python mantis_cli.py "clear errors"
python mantis_cli.py washall
python mantis_cli.py washallchips
python mantis_cli.py primeall
python mantis_cli.py <path_to_file.dl.txt>
“”"
import os
import sys
import time
import shutil
from mantis_client import MantisClient, DispenseStatus
def check_server_error(client):
"""Check for errors on the Mantis server. Returns True if errors were found."""
errors = client.get_app_error_messages()
if errors and len(errors) > 0:
print()
print("Error on server:")
for idx, err in enumerate(errors, 1):
print(f"{idx}. {err}")
input("Press Enter to clear error.")
client.clear_error()
return True
return False
def wait_hardware_busy(client, exe_time, print_position):
"""Poll the Mantis until it is no longer busy."""
time.sleep(5)
status = client.get_status()
print(f"Mantis is busy: {status == DispenseStatus.BUSY}")
while status == DispenseStatus.BUSY:
if print_position:
well = client.get_last_dispensed_well_name()
if well:
print(f"Dispensed position = {well}")
if exe_time > 50000:
time.sleep(exe_time / 100000) # C# sleeps ms, Python sleeps seconds
else:
time.sleep(0.05)
status = client.get_status()
if check_server_error(client):
return
print("WaitHardwareBusy Complete")
def cmd_help():
print("Pass the dispense list file name and path to execute a dispense.")
print("Pass CLEAR ERRORS to clear any errors.")
print("Pass TEST to check if the Mantis is in remote mode (returns 1 for remote mode, 0 for anything else).")
print("Pass HOME to home the arms.")
print("Pass WASHALL to Wash all inputs on both stations")
print("Pass WASHALLCHIPS to wash all chips individually")
print("Pass PRIMEALL to prime all inputs")
def cmd_test():
try:
client = MantisClient()
result = client.get_service_status("localhost")
print(f"Mantis Connected: {result}")
check_server_error(client)
return 1
except Exception:
print("Mantis is not connected")
return 0
def cmd_clear_error():
client = MantisClient()
try:
print()
client.clear_error()
print()
input("Press Enter to continue.")
except Exception as ex:
try:
if check_server_error(client):
return
except Exception:
on_unknown_error(str(ex))
def cmd_home():
client = MantisClient()
try:
print("Homing Mantis Arms...")
client.home_arms()
print()
print("Done.")
except Exception as ex:
try:
if check_server_error(client):
return
except Exception:
on_unknown_error(str(ex))
def cmd_wash_all():
client = MantisClient()
print("WashAll")
try:
print("Wash all try")
client.wash_all()
if check_server_error(client):
return
wait_hardware_busy(client, 10000, False)
except Exception as ex:
try:
if check_server_error(client):
return
except Exception:
on_unknown_error(str(ex))
def cmd_wash_all_chips():
client = MantisClient()
exe_time_s = 20 # 20000ms -> 20s
try:
for inp in client.get_all_available_inputs():
chip_serial = inp.ChipSerialNumber
chip_type = inp.ChipType
input_number = int(inp.InputNumber)
print(f'==> Starting on {chip_type} chip in slot {input_number} with serial number "{chip_serial}".')
if not chip_serial:
print("Chip has no serial number. Skipping.")
continue
print("Start attaching")
client.attach(input_number)
wait_hardware_busy(client, exe_time_s \* 1000, False)
if check_server_error(client):
return
print("Done attaching")
if chip_type == "Low Volume":
prime_vol = secondary_vol = rinse_vol = 150.4
elif chip_type == "High Volume":
prime_vol = secondary_vol = rinse_vol = 754.0
else:
print(f"Don't know what to do with a {chip_type} chip type")
print("Aborting.")
return
print(f"Start washing with primeVolume of {prime_vol}, secondaryVolume of {secondary_vol}, and rinseVolume of {rinse_vol}")
client.wash_double_with_volume(prime_vol, secondary_vol, rinse_vol)
wait_hardware_busy(client, exe_time_s \* 1000, False)
if check_server_error(client):
return
print("Done washing")
print("Start detaching")
client.detach()
wait_hardware_busy(client, exe_time_s \* 1000, False)
if check_server_error(client):
return
print("Done detaching")
print(f"==> Finished with {chip_type} chip in slot {input_number}")
if check_server_error(client):
return
wait_hardware_busy(client, 10000, False)
except Exception as ex:
try:
if check_server_error(client):
return
except Exception:
on_unknown_error(str(ex))
def cmd_prime_all(volume=250.0):
client = MantisClient()
print("PrimeAllWithVolume")
try:
client.prime_all_with_volume(volume)
if check_server_error(client):
return
wait_hardware_busy(client, 10000, False)
except Exception as ex:
try:
if check_server_error(client):
return
except Exception:
on_unknown_error(str(ex))
def cmd_dispense_list(dispense_list_path):
client = MantisClient()
try:
client.load_dispense_list(dispense_list_path)
if check_server_error(client):
return
dl_info = client.get_dispense_list_info()
if check_server_error(client):
return
exe_time = dl_info.EstimatedExecutionTime
status = client.get_status()
if status == DispenseStatus.READY_TO_DISPENSE:
client.run()
if check_server_error(client):
return
print(f"Loading Mantis dispense file: {dispense_list_path}")
print("Running .. wait")
wait_hardware_busy(client, int(exe_time), True)
else:
print()
print("Dispense not ready. Dispense prerequisites not met.")
input("Press Enter to continue.")
return
except Exception as ex:
try:
print(f"There was an error: {ex}")
if check_server_error(client):
return
except Exception:
on_unknown_error(str(ex))
def edit_dispense_list(dispense_list_path, well_list, volume):
"""Edit a dispense list file, replacing well volumes based on the well_list mask."""
if not os.path.exists(dispense_list_path):
print(f"The dispense list file does not exist: {dispense_list_path}")
input("Press Enter to continue.")
return
if len(well_list) != 96:
print(f"The well list does not have 96 values: {len(well_list)}")
input("Press Enter to continue.")
return
temp_path = dispense_list_path + "temp"
backup_path = dispense_list_path + "backup"
with open(dispense_list_path, "r") as sr, open(temp_path, "w") as sw:
\# Copy first 7 header lines
for \_ in range(7):
line = sr.readline()
sw.write(line)
\# Write 8 rows x 12 columns of volume data
for i in range(8):
for j in range(12):
idx = (i \* 12) + j
if well_list\[idx\] == "1":
sw.write(str(volume))
else:
sw.write("0")
sw.write("\\t")
sw.write("\\n")
\# Replace original with temp, keeping a backup
if os.path.exists(backup_path):
os.remove(backup_path)
shutil.copy2(dispense_list_path, backup_path)
shutil.move(temp_path, dispense_list_path)
print("Dispense list edited...")
input("Press Enter to continue.")
def on_unknown_error(message):
print()
print(f"Unknown error: \\n{message}")
input("Press Enter to continue.")
def error_recovery_loop(args):
"""Interactive error recovery menu matching the C# behavior."""
while True:
try:
print("You have the following options:")
print("- Retry the initial command (R)")
print("- Home the Mantis (H)")
print("- Attempt to clear the software error on the Mantis (C)")
print("- Exit this application and leave the Mantis as is (X)")
response = input("Type a letter to continue... ").strip().upper()
if response == "R":
print()
print("Retrying Command...")
return main(args)
if response == "H":
print()
print("Homing Arms...")
cmd_home()
if response == "C":
print()
print("Clearing Error...")
cmd_clear_error()
if response == "X":
print()
print("Continuing...")
return 0
except Exception as nx:
print(f"There was another error: {nx}")
def main(args=None):
if args is None:
args = sys.argv\[1:\]
try:
arguments = " ".join(args)
if arguments.upper() == "HELP":
cmd_help()
return 0
if arguments.upper() == "CLEAR ERRORS":
cmd_clear_error()
return 0
if arguments.upper() == "TEST":
return cmd_test()
if arguments.upper() == "HOME":
cmd_home()
return 0
if arguments.upper() == "WASHALL":
cmd_wash_all()
return 0
if arguments.upper() == "WASHALLCHIPS":
cmd_wash_all_chips()
return 0
if arguments.upper() == "PRIMEALL":
cmd_prime_all(250)
return 0
if ".dl.txt" in arguments:
cmd_dispense_list(arguments)
return 0
print("Unknown command. Run with 'help' for usage.")
return 0
except Exception as ex:
print(f"There was an error with the operation: {ex}")
return error_recovery_loop(args)
if _name_ == “_main_”:
sys.exit(main())
Hi,
I have used the Https library for sending/receiving commands from Venus to the rest API of Formulatrix F.A.S.T., Without knowing anything about the mantis, would something similar be an alternative?
I would guess that this is as simple as it gets, while also using a library.