spectramax.py

"""
Spectramax Plate Reader Controller
Python interface for controlling Molecular Devices SpectraMax plate readers
via the SoftMax Pro Software Automation API
"""

import sys
import threading
from typing import Optional, Callable, Dict

# Add .NET assemblies path - adjust this path according to your installation
# Default paths from the documentation:
# 32-bit: C:\Program Files\Molecular Devices\SoftMax Pro 7.0 Automation SDK
# 64-bit: C:\Program Files (x86)\Molecular Devices\SoftMax Pro 7.0 Automation SDK
AUTOMATION_SDK_PATH = (
    r"C:\Program Files (x86)\Molecular Devices\SoftMax Pro 7.0 Automation SDK"
)

try:
    import clr

    # Add reference to the automation assemblies
    clr.AddReference("System")
    sys.path.append(AUTOMATION_SDK_PATH)
    clr.AddReference("SoftMaxPro.AutomationClient")
    clr.AddReference("SoftMaxPro.AutomationInterface")
    clr.AddReference("SoftMaxPro.AutomationExtensions")

    # Import the automation client
    from SoftMaxPro.AutomationClient import SMPAutomationClient

    print("Successfully loaded SoftMax Pro automation assemblies")

except ImportError as e:
    print(f"Error importing .NET assemblies: {e}")
    print("Make sure pythonnet is installed: pip install pythonnet")
    print(f"And verify the automation SDK path: {AUTOMATION_SDK_PATH}")
    sys.exit(1)


