import serial from serial import SerialException from enum import IntEnum import time class Command(IntEnum): REQ_ORDER_VERSION = 0x9C REQ_GET_DATA = 0x9D REQ_SET_CONFIG = 0x9E REQ_GET_CONFIG = 0x9F REQ_ENABLE_HEATER = 0xA0 SET_MANUAL_COOLER_ORDER = 0xA1 SET_AUTO_COOLER_ORDER = 0xA2 GET_INFOS_EF_LENS = 0xA3 SET_LENS_DIAPH = 0xA4 SET_LENS_FOCUS = 0xA5 REQ_GET_DATAEXT = 0xA6 GET_SNR_CAMERA = 0xA9 class Camera_type(IntEnum): CAMERA_SEEING_Mon = 0x06 CAMERA_MINI_CYCLOP = 0x10 CAMERA_OMEAEXT = 0x11 CAMERA_IS_WEXT_2 = 0x13 CAMERA_OMEA_2 = 0x14 CAMERA_OMEA_8 = 0x15 CAMERA_OMEA_8_ExtWea = 0x16 CAMERA_OMEA_5 = 0x17 CAMERA_OMEA_5_ExtWea = 0x18 CAMERA_EUDA2_ASI1600 = 0x19 CAMERA_OMEA_9 = 0x20 CAMERA_EUDA3_ASI16200 = 0x21 CAMERA_OMEA_9_ExtWea = 0x22 CAMERA_OMEA_4 = 0x23 CAMERA_OMEA_4_ExtWea = 0x24 class FNumLens_type(IntEnum): #EUDA 2M lens F2_8 = 32 F3_1 = 34 F3_6 = 36 F4_0 = 40 F4_4 = 42 F5_2 = 46 F5_6 = 48 F6_2 = 50 F7_3 = 54 F8_0 = 56 F9_5 = 60 F10_4 = 62 F11 = 64 F13 = 68 F16 = 72 F19 = 76 F21 = 78 F22 = 79 _5V_ = 5.0 ; Iters = 16.0; _10bits_ = 1024.0; Global_Offset_RH= 0.1515 ; Global_Factor_RH= 1.0/0.00636; _ON_ =1; _OFF_=0; class ComObjRs232: def __init__(self): self.ser = None self.com_opened = False self.port_com_n = None self.write_log = None # Assign a function like print for logging self.trial = 0 self.last_order = 0 # First byte of last sent command def open_com(self, num_com: int, total_timeout_ms_rs232_read: int = 100): """ Opens the specified COM port with settings matching the original Delphi code. :param num_com: COM port number (e.g., 3 for COM3, 15 for COM15) :param total_timeout_ms_rs232_read: Read total timeout constant in ms (default 100) """ if self.com_opened: return # Already open → exit like in the original self.port_com_n = num_com port_name = f"COM{num_com}" try: self.ser = serial.Serial( port=port_name, baudrate=115200, bytesize=8, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=total_timeout_ms_rs232_read / 1000.0, # ReadTotalTimeoutConstant in seconds write_timeout=0.5, # WriteTotalTimeoutConstant = 500 ms xonxoff=False, rtscts=False, dsrdtr=False ) # Additional fine-tuning to match Windows ReadIntervalTimeout=0 and Multipliers=0 # pyserial's timeout behavior already approximates this when using the above values self.com_opened = True if self.write_log: self.write_log(f"COM{num_com} opened successfully (handle: {self.ser})") except SerialException as e: error_msg = f"Cannot open COM{num_com}. Likely another application is using it! ({e})" if self.write_log: self.write_log(error_msg) raise SerialException(error_msg) from e except Exception as e: error_msg = f"Unexpected error opening COM{num_com}: {e}" if self.write_log: self.write_log(error_msg) raise def write_buffer(self, data: bytes): """Equivalent to Delphi WriteBuffer: sends data and checks full write""" if not self.com_opened or not self.ser: raise ExceptionRS232("COM port not open") # Purge receive buffer before sending (like PURGE_RXCLEAR) self.ser.reset_input_buffer() written = self.ser.write(data) self.ser.flush() # Ensure all data is sent if written != len(data): error_msg = f"WriteBuffer error: only {written} bytes written out of {len(data)} (last order: {self.last_order:#04x})" if self.write_log: self.write_log(error_msg) raise ExceptionRS232(error_msg) if self.write_log: self.write_log(f"WriteBuffer: sent {len(data)} bytes: {list(data)}") def read_buffer(self) -> bytes: """ Equivalent to Delphi ReadBuffer. Reads exactly: - 2 bytes: header (0x1F) + payload_length - then payload_length bytes (includes data + checksum) Returns the full received frame (header + length + payload + checksum) """ if not self.com_opened or not self.ser: raise ExceptionRS232("COM port not open") try: # Step 1: Read exactly 2 bytes (header + length byte) header = self.ser.read(2) if len(header) != 2: error_msg = f"ReadBuffer error: expected 2 header bytes, got {len(header)} (last order: {self.last_order:#04x})" if self.write_log: self.write_log(error_msg) raise ExceptionRS232(error_msg) if header[0] != 0x1F: error_msg = f"Bad response header: expected 0x1F, got {header[0]:#04x} (last order: {self.last_order:#04x})" if self.write_log: self.write_log(error_msg) raise ExceptionRS232(error_msg) payload_length = header[1] # This is the number of bytes after the length byte # Step 2: Read exactly payload_length bytes payload = self.ser.read(payload_length) if len(payload) != payload_length: error_msg = f"ReadBuffer data error: expected {payload_length} bytes, got {len(payload)} (last order: {self.last_order:#04x})" if self.write_log: self.write_log(error_msg) raise ExceptionRS232(error_msg) full_response = header + payload if self.write_log: self.write_log(f"ReadBuffer: received {len(full_response)} bytes: {list(full_response)}") return full_response except serial.SerialException as e: error_msg = f"COM Exception in ReadBuffer: {e}" if self.write_log: self.write_log(error_msg) raise ExceptionRS232(error_msg) from e def compute_checksum(self, data: bytes) -> int: """Sum of all bytes except the last one (checksum itself) → 8-bit""" if len(data) <= 1: return 0 return sum(data[:-1]) & 0xFF def send_order_receive_feedback(self, in_order: bytes) -> bytes: """ Exact equivalent of Delphi SendOrder_ReceiveFeedBack Returns: extracted payload (without header, length byte, or checksum) """ DEBUT_TRAME = 0x1D RET_TRAME = 0x1F self.trial += 1 try: # Build frame: [0x1D] [payload_len] [data...] [checksum] payload_len = len(in_order) + 1 # +1 for the command identifier byte frame = bytearray([DEBUT_TRAME, payload_len]) frame.extend(in_order) self.last_order = in_order[0] if in_order else 0 # Compute checksum over all except the future checksum byte checksum = self.compute_checksum(frame + b'\x00') frame.append(checksum) # Send self.write_buffer(bytes(frame)) # Receive full response frame response = self.read_buffer() # Verify checksum if self.compute_checksum(response) != response[-1]: error_msg = f"Checksum error on response for order {self.last_order:#04x}" if self.write_log: self.write_log(error_msg) raise ExceptionRS232(error_msg) # Extract payload: response[2:-1] → data after length byte, before checksum extracted = response[2:-1] # Apply the length-based clipping (as in original: NbTotal = OutFeedBack[1]-1) reported_data_len = response[1] - 1 if len(extracted) < reported_data_len: raise ExceptionRS232(f"Response shorter than declared length (last order: {self.last_order:#04x})") extracted = extracted[:reported_data_len] self.trial = 0 # Success return bytes(extracted) except Exception as e: if self.trial < 4: if self.write_log: self.write_log(f"Retry {self.trial}/4 after error: {e}") return self.send_order_receive_feedback(in_order) else: final_msg = f"Communication failed after 4 attempts (last order: {self.last_order:#04x}): {str(e)}" if self.write_log: self.write_log(final_msg) raise ExceptionRS232(final_msg) from e @staticmethod def ADU_ToTemp(value: int) -> float: result = (value / (Iters * _10bits_)) * _5V_ # Volts result = (result - 0.5) * 100.0 return result @staticmethod def ADU_ToRH(Value: int) -> float: result=(Value/(Iters*_10bits_)); result=(result-Global_Offset_RH)*Global_Factor_RH; return result ##################################################################################################################################################### ##################################################################################################################################################### ##################################################################################################################################################### # Example usage: if __name__ == "__main__": com = ComObjRs232() com.write_log = print # Enable logging to console try: com.open_com(4) # Try to open COMx print("COM Port is now opened") # ... use the port ... cmd = bytes([Command.REQ_ORDER_VERSION]) # Replace with your actual command cmd = bytes([Command.REQ_GET_DATA,0,0,0]) out_feedback = com.send_order_receive_feedback(cmd) print("Response payload:", list(out_feedback)) # Assuming you just received the response payload: # out_feedback = com.send_order_receive_feedback(in_order) # bytes object if len(out_feedback)==4: VersionStr = f"{out_feedback[0]}.{out_feedback[1]}" MajorVersion=out_feedback[0] MinorVersion=out_feedback[1] print("Firmware version:", VersionStr) Type_cam=out_feedback[2] match Type_cam: case Camera_type.CAMERA_EUDA2_ASI1600: print("Camera type: EUDA 2M") case Camera_type.CAMERA_OMEA_4: print("Camera type: OMEA 4") case Camera_type.CAMERA_OMEA_9: print("Camera type: OMEA 9") case Camera_type.CAMERA_SEEING_Mon: print("Camera type: SEEING MON") case Camera_type.CAMERA_MINI_CYCLOP: print("Camera type: CYCLOPE") case Camera_type.CAMERA_OMEAEXT: print("Camera type: OMEA 3/6") case Camera_type.CAMERA_IS_WEXT_2: print("Camera type: OMEA 3/6 + Weather module") case Camera_type.CAMERA_OMEA_2: print("Camera type: OMEA 2") case Camera_type.CAMERA_OMEA_8: print("Camera type: OMEA 8") case Camera_type.CAMERA_OMEA_8_ExtWea: print("Camera type: OMEA 8 + Weather module") case Camera_type.CAMERA_OMEA_5: print("Camera type: OMEA 5") case Camera_type.CAMERA_OMEA_5_ExtWea: print("Camera type: OMEA 5 + Weather module") case Camera_type.CAMERA_EUDA2_ASI1600: print("Camera type: EUDA 2M") case Camera_type.CAMERA_OMEA_9: print("Camera type: OMEA 9") case Camera_type.CAMERA_EUDA3_ASI16200: print("Camera type: EUDA 3M") case Camera_type.CAMERA_OMEA_9_ExtWea: print("Camera type: OMEA 9 + Weather module") case Camera_type.CAMERA_OMEA_4_ExtWea: print("Camera type: OMEA 4 + Weather module") case _: print(f"Unknown camera type: {out_feedback[2]} 0x{out_feedback[2]:02X}") raise ExceptionRS232("Script stopped, fatal error") # Add more cases as needed else: print(f"Warning: Invalid response payload : {len(out_feedback)} /4") ######################## # Get Camera serial number cmd = bytes([Command.GET_SNR_CAMERA]) out_feedback = com.send_order_receive_feedback(cmd) snr_bytes = out_feedback[:8] # Limit to 8 bytes max snr_str = snr_bytes.decode('latin-1') # Convert to AnsiString equivalent snr = snr_str[2:8] # Copy from pos 3, length 6 print(f"Serial number is : '{snr}'\n") ######################## cmd = bytes([Command.REQ_GET_DATA]) out_feedback = com.send_order_receive_feedback(cmd) Value16b = (out_feedback[0] << 8) | out_feedback[1] print(f"External temperature ADU : {Value16b}\n") ExTemp=com.ADU_ToTemp(Value16b) print(f"External temperature °C : {ExTemp:1.1f}\n") Value16b = (out_feedback[2] << 8) | out_feedback[3] ExRHt=com.ADU_ToRH(Value16b) print(f"External RH (%) : {ExRHt:1.1f}\n") Value16b = (out_feedback[4] << 8) | out_feedback[5] IntTemp=com.ADU_ToTemp(Value16b) print(f"Internal temperature (°C) : {IntTemp:1.1f}\n") Value16b = (out_feedback[13] << 8) | out_feedback[14] IntRHt=com.ADU_ToRH(Value16b) print(f"Internal RH (%) : {IntRHt:1.1f}\n") ActiveHeater = out_feedback[6] > 0 AutoTempRH_Heat = out_feedback[7] > 0 PowerHeaterLevel = out_feedback[8] if ActiveHeater==1: print("Dome Heater is active !\n") else: print("Dome Heater is inactive !\n") print(f"Dome power heater level se to (%) : {PowerHeaterLevel:1.0f}\n") Value16b = (out_feedback[15] << 8) | out_feedback[16] DomeTemp=com.ADU_ToTemp(Value16b) print(f"Dome temperature °C : {DomeTemp:1.1f}\n") # Get info from lens - This wakes up lens UC cmd = bytes([Command.GET_INFOS_EF_LENS]) out_feedback = com.send_order_receive_feedback(cmd) PresenceLens=out_feedback[0]==0xAA if PresenceLens==1: print("Lens detected") lensID=(out_feedback[1] << 8) | out_feedback[2] print(f"Lens ID is 0x{lensID:02X}\n") time.sleep(0.25) # Do not hammer lens UC! in_order = bytearray() # Home focus to min position in_order.clear() in_order.append(Command.SET_LENS_FOCUS) # command ID in_order.append(0x13) # subcommand in_order.append(0x00) # param1 in_order.append(0x00) # param2 out_feedback = com.send_order_receive_feedback(bytes(in_order)) print("Lens focus homed\n") time.sleep(0.25) steps = 500 # example value=500 (this is relative steps from home position) in_order.clear() in_order.append(Command.SET_LENS_FOCUS) # command ID in_order.append(0x14) # subcommand in_order.extend(steps.to_bytes(2, 'big')) # append Steps as big-endian 2 bytes out_feedback = com.send_order_receive_feedback(bytes(in_order)) print(f"Lens focus was set to {steps}\n") time.sleep(0.5) # Home Iris to min position in_order.clear() in_order.append(Command.SET_LENS_DIAPH) in_order.append(0xFF) # subcommand in_order.append(0x00) # param1 in_order.append(0x00) # param2 out_feedback = com.send_order_receive_feedback(bytes(in_order)) print("Lens fully opened") # This is the "home" position of the iris time.sleep(0.4) # EUDA 2M lens # Then apply offset to close Iris to a given value from full open value i.e f/2.8 base_zero=FNumLens_type.F2_8 valueDiaph=FNumLens_type.F8_0-base_zero in_order.clear() in_order.append(Command.SET_LENS_DIAPH) in_order.append(0x77) # subcommand in_order.append(valueDiaph) # param1 out_feedback = com.send_order_receive_feedback(bytes(in_order)) print(f"Lens diaph value applied -> {valueDiaph}") print("Lens closed to F/8\n") input("Press Enter to continue...\n") # Home back Iris to min position cmd = bytes([Command.SET_LENS_DIAPH,0xFF,0x00,0x00]) out_feedback = com.send_order_receive_feedback(cmd) print("Script completed\n") else: print("No lens detected") ######################## com.ser.close() except ExceptionRS232 as ex: print("RS232 Error:", ex) except Exception as ex: print("Unexpected error:", ex) finally: if com.ser and com.ser.is_open: com.ser.close()