""" MAS100Libra Control Library Author: Lucas Armbrecht Created: 2025-01-10 Last Modified: 2025-02-13 Description: This library provides functions to control the MAS-100 Libra instrument using Modbus TCP. Dependencies: - pymodbus: Install using `pip install pymodbus` Usage: from pymodbus.client import ModbusTcpClient mas100 = MAS100Libra('192.168.1.100')S mas100.get_device_state() ... Classes: MAS100Libra - A class to interact with the MAS-100 Libra device. Methods: __init__(self, ip) - Initializes the Modbus TCP client, input parameter IP must be given in format '192.168.1.10'. close_connection(self) - Closes the Modbus TCP connection. get_serial_number(self) - Retrieves the serial number of the instrument. get_instrument_label(self) - Retrieves the instrument label of the instrument. get_instrument_hw_version(self) - Retrieves the hardware version of the instrument. get_instrument_fw_version(self) - Retrieves the firmware version of the instrument. get_instrument_fw_build(self) - Retrieves the firmware build version of the instrument. init_device(self) - Initializes the instrument. start_device(self) - Starts the instrument. stop_device(self) - Stops the instrument. next_plate(self) - triggers acoverExposedPlate process and exposure of next loaded plate. acknowledge_error(self) - Acknowledges any errors. start_decontamination(self) - Starts the decontamination process. Only possible from ready state. stop_decontamination(self) - Stops the decontamination process and returns to power on state. confirm_intervention(self) - Confirms user intervention. get_remaining_plates(self) - Gets the number of remaining plates. get_finished_plates(self) - Gets the number of finished plates. get_remaining_sampling_time(self) - Gets the remaining sampling time in minutes. get_device_state(self) - Gets the current state of the device. get_error_states(self) - Gets the current error states. write_plate_id(self, plate_barcode, plate_position) - Writes a plate barcode to a specified (i.e. slot 1 = first exposed plate). _get_single_register_value(self, address) - Helper method to get a single register value. _get_string_from_registers(self, start_address, count) - Helper method to get a string from registers. write_coil(self, address) - Helper method to write a coil value. read_registers(self, start_address, count) - Helper method to read registers. extract_string_from_registers(self, registers) - Helper method to extract a string from registers. """ from pymodbus.client import ModbusTcpClient class MAS100Libra: def __init__(self, ip): self.client = ModbusTcpClient(ip) def close_connection(self): self.client.close() def get_serial_number(self): return self._get_string_from_registers(100, 8) def get_instrument_label(self): return self._get_string_from_registers(107, 9) def get_instrument_hw_version(self): return self._get_string_from_registers(116, 8) def get_instrument_fw_version(self): return self._get_string_from_registers(123, 8) def get_instrument_fw_build(self): return self._get_string_from_registers(131, 8) def init_device(self): self.write_coil(0) def start_device(self): self.write_coil(1) def stop_device(self): self.write_coil(2) def next_plate(self): self.write_coil(3) def acknowledge_error(self): self.write_coil(4) def start_decontamination(self): self.write_coil(5) def stop_decontamination(self): self.write_coil(6) def confirm_intervention(self): self.write_coil(7) def get_remaining_plates(self): return self._get_single_register_value(3) def get_finished_plates(self): return self._get_single_register_value(4) def get_remaining_sampling_time(self): return self._get_single_register_value(5) def get_device_state(self): state_map = { 0: 'PowerOff', 1: 'PowerOn', 2: 'Error', 3: 'Init', 4: 'Ready', 5: 'ExposePlate', 6: 'Sampling', 7: 'CoverExposedPlate', 8: 'Finished', 9: 'DecoMode', 10: 'Instrument check', 11: 'Timer check', 12: 'Awaiting user intervention' } result = self.read_registers(6, 1) return state_map.get(result[0], 'Unknown device state') if result else 'Error reading the register' def get_error_states(self): error_codes = [ "na", "na", "na", "na", "na", "na", "na", "na", "na", "na", "na", "na", "na", "E-1201: Not homed", "E-1200: Unexpected limit sensor triggered", "E-1202: Axis motion timeout", "na", "na", "na", "na", "na", "na", "na", "na", "na", "na", "na", "na", "na", "W-2103: NTP server did not respond", "W-2101: No DNS lookup is available", "W-2102: Failed to resolve NTP server address", "na", "na", "na", "na", "na", "na", "W-8803: User intervention needed", "E-8803: Error during plate shuffle back", "E-6404: Plate stack loaded to wrong plate carrier", "E-6403: Plates present during Initialization", "E-8802: Plate lost during handling", "E-8812: Plate lost during handling", "E-8801: Lid lost during handling", "E-8811: Lid lost during handling", "E-6402: Plate count exceeds limit", "E-6401: No plates loaded" ] results = [self.read_registers(i, 1) for i in range(3)] if all(results): binary_string = ''.join(format(result[0], '016b') for result in results) return [error_codes[i] for i, bit in enumerate(binary_string) if bit == '1' and i < len(error_codes)] return [] def write_plate_id(self, plate_barcode, plate_position): registers = [(ord(plate_barcode[i]) << 8) | (ord(plate_barcode[i + 1]) if i + 1 < len(plate_barcode) else 0) for i in range(0, len(plate_barcode), 2)] registers.append(0) # Null terminator start_address = (plate_position * 25) -25 response = self.client.write_registers(start_address, registers[:25], slave=1) if response.isError(): print(f"Failed to write registers. Response: {response}") else: print(f"Successfully wrote PlateId to slot {plate_position}: {plate_barcode}") def _get_single_register_value(self, address): result = self.read_registers(address, 1) return result[0] if result else 'Error reading the register' def _get_string_from_registers(self, start_address, count): registers = self.read_registers(start_address, count) return self.extract_string_from_registers(registers) if registers else 'na' def write_coil(self, address): self.client.write_coil(address, True, slave=1) def read_registers(self, start_address, count): result = self.client.read_input_registers(start_address, count, slave=1) if not result.isError(): return result.registers print('Error reading registers') return None def extract_string_from_registers(self, registers): return ''.join(chr((reg >> 8) & 0xFF) + chr(reg & 0xFF) for reg in registers).strip('\x00')