class Spectramax:
    """
    Python wrapper for SoftMax Pro Automation API
    Provides high-level control of SpectraMax plate readers
    """

    def __init__(self, server: str = "localhost", port: int = 9000):
        """
        Initialize the Spectramax controller

        Args:
            server: SoftMax Pro server address (default: localhost)
            port: Server port (default: 9000)
        """
        self.server = server
        self.port = port
        self.client = None
        self.connected = False
        self.command_callbacks: Dict[int, Callable] = {}
        self.last_command_id = -1

        # Event flags
        self.command_completed_event = threading.Event()
        self.instrument_status = "UNKNOWN"

    def initialize(self) -> bool:
        """
        Initialize connection to SoftMax Pro software

        Returns:
            bool: True if connection successful, False otherwise
        """
        try:
            self.client = SMPAutomationClient()

            # Subscribe to events before initializing
            self.client.CommandCompleted += self._on_command_completed
            self.client.ErrorReport += self._on_error_report
            self.client.InstrumentStatusChanged += self._on_instrument_status_changed

            # Initialize connection
            if self.port != 9000:
                self.connected = self.client.Initialize(self.server, self.port)
            else:
                self.connected = self.client.Initialize(self.server)

            if self.connected:
                print(
                    f"Successfully connected to SoftMax Pro at {self.server}:{self.port}"
                )
            else:
                print("Failed to connect to SoftMax Pro")

            return self.connected

        except Exception as e:
            print(f"Connection error: {e}")
            return False

    def disconnect(self):
        """Disconnect from SoftMax Pro and clean up resources"""
        if self.client and self.connected:
            try:
                # Unsubscribe from events
                self.client.CommandCompleted -= self._on_command_completed
                self.client.ErrorReport -= self._on_error_report
                self.client.InstrumentStatusChanged -= (
                    self._on_instrument_status_changed
                )

                # Dispose the client
                self.client.Dispose()
                print("Disconnected from SoftMax Pro")

            except Exception as e:
                print(f"Error during disconnect: {e}")
            finally:
                self.connected = False
                self.client = None

    def _on_command_completed(self, sender, e):
        """Handle command completion events"""
        print(f"Command {e.QueueID} completed")
        if e.StringResult:
            print(f"Result: {e.StringResult}")
        if e.IntResult != 0:
            print(f"Int Result: {e.IntResult}")
        if hasattr(e, "DoubleResult") and e.DoubleResult != 0:
            print(f"Double Result: {e.DoubleResult}")

        # Call registered callback if exists
        if e.QueueID in self.command_callbacks:
            callback = self.command_callbacks.pop(e.QueueID)
            callback(e)

        self.last_command_id = e.QueueID
        self.command_completed_event.set()

        if e.QueueEmpty:
            print("Command queue is empty")

    def _on_error_report(self, sender, e):
        """Handle error events"""
        print(f"ERROR - Command {e.QueueID}: {e.Error}")

    def _on_instrument_status_changed(self, sender, e):
        """Handle instrument status change events"""
        self.instrument_status = e.Status
        print(f"Instrument status changed to: {e.Status}")

    def _wait_for_command(self, timeout: float = 30.0) -> bool:
        """
        Wait for the last command to complete

        Args:
            timeout: Maximum time to wait in seconds

        Returns:
            bool: True if command completed, False if timeout
        """
        self.command_completed_event.clear()
        return self.command_completed_event.wait(timeout)

    def open_drawer(self) -> bool:
        """
        Open the plate drawer

        Returns:
            bool: True if operation successful
        """
        if not self.connected:
            print("Not connected to SoftMax Pro")
            return False

        try:
            self.client.OpenDrawer()
            print("Opening drawer...")
            self._wait_for_command()
            return True
        except Exception as e:
            print(f"Error opening drawer: {e}")
            return False

    def close_drawer(self) -> bool:
        """
        Close the plate drawer

        Returns:
            bool: True if operation successful
        """
        if not self.connected:
            print("Not connected to SoftMax Pro")
            return False

        try:
            self.client.CloseDrawer()
            print("Closing drawer...")
            self._wait_for_command()
            return True
        except Exception as e:
            print(f"Error closing drawer: {e}")
            return False

    def read_UVVis(self, method: str, results: str) -> bool:
        """
        Perform a UV-Vis read using the specified method and save results

        Args:
            method: File path to the method file (.spr protocol file)
            results: Base file path for results (without extension)

        Returns:
            bool: True if read and save operations successful
        """
        if not self.connected:
            print("Not connected to SoftMax Pro")
            return False

        try:
            print(f"Starting UV-Vis read with method: {method}")
            print(f"Results will be saved to: {results}")

            # Step 1: Close drawer
            print("Closing drawer...")
            self.client.CloseDrawer()
            self._wait_for_command()

            # Step 2: Close all documents
            print("Closing all documents...")
            self.client.CloseAllDocuments()
            self._wait_for_command()

            # Step 3: Open the method file
            print(f"Opening method file: {method}")
            self.client.OpenFile(method)
            self._wait_for_command()

            # Step 4: Start the read
            print("Starting read operation...")
            command_id = self.client.StartRead()
            print(f"Read started (Command ID: {command_id})")
            self._wait_for_command(
                timeout=120.0
            )  # Extended timeout for read operations

            # Step 5: Save as .sda file
            saveas_spr = results + ".sda"
            print(f"Saving as SoftMax Pro data file: {saveas_spr}")
            self.client.SaveAs(saveas_spr)
            self._wait_for_command()

            # Step 6: Export as CSV (COLUMNS format)
            saveas_csv = results + ".csv"
            print(f"Exporting as CSV: {saveas_csv}")
            self.client.ExportAs(saveas_csv, SMPAutomationClient.ExportAsFormat.COLUMNS)
            self._wait_for_command()

            # Step 7: Export as XML
            saveas_xml = results + ".xml"
            print(f"Exporting as XML: {saveas_xml}")
            self.client.ExportAs(saveas_xml, SMPAutomationClient.ExportAsFormat.XML)
            self._wait_for_command()

            print("UV-Vis read completed successfully")
            print("Files saved:")
            print(f"  - Data file: {saveas_spr}")
            print(f"  - CSV export: {saveas_csv}")
            print(f"  - XML export: {saveas_xml}")

            return True

        except Exception as e:
            print(f"Error during UV-Vis read: {e}")
            return False

    # Additional utility methods

    def setup_reader(
        self,
        port: str = "Offline",
        instrument: str = "SPECTRAmax M5",
        simulation: bool = True,
    ) -> bool:
        """
        Configure the plate reader

        Args:
            port: Communication port (e.g., "COM1" or "Offline")
            instrument: Instrument model name
            simulation: Enable simulation mode

        Returns:
            bool: True if setup successful
        """
        if not self.connected:
            print("Not connected to SoftMax Pro")
            return False

        try:
            # Set reader type
            self.client.SetReader(port, instrument)
            self._wait_for_command()

            # Enable/disable simulation mode
            self.client.SetSimulationMode(simulation)
            self._wait_for_command()

            print(
                f"Reader setup complete: {instrument} on {port}, simulation={simulation}"
            )
            return True

        except Exception as e:
            print(f"Reader setup error: {e}")
            return False

    def create_new_document(self) -> bool:
        """Create a new SoftMax Pro document"""
        if not self.connected:
            return False

        try:
            self.client.NewDocument()
            self._wait_for_command()
            print("New document created")
            return True
        except Exception as e:
            print(f"Error creating document: {e}")
            return False

    def select_plate_section(self, section_name: str = "Plate1") -> bool:
        """
        Select a plate section for reading

        Args:
            section_name: Name of the plate section

        Returns:
            bool: True if selection successful
        """
        if not self.connected:
            return False

        try:
            self.client.SelectSection(section_name)
            self._wait_for_command()
            print(f"Selected section: {section_name}")
            return True
        except Exception as e:
            print(f"Error selecting section: {e}")
            return False

    def get_data(self) -> Optional[str]:
        """
        Get the current plate data

        Returns:
            str: Plate data as tab-delimited string, or None if error
        """
        if not self.connected:
            return None

        try:
            # Store result in a container we can access from callback
            result_container = {"data": None}

            def data_callback(e):
                result_container["data"] = e.StringResult

            command_id = self.client.GetDataCopy()
            self.command_callbacks[command_id] = data_callback

            if self._wait_for_command():
                return result_container["data"]
            else:
                print("Timeout waiting for data")
                return None

        except Exception as e:
            print(f"Get data error: {e}")
            return None

    def set_temperature(self, temperature: float) -> bool:
        """
        Set the incubator temperature

        Args:
            temperature: Temperature in Celsius (0 to turn off)

        Returns:
            bool: True if successful
        """
        if not self.connected:
            return False

        try:
            self.client.SetTemperature(temperature)
            self._wait_for_command()
            print(f"Temperature set to {temperature}°C")
            return True
        except Exception as e:
            print(f"Temperature setting error: {e}")
            return False

    def get_instrument_status(self) -> str:
        """
        Get current instrument status

        Returns:
            str: Current instrument status
        """
        return self.instrument_status

    def save_data(self, file_path: str) -> bool:
        """
        Save the current document

        Args:
            file_path: Full path for saving (use .sda extension for data files)

        Returns:
            bool: True if save successful
        """
        if not self.connected:
            return False

        try:
            self.client.SaveAs(file_path)
            self._wait_for_command()
            print(f"Data saved to: {file_path}")
            return True
        except Exception as e:
            print(f"Save error: {e}")
            return False

    def export_data(self, file_path: str, format_type: str = "XML") -> bool:
        """
        Export data in specified format

        Args:
            file_path: Full path for export
            format_type: Export format ("XML", "PLATE", "COLUMNS")

        Returns:
            bool: True if export successful
        """
        if not self.connected:
            return False

        try:
            # Map format strings to enum values
            format_map = {
                "XML": SMPAutomationClient.ExportAsFormat.XML,
                "PLATE": SMPAutomationClient.ExportAsFormat.PLATE,
                "COLUMNS": SMPAutomationClient.ExportAsFormat.COLUMNS,
                "TIME": SMPAutomationClient.ExportAsFormat.TIME,
            }

            export_format = format_map.get(format_type.upper())
            if export_format is None:
                print(f"Invalid format: {format_type}")
                return False

            self.client.ExportAs(file_path, export_format)
            self._wait_for_command()
            print(f"Data exported to: {file_path}")
            return True
        except Exception as e:
            print(f"Export error: {e}")
            return False