diff --git a/src/comet/emulator/cts/itc.py b/src/comet/emulator/cts/itc.py index 5a74479..ea2468e 100644 --- a/src/comet/emulator/cts/itc.py +++ b/src/comet/emulator/cts/itc.py @@ -31,92 +31,92 @@ def __init__(self) -> None: self.program: int = 0 - @message(r'^T$') + @message(r"T$") def get_t(self) -> str: return datetime.datetime.now().strftime("T%d%m%y%H%M%S") - @message(r'^(t\d{6}\d{6})$') + @message(r"(t\d{6}\d{6})$") def set_t(self, value) -> str: t = datetime.datetime.strptime(value, "t%d%m%y%H%M%S") return t.strftime("t%d%m%y%H%M%S") - @message(r'^(A0)$') + @message(r"(A0)$") def get_a0(self, channel) -> str: self.current_temp += random.uniform(-.25, +.25) self.current_temp = min(60., max(20., self.current_temp)) return f"{channel} {self.current_temp:05.1f} {self.target_temp:05.1f}" - @message(r'^(A[34])$') + @message(r"(A[34])$") def get_a3(self, channel) -> str: return fake_analog_channel(channel, -45., +185.) - @message(r'^(A1)$') + @message(r"(A1)$") def get_a1(self, channel) -> str: self.current_humid += random.uniform(-.25, +.25) self.current_humid = min(95., max(15., self.current_humid)) return f"{channel} {self.current_humid:05.1f} {self.target_humid:05.1f}" - @message(r'^(A2)$') + @message(r"(A2)$") def get_a2(self, channel) -> str: return fake_analog_channel(channel, +0., +15.) - @message(r'^(A[56])$') + @message(r"(A[56])$") def get_a5(self, channel) -> str: return fake_analog_channel(channel, +5., +98.) - @message(r'^(A7)$') + @message(r"(A7)$") def get_a7(self, channel) -> str: return fake_analog_channel(channel, -50., +150.) - @message(r'^(A8)$') + @message(r"(A8)$") def get_a8(self, channel) -> str: return fake_analog_channel(channel, -80., +190.) - @message(r'^(A9)$') + @message(r"(A9)$") def get_a9(self, channel) -> str: return fake_analog_channel(channel, -0., +25.) - @message(r'^(A\:)$') + @message(r"(A\:)$") def get_a10(self, channel) -> str: return fake_analog_channel(channel, -50., +100.) - @message(r'^(A\;)$') + @message(r"(A\;)$") def get_a11(self, channel) -> str: return fake_analog_channel(channel, -0., +25.) - @message(r'^(A\<)$') + @message(r"(A\<)$") def get_a12(self, channel) -> str: return fake_analog_channel(channel, +2., +5.) - @message(r'^(A[\=\>])$') + @message(r"(A[\=\>])$") def get_a13(self, channel) -> str: return fake_analog_channel(channel, -100., +200.) - @message(r'^(A\?)$') + @message(r"(A\?)$") def get_a14(self, channel) -> str: return fake_analog_channel(channel, -80., +200.) - @message(r'^a[1-7]\s(-?\d+.\d)$') + @message(r"a[1-7]\s(-?\d+.\d)$") def set_a15(self, value) -> str: return "a" - @message(r'^a[0-6]\s+\d+\.\d+$') + @message(r"a[0-6]\s+\d+\.\d+$") def set_a(self) -> str: return "a" - @message(r'^O$') + @message(r"O$") def get_o(self) -> str: return "O1000000000000" - @message(r'^S$') + @message(r"S$") def get_s(self) -> str: return "S11110100\x06" - @message(r'^P$') + @message(r"P$") def get_p(self) -> str: return f"P{self.program:03d}" - @message(r'^p(\d{3})$') + @message(r"p(\d{3})$") def set_p(self, program) -> str: self.program = int(program) return f"p{self.program:03d}" diff --git a/src/comet/emulator/ers/ac3.py b/src/comet/emulator/ers/ac3.py index 0eb4143..49bcdf7 100644 --- a/src/comet/emulator/ers/ac3.py +++ b/src/comet/emulator/ers/ac3.py @@ -85,47 +85,47 @@ def __init__(self) -> None: self.state = State() self.logic = Logic(self.state) - @message(r"^RC$") + @message(r"RC$") def get_temperature(self) -> str: self.logic.update_state() return f"C{int(self.state.temperature * 10):+05d}" - @message(r"^RT$") + @message(r"RT$") def get_target_temperature(self) -> str: return f"T{int(self.state.target_temperature * 10):+05d}" - @message(r"^ST([+-]\d{4})$") + @message(r"ST([+-]\d{4})$") def set_target_temperature(self, value: str) -> str: self.state.target_temperature = float(value) / 10 return "OK" - @message(r"^RO$") + @message(r"RO$") def get_mode(self) -> str: return f"O{self.state.mode:d}" - @message(r"^SO([1234])$") + @message(r"SO([1234])$") def set_mode(self, value: str) -> str: self.state.mode = Mode(int(value)) return "OK" - @message(r"^RF$") + @message(r"RF$") def get_dewpoint(self) -> str: return f"F{int(self.state.dewpoint * 10):+05d}" - @message(r"^RD$") + @message(r"RD$") def get_dewpoint_control_status(self) -> str: return f"D{int(self.state.dewpoint_control_status)}" - @message(r"^SD([01])$") + @message(r"SD([01])$") def set_dewpoint_control_status(self, value: str) -> str: self.state.dewpoint_control_status = bool(int(value)) return "OK" - @message(r"^RH$") + @message(r"RH$") def get_hold_mode(self) -> str: return f"H{self.state.hold_mode:02d}" - @message(r"^SH([01])$") + @message(r"SH([01])$") def set_hold_mode(self, value: str) -> str: val = int(value) if val == 0: @@ -134,12 +134,12 @@ def set_hold_mode(self, value: str) -> str: self.state.hold_mode = 11 return "OK" - @message(r"^RI$") + @message(r"RI$") def get_control_status(self) -> str: self.logic.update_state() return f"I{self.state.control_status:d}" - @message(r"^RE$") + @message(r"RE$") def get_error(self) -> str: return "E000" diff --git a/src/comet/emulator/hephy/brandbox.py b/src/comet/emulator/hephy/brandbox.py index 6820440..6bab797 100644 --- a/src/comet/emulator/hephy/brandbox.py +++ b/src/comet/emulator/hephy/brandbox.py @@ -39,34 +39,34 @@ def __init__(self) -> None: def opened_channels(self) -> set[str]: return set(self.CHANNELS) - self.closed_channels - @message(r'\*IDN\?') + @message(r"\*IDN\?$") def get_idn(self) -> str: return self.options.get("identity", self.IDENTITY) - @message(r'\*RST') + @message(r"\*RST$") def set_rst(self) -> str: return self.SUCCESS - @message(r'\*CLS') + @message(r"\*CLS$") def set_cls(self) -> str: return self.SUCCESS - @message(r'\*STB\?') + @message(r"\*STB\?$") def get_stb(self) -> str: states = [] for channel in self.CHANNELS: states.append("1" if channel in self.closed_channels else "0") return ",".join(states) - @message(r'\*STR\?') + @message(r"\*STR\?$") def get_str(self) -> str: return "0" - @message(r'\*OPC\?') + @message(r"\*OPC\?$") def get_opc(self) -> str: return "1" - @message(r':CLOS (.+)') + @message(r":CLOS (.+)$") def set_close(self, channels) -> str: for channel in split_channels(channels): if self.has_channel(channel): @@ -75,7 +75,7 @@ def set_close(self, channels) -> str: return self.COMMAND_ERROR return self.SUCCESS - @message(r':OPEN (.+)') + @message(r":OPEN (.+)$") def set_open(self, channels) -> str: for channel in split_channels(channels): if self.has_channel(channel): @@ -85,19 +85,19 @@ def set_open(self, channels) -> str: return self.COMMAND_ERROR return self.SUCCESS - @message(r':CLOS:STAT\?') + @message(r":CLOS:STAT\?$") def get_close_state(self) -> str: return join_channels(sorted(self.closed_channels)) - @message(r':OPEN:STAT\?') + @message(r":OPEN:STAT\?$") def get_open_state(self) -> str: return join_channels(sorted(self.opened_channels)) - @message(r':DEBUG?') + @message(r":DEBUG?$") def get_debug(self) -> str: return self.COMMAND_ERROR - @message(r'SET:A_(ON|OFF)') + @message(r"SET:A_(ON|OFF)$") def set_a(self, state) -> str: for channel in ("A1", "A2"): if state == "ON": @@ -107,7 +107,7 @@ def set_a(self, state) -> str: self.closed_channels.remove(channel) return self.SUCCESS - @message(r'SET:B_(ON|OFF)') + @message(r"SET:B_(ON|OFF)$") def set_b(self, state) -> str: for channel in ("B1", "B2"): if state == "ON": @@ -117,7 +117,7 @@ def set_b(self, state) -> str: self.closed_channels.remove(channel) return self.SUCCESS - @message(r'SET:C_(ON|OFF)') + @message(r"SET:C_(ON|OFF)$") def set_c(self, state) -> str: for channel in ("C1", "C2"): if state == "ON": @@ -127,7 +127,7 @@ def set_c(self, state) -> str: self.closed_channels.remove(channel) return self.SUCCESS - @message(r'SET:(A1|A2|B1|B2|C1|C2)_(ON|OFF)') + @message(r"SET:(A1|A2|B1|B2|C1|C2)_(ON|OFF)$") def set_abc(self, channel, state) -> str: if state == "ON": self.closed_channels.add(channel) @@ -136,52 +136,52 @@ def set_abc(self, channel, state) -> str: self.closed_channels.remove(channel) return self.SUCCESS - @message(r'SET:MOD (IV|CV)') + @message(r"SET:MOD (IV|CV)$") def set_mod(self, mod) -> str: if mod not in self.MODS: return self.COMMAND_ERROR self.mod = mod return self.SUCCESS - @message(r'GET:A \?') + @message(r"GET:A \?$") def get_a(self) -> str: states = [] for channel in ("A1", "A2"): states.append(format_state(channel in self.closed_channels)) return ",".join(states) - @message(r'GET:B \?') + @message(r"GET:B \?$") def get_b(self) -> str: states = [] for channel in ("B1", "B2"): states.append(format_state(channel in self.closed_channels)) return ",".join(states) - @message(r'GET:C \?') + @message(r"GET:C \?$") def get_c(self) -> str: states = [] for channel in ("C1", "C2"): states.append(format_state(channel in self.closed_channels)) return ",".join(states) - @message(r'GET:(A1|A2|B1|B2|C1|C2) \?') + @message(r"GET:(A1|A2|B1|B2|C1|C2) \?$") def get_abc(self, channel) -> str: return format_state(channel in self.closed_channels) - @message(r'GET:MOD \?') + @message(r"GET:MOD \?$") def get_mod(self) -> str: return format(self.mod) - @message(r'GET:TST \?') + @message(r"GET:TST \?$") def get_test(self) -> str: return {False: "OFF", True: "ON"}[self.test_state] - @message(r'SET:TST (ON|OFF)') + @message(r"SET:TST (ON|OFF)$") def set_test(self, value) -> str: self.test_state = value == "ON" return self.SUCCESS - @message(r'.*') + @message(r".*") def unknown_message(self) -> str: return self.COMMAND_ERROR diff --git a/src/comet/emulator/hephy/environbox.py b/src/comet/emulator/hephy/environbox.py index 9bebf77..1af1ed1 100644 --- a/src/comet/emulator/hephy/environbox.py +++ b/src/comet/emulator/hephy/environbox.py @@ -203,11 +203,11 @@ def create_pc_data(self) -> list[str]: pc_data[38] = format(self.pt100_2_enabled, "d") return pc_data - @message(r"\*IDN\?") + @message(r"\*IDN\?$") def get_idn(self) -> str: return self.options.get("identity", self.IDENTITY) - @message(r"SET:NEW_ADDR (\d+)") + @message(r"SET:NEW_ADDR (\d+)$") def set_new_addr(self, address) -> str: value = int(address) if value not in type(self).SENSOR_ADRESSES: @@ -215,21 +215,21 @@ def set_new_addr(self, address) -> str: self.sensor_address[0] = value return self.SUCCESS - @message(r"SET:TEST_LED (ON|OFF)") + @message(r"SET:TEST_LED (ON|OFF)$") def set_test_led(self, value) -> str: self.test_led = value == "ON" return self.SUCCESS - @message(r"GET:TEST_LED \?") + @message(r"GET:TEST_LED \?$") def get_test_led(self) -> str: return {True: "1", False: "0"}[self.test_led] - @message(r"SET:DISCHARGE (AUTO|ON|OFF)") + @message(r"SET:DISCHARGE (AUTO|ON|OFF)$") def set_discharge(self, state) -> str: self.discharge = state return self.SUCCESS - @message(r"SET:DISCHARGE_TIME (\d+)") + @message(r"SET:DISCHARGE_TIME (\d+)$") def set_discharge_time(self, seconds) -> str: seconds = int(seconds) if 0 <= seconds <= 9999: @@ -237,43 +237,43 @@ def set_discharge_time(self, seconds) -> str: return self.SUCCESS return format_error(44) - @message(r"GET:DISCHARGE_TIME \?") + @message(r"GET:DISCHARGE_TIME \?$") def get_discharge_time(self) -> str: return format(self.discharge_time) - @message(r"SET:PT100_1 (ON|OFF)") + @message(r"SET:PT100_1 (ON|OFF)$") def set_pt100_1(self, value) -> str: self.pt100_1_enabled = value == "ON" return self.SUCCESS - @message(r"GET:PT100_1 \?") + @message(r"GET:PT100_1 \?$") def get_pt100_1(self) -> str: return format(self.pt100_1, ".2F") - @message(r"SET:PT100_2 (ON|OFF)") + @message(r"SET:PT100_2 (ON|OFF)$") def set_pt100_2(self, value) -> str: self.pt100_2_enabled = value == "ON" return self.SUCCESS - @message(r"GET:PT100_2 \?") + @message(r"GET:PT100_2 \?$") def get_pt100_2(self) -> str: return format(self.pt100_2, ".2F") - @message(r"SET:CTRL (ON|OFF)") + @message(r"SET:CTRL (ON|OFF)$") def set_ctrl(self, value) -> str: self.pid_control = value == "ON" return self.SUCCESS - @message(r"GET:CTRL \?") + @message(r"GET:CTRL \?$") def get_ctrl(self) -> str: return format(self.pid_control, "d") - @message(r"SET:CTRL_MODE (HUM|DEW)") + @message(r"SET:CTRL_MODE (HUM|DEW)$") def set_ctrl_mode(self, mode) -> str: self.pid_control_mode = mode return self.SUCCESS - @message(r"SET:SETPOINT (.+)") + @message(r"SET:SETPOINT (.+)$") def set_setpoint(self, setpoint) -> str: try: self.pid_setpoint = float(setpoint) @@ -281,11 +281,11 @@ def set_setpoint(self, setpoint) -> str: return format_error(30) return self.SUCCESS - @message(r"GET:SETPOINT \?") + @message(r"GET:SETPOINT \?$") def get_setpoint(self) -> str: return format(self.pid_setpoint, ".2F") - @message(r"SET:PID_KP (.+)") + @message(r"SET:PID_KP (.+)$") def set_pid_kp(self, value) -> str: try: self.pid_kp = float(value) @@ -293,11 +293,11 @@ def set_pid_kp(self, value) -> str: return format_error(31) return self.SUCCESS - @message(r"GET:PID_KP \?") + @message(r"GET:PID_KP \?$") def get_pid_kp(self) -> str: return format(self.pid_kp, ".2F") - @message(r"SET:PID_KI (.+)") + @message(r"SET:PID_KI (.+)$") def set_pid_ki(self, value) -> str: try: self.pid_ki = float(value) @@ -305,11 +305,11 @@ def set_pid_ki(self, value) -> str: return format_error(32) return self.SUCCESS - @message(r"GET:PID_KI \?") + @message(r"GET:PID_KI \?$") def get_pid_ki(self) -> str: return format(self.pid_ki, ".2F") - @message(r"SET:PID_KD (.+)") + @message(r"SET:PID_KD (.+)$") def set_pid_kd(self, value) -> str: try: self.pid_kd = float(value) @@ -317,11 +317,11 @@ def set_pid_kd(self, value) -> str: return format_error(33) return self.SUCCESS - @message(r"GET:PID_KD \?") + @message(r"GET:PID_KD \?$") def get_pid_kd(self) -> str: return format(self.pid_kd, ".2F") - @message(r"SET:PID_KP2 (.+)") + @message(r"SET:PID_KP2 (.+)$") def set_pid_kp2(self, value) -> str: try: self.pid_kp2 = float(value) @@ -329,11 +329,11 @@ def set_pid_kp2(self, value) -> str: return format_error(34) return self.SUCCESS - @message(r"GET:PID_KP2 \?") + @message(r"GET:PID_KP2 \?$") def get_pid_kp2(self) -> str: return format(self.pid_kp2, ".2F") - @message(r"SET:PID_KI2 (.+)") + @message(r"SET:PID_KI2 (.+)$") def set_pid_ki2(self, value) -> str: try: self.pid_ki2 = float(value) @@ -341,11 +341,11 @@ def set_pid_ki2(self, value) -> str: return format_error(35) return self.SUCCESS - @message(r"GET:PID_KI2 \?") + @message(r"GET:PID_KI2 \?$") def get_pid_ki2(self) -> str: return format(self.pid_ki2, ".2F") - @message(r"SET:PID_KD2 (.+)") + @message(r"SET:PID_KD2 (.+)$") def set_pid_kd2(self, value) -> str: try: self.pid_kd2 = float(value) @@ -353,65 +353,65 @@ def set_pid_kd2(self, value) -> str: return format_error(36) return self.SUCCESS - @message(r"GET:PID_KD2 (.+)") + @message(r"GET:PID_KD2 (.+)$") def get_pid_kd2(self) -> str: return format(self.pid_kd2, ".2F") - @message(r"SET:PID_MIN (\d+)") + @message(r"SET:PID_MIN (\d+)$") def set_pid_min(self, value) -> str: self.pid_min = int(value) return self.SUCCESS - @message(r"GET:PID_MIN \?") + @message(r"GET:PID_MIN \?$") def get_pid_min(self) -> str: return format(self.pid_min, "d") - @message(r"SET:PID_MAX (\d+)") + @message(r"SET:PID_MAX (\d+)$") def set_pid_max(self, value) -> str: self.pid_max = int(value) return self.SUCCESS - @message(r"GET:PID_MAX \?") + @message(r"GET:PID_MAX \?$") def get_pid_max(self) -> str: return format(self.pid_max, "d") - @message(r"SET:PID_SAMPLE_TIME (\d+)") + @message(r"SET:PID_SAMPLE_TIME (\d+)$") def set_pid_sample_time(self, value) -> str: self.pid_sample_time = int(value) return self.SUCCESS - @message(r"GET:PID_SAMPLE_TIME \?") + @message(r"GET:PID_SAMPLE_TIME \?$") def get_pid_sample_time(self) -> str: return format(self.pid_sample_time, "d") - @message(r"SET:PID_PROP_MODE (M|E)") + @message(r"SET:PID_PROP_MODE (M|E)$") def set_pid_prop_mode(self, mode) -> str: self.pid_prop_mode = mode return self.SUCCESS - @message(r"GET:PID_PROP_MODE \?") + @message(r"GET:PID_PROP_MODE \?$") def get_pid_prop_mode(self) -> str: return format(self.pid_prop_mode_index, "d") - @message(r"SET:PID_THRESHOLD (\d+)") + @message(r"SET:PID_THRESHOLD (\d+)$") def set_pid_threshold(self, value) -> str: self.pid_threshold = int(value) return self.SUCCESS - @message(r"GET:PID_THRESHOLD \?") + @message(r"GET:PID_THRESHOLD \?$") def get_pid_threshold(self) -> str: return format(self.pid_threshold, "d") - @message(r"SET:PARAMETER_SET (1|2)") + @message(r"SET:PARAMETER_SET (1|2)$") def set_parameter_set(self, value) -> str: self.parameter_set = int(value) return self.SUCCESS - @message(r"GET:PARAMETER_SET \?") + @message(r"GET:PARAMETER_SET \?$") def get_parameter_set(self) -> str: return format(self.parameter_set, "d") - @message(r"SET:PARA_THRESHOLD (.+)") + @message(r"SET:PARA_THRESHOLD (.+)$") def set_parameter_threshold(self, value) -> str: try: self.parameter_threshold = float(value) @@ -419,164 +419,164 @@ def set_parameter_threshold(self, value) -> str: return format_error(38) return self.SUCCESS - @message(r"GET:PARA_THRESHOLD \?") + @message(r"GET:PARA_THRESHOLD \?$") def get_parameter_threshold(self) -> str: return format(self.parameter_threshold, ".2F") - @message(r"SET:DAC (\d+)") + @message(r"SET:DAC (\d+)$") def set_dac(self, value) -> str: self.dac = int(value) return self.SUCCESS - @message(r"SET:MICROSCOPE_CTRL (ON|OFF)") + @message(r"SET:MICROSCOPE_CTRL (ON|OFF)$") def set_microscope_control(self, value) -> str: self.microscope_control = value == "ON" return self.SUCCESS - @message(r"GET:MICROSCOPE_CTRL \?") + @message(r"GET:MICROSCOPE_CTRL \?$") def get_microscope_control(self) -> str: return format(int(self.microscope_control)) - @message(r"SET:MICROSCOPE_LIGHT (ON|OFF)") + @message(r"SET:MICROSCOPE_LIGHT (ON|OFF)$") def set_microscope_light(self, value) -> str: self.microscope_light = value == "ON" return self.SUCCESS - @message(r"GET:MICROSCOPE_LIGHT \?") + @message(r"GET:MICROSCOPE_LIGHT \?$") def get_microscope_light(self) -> str: return format(int(self.microscope_light)) - @message(r"SET:MICROSCOPE_CAM (ON|OFF)") + @message(r"SET:MICROSCOPE_CAM (ON|OFF)$") def set_microscope_camera(self, value) -> str: self.microscope_camera = value == "ON" return self.SUCCESS - @message(r"GET:MICROSCOPE_CAM \?") + @message(r"GET:MICROSCOPE_CAM \?$") def get_microscope_camera(self) -> str: return format(int(self.microscope_camera)) - @message(r"SET:PROBCARD_LIGHT (ON|OFF)") + @message(r"SET:PROBCARD_LIGHT (ON|OFF)$") def set_probecard_light(self, value) -> str: self.probecard_light = value == "ON" return self.SUCCESS - @message(r"GET:PROBCARD_LIGHT \?") + @message(r"GET:PROBCARD_LIGHT \?$") def get_probecard_light(self) -> str: return format(int(self.probecard_light)) - @message(r"SET:PROBCARD_CAM (ON|OFF)") + @message(r"SET:PROBCARD_CAM (ON|OFF)$") def set_probecard_camera(self, value) -> str: self.probecard_camera = value == "ON" return self.SUCCESS - @message(r"GET:PROBCARD_CAM \?") + @message(r"GET:PROBCARD_CAM \?$") def get_probecard_camera(self) -> str: return format(self.probecard_camera, "d") - @message(r"SET:LASER_SENSOR (ON|OFF)") + @message(r"SET:LASER_SENSOR (ON|OFF)$") def set_laser_sensor(self, value) -> str: self.laser_sensor = value == "ON" return self.SUCCESS - @message(r"SET:BOX_LIGHT (ON|OFF)") + @message(r"SET:BOX_LIGHT (ON|OFF)$") def set_box_light(self, value) -> str: self.box_light = value == "ON" return self.SUCCESS - @message(r"GET:LIGHT \?") + @message(r"GET:LIGHT \?$") def get_box_light(self) -> str: return format(self.box_light, "d") - @message(r'SET:ENV_II_PCB (YES|NO)') + @message(r"SET:ENV_II_PCB (YES|NO)$") def set_env_ii_pcb(self, value) -> str: self.env_ii_pcb = value == "YES" return self.SUCCESS - @message(r"SET:PID_DOOR_STOP (ON|OFF)") + @message(r"SET:PID_DOOR_STOP (ON|OFF)$") def set_pid_door_stop(self, value) -> str: self.pid_door_stop = value == "ON" return self.SUCCESS - @message(r"GET:PID_DOOR_STOP \?") + @message(r"GET:PID_DOOR_STOP \?$") def get_pid_door_stop(self) -> str: return format(int(self.pid_door_stop) + 1, "d") # [1=OFF,2=ON] - @message(r"SET:DOOR_AUTO_LIGHT (ON|OFF)") + @message(r"SET:DOOR_AUTO_LIGHT (ON|OFF)$") def set_door_auto_light(self, value) -> str: self.door_auto_light = value == "ON" return self.SUCCESS - @message(r"GET:DOOR_AUTO_LIGHT \?") + @message(r"GET:DOOR_AUTO_LIGHT \?$") def get_door_auto_light(self) -> str: return format(int(self.door_auto_light) + 1, "d") # [1=OFF,2=ON] - @message(r'SET:DEBUG_MODE (ON|OFF)\?') + @message(r"SET:DEBUG_MODE (ON|OFF)\?$") def set_debug_mode(self, value) -> None: self.debug_mode = value == "ON" - @message(r'GET:DEBUG_MODE \?') + @message(r"GET:DEBUG_MODE \?$") def get_debug_mode(self) -> str: return format(int(self.debug_mode) + 1, "d") # [1=Off/2=On] - @message(r'GET:FREE_MEMORY \?') + @message(r"GET:FREE_MEMORY \?$") def get_free_memory(self) -> str: return format(self.free_memory, "d") - @message(r'GET:UPTIME \?') + @message(r"GET:UPTIME \?$") def get_uptime(self) -> str: delta_seconds = time.time() - self.boot_timestamp days, hours, minutes, seconds = split_seconds(delta_seconds) return f"{days:02d},{hours:02d},{minutes:02d},{seconds:02d}" - @message(r'GET:CHIP_ADDR (1|2|3)') + @message(r"GET:CHIP_ADDR (1|2|3)$") def get_chip_addr(self, value) -> str: return format(self.sensor_address[int(value) - 1], "d") - @message(r'GET:CHIP_NBR \?') + @message(r"GET:CHIP_NBR \?$") def get_chip_nbr(self) -> str: return format(self.sensor_count, "d") - @message(r"GET:TEMP \?") + @message(r"GET:TEMP \?$") def get_temp(self) -> str: return format(self.box_temperature, ".2F") - @message(r"GET:HUM \?") + @message(r"GET:HUM \?$") def get_hum(self) -> str: return format(self.box_humidity, ".2F") - @message(r"GET:LUX \?") + @message(r"GET:LUX \?$") def get_lux(self) -> str: return format(self.box_lux, ".1F") - @message(r"GET:FLOW_RATE \?") + @message(r"GET:FLOW_RATE \?$") def get_flow_rate(self) -> str: return format(self.flow_rate, ".1F") - @message(r"GET:VALVE_ON \?") + @message(r"GET:VALVE_ON \?$") def get_valve_on(self) -> str: return format(self.valve_on, "d") - @message(r"GET:DOOR \?") + @message(r"GET:DOOR \?$") def get_door(self) -> str: return format(int(self.door_open)) - @message(r"GET:LASER \?") + @message(r"GET:LASER \?$") def get_laser(self) -> str: return format(int(self.laser_enabled)) - @message(r"GET:LASER_POWER \?") + @message(r"GET:LASER_POWER \?$") def get_laser_power(self) -> str: return format(self.laser_power, "d") - @message(r"GET:PC_DATA \?") + @message(r"GET:PC_DATA \?$") def get_pc_data(self) -> str: return ",".join(map(format, self.create_pc_data())) - @message(r"GET:RELAY_STATUS \?") + @message(r"GET:RELAY_STATUS \?$") def get_relay_status(self) -> str: return format(self.power_relay_states()) - @message(r"GET:ENV \?") + @message(r"GET:ENV \?$") def get_env(self) -> str: values = [ format(self.box_temperature, ".1F"), @@ -586,7 +586,7 @@ def get_env(self) -> str: ] return ",".join(values) - @message(r"GET:VERSION \?") + @message(r"GET:VERSION \?$") def get_version(self) -> str: return self.options.get("version", self.VERSION) diff --git a/src/comet/emulator/hephy/shuntbox.py b/src/comet/emulator/hephy/shuntbox.py index 1d66bee..aca7acd 100644 --- a/src/comet/emulator/hephy/shuntbox.py +++ b/src/comet/emulator/hephy/shuntbox.py @@ -28,42 +28,42 @@ def __init__(self) -> None: def uptime(self) -> int: return int(round(time.time() - self.start_time)) - @message(r'\*IDN\?') + @message(r"\*IDN\?$") def get_idn(self) -> str: return self.options.get("identity", self.IDENTITY) - @message(r'GET:UP \?') + @message(r"GET:UP \?$") def get_up(self) -> str: return format(self.uptime) - @message(r'GET:RAM \?') + @message(r"GET:RAM \?$") def get_ram(self) -> str: return format(self.MEMORY_BYTES) - @message(r'GET:TEMP ALL') + @message(r"GET:TEMP ALL$") def get_temp_all(self) -> str: values = [] for i in range(self.CHANNELS): values.append(format(random.uniform(22.0, 26.0), ".1f")) return ",".join(values) - @message(r'GET:TEMP (\d+)') + @message(r"GET:TEMP (\d+)$") def get_temp(self, value) -> str: return format(random.uniform(22.0, 26.0), ".1f") - @message(r'SET:REL_(ON|OFF) (\d+|ALL)') + @message(r"SET:REL_(ON|OFF) (\d+|ALL)$") def set_rel(self, state, value) -> str: return self.SUCCESS - @message(r'GET:REL (\d+)') + @message(r"GET:REL (\d+)$") def get_rel(self, value) -> str: return "0" - @message(r'GET:REL ALL') + @message(r"GET:REL ALL$") def get_rel_all(self) -> str: return ",".join(["0"] * (self.CHANNELS + 4)) - @message(r'.*') + @message(r".*") def unknown_message(self) -> str: return format_error(99) diff --git a/src/comet/emulator/itk/corvustt.py b/src/comet/emulator/itk/corvustt.py index 1410ff8..e3da8fa 100644 --- a/src/comet/emulator/itk/corvustt.py +++ b/src/comet/emulator/itk/corvustt.py @@ -79,78 +79,78 @@ def load_options(self, options: Mapping[str, Any]) -> None: if isinstance(z := unit.get("z"), int): self.z_unit = int(max(1, min(3, z))) - @message(r'^identify$') + @message(r"identify$") def get_identify(self) -> str: return self.identity - @message(r'^version$') + @message(r"version$") def get_version(self) -> str: return self.version - @message(r'^getmacadr$') + @message(r"getmacadr$") def get_macadr(self) -> str: return self.mac_address - @message(r'^getserialno$') + @message(r"getserialno$") def get_serialno(self) -> str: return self.serial_no - @message(r'^getoptions$') + @message(r"getoptions$") def get_options(self) -> int: return 0x3 - @message(r'^getticks|gt$') + @message(r"getticks|gt$") def get_ticks(self) -> str: dt = time.monotonic() - self.ticks_t0 ticks = int(dt * (1 / 250e-6)) # 250us ticks return f"{ticks:d}" - @message(r'^(\d)\s+beep$') + @message(r"(\d)\s+beep$") def set_beep(self, millisec) -> None: ... - @message(r'^reset$') + @message(r"reset$") def set_reset(self) -> None: self.getcaldone = [0, 0, 0] self.status = 0 - @message(r'^status|st$') + @message(r"status|st$") def get_status(self) -> str: return f"{self.status:d}" - @message(r'^pos|p$') + @message(r"pos|p$") def get_pos(self) -> str: return f'{self.x_pos:.6f} {self.y_pos:.6f} {self.z_pos:.6f}' - @message(r'^(.+)\s+setlimit$') + @message(r"(.+)\s+setlimit$") def set_limit(self, value) -> None: a1, b1, c1, a2, b2, c2 = map(float, value.split()) self.table_limits = TableLimits(a1, b1, c1, a2, b2, c2) - @message(r'^getlimit$') + @message(r"getlimit$") def get_limit(self) -> tuple[str, str, str]: a1, b1, c1, a2, b2, c2 = astuple(self.table_limits) return f"{a1:.6f} {b1:.6f}", f"{c1:.6f} {a2:.6f}", f"{b2:.6f} {c2:.6f}" - @message(r'^([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+(?:move|m)$') + @message(r"([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+(?:move|m)$") def set_move(self, x, y, z) -> None: self.x_pos = max(0.0, float(x)) self.y_pos = max(0.0, float(y)) self.z_pos = max(0.0, float(z)) - @message(r'^([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+(?:rmove|r)$') + @message(r"([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+(?:rmove|r)$") def set_rmove(self, x, y, z) -> None: self.x_pos = max(0.0, self.x_pos + float(x)) self.y_pos = max(0.0, self.y_pos + float(y)) self.z_pos = max(0.0, self.z_pos + float(z)) - @message(r'^randmove$') + @message(r"randmove$") def set_randmove(self) -> None: a1, b1, c1, a2, b2, c2 = astuple(self.table_limits) self.x_pos = random.uniform(a1, a2) self.y_pos = random.uniform(b1, b2) self.z_pos = random.uniform(c1, c2) - @message(r'^(1|2|3)\s+getcaldone$') + @message(r"(1|2|3)\s+getcaldone$") def get_caldone(self, axis) -> Optional[str]: if axis == "1": return f"{self.getcaldone[0]}" @@ -160,7 +160,7 @@ def get_caldone(self, axis) -> Optional[str]: return f"{self.getcaldone[2]}" return None - @message(r'^(-1|1|2|3)\s+getaxis$') + @message(r"(-1|1|2|3)\s+getaxis$") def get_axis(self, axis) -> Optional[str]: if axis == "-1": a1, a2, a3 = self.getaxis @@ -173,23 +173,23 @@ def get_axis(self, axis) -> Optional[str]: return f"{self.getaxis[2]}" return None - @message(r'^geterror|ge$') + @message(r"geterror|ge$") def get_error(self) -> int: return self.geterror - @message(r'^getmerror|gme$') + @message(r"getmerror|gme$") def get_merror(self) -> int: return self.getmerror - @message(r'^(0|1)\s+(?:joystick|j)$') + @message(r"(0|1)\s+(?:joystick|j)$") def set_joystick(self, value) -> None: self.joystick = bool(int(value)) - @message(r'^getjoystick|gj$') + @message(r"getjoystick|gj$") def get_joystick(self) -> str: return f"{self.joystick:d}" - @message(r'^(-1|1|2|3)\s+getunit$') + @message(r"(-1|1|2|3)\s+getunit$") def get_unit(self, axis) -> Optional[str]: if axis == "-1": return f"{self.x_unit} {self.y_unit} {self.z_unit} 1" @@ -201,7 +201,7 @@ def get_unit(self, axis) -> Optional[str]: return f"{self.z_unit}" return None - @message(r'^(\d)\s+(1|2|3)\s+setunit$') + @message(r"(\d)\s+(1|2|3)\s+setunit$") def set_unit(self, value, axis) -> None: if axis == "1": self.x_unit = int(value) @@ -210,7 +210,7 @@ def set_unit(self, value, axis) -> None: if axis == "3": self.z_unit = int(value) - @message(r'^(1|2|3)\s+ncal$') + @message(r"(1|2|3)\s+ncal$") def set_ncal(self, axis) -> None: if axis == "1": self.getcaldone[0] = 0x1 @@ -219,7 +219,7 @@ def set_ncal(self, axis) -> None: if axis == "3": self.getcaldone[2] = 0x1 - @message(r'^(1|2|3)\s+nrm$') + @message(r"(1|2|3)\s+nrm$") def set_nrm(self, axis) -> None: if axis == "1": self.getcaldone[0] |= 0x2 diff --git a/src/comet/emulator/itk/hydra.py b/src/comet/emulator/itk/hydra.py index 86a07c3..2fc02f5 100644 --- a/src/comet/emulator/itk/hydra.py +++ b/src/comet/emulator/itk/hydra.py @@ -49,34 +49,34 @@ def load_options(self, options: Mapping[str, Any]) -> None: if isinstance(cpu_temperature := options.get("cpu_temperature"), float): self.cpu_temperature = cpu_temperature - @message(r'^identify$') + @message(r"identify$") def get_identify(self) -> str: return self.identity - @message(r'^getversion|version$') + @message(r"getversion|version$") def get_version(self) -> float: return self.version # double! - @message(r'^getmacadr$') + @message(r"getmacadr$") def get_macadr(self) -> str: return self.mac_address - @message(r'^getserialno$') + @message(r"getserialno$") def get_serialno(self) -> str: return self.serial_no - @message(r'^getproductid$') + @message(r"getproductid$") def get_productid(self) -> str: return "hydra" - @message(r'^getcputemp$') + @message(r"getcputemp$") def get_cputemp(self) -> float: return self.cpu_temperature - @message(r'^reset$') + @message(r"reset$") def set_reset(self) -> None: ... - @message(r'^status|st$') + @message(r"status|st$") def get_status(self) -> int: status = 0 all_cal = int(all([value & 0x1 for value in self.calibrate.values()])) @@ -87,7 +87,7 @@ def get_status(self) -> int: status |= ((all_rm & 0x1) << 4) return status - @message(r'^(1|2)\s+(?:nstatus|nst|est|ast)$') + @message(r"(1|2)\s+(?:nstatus|nst|est|ast)$") def get_nstatus(self, axis) -> int: status = 0 cal = int(self.calibrate[axis] & 0x1 == 0x1) @@ -98,24 +98,24 @@ def get_nstatus(self, axis) -> int: status |= ((rm & 0x1) << 4) return status - @message(r'^(1|2)\s+np$') + @message(r"(1|2)\s+np$") def get_np(self, axis) -> float: if axis == "1": return self.x_pos else: return self.y_pos - @message(r'^([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+m$') + @message(r"([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+m$") def set_move(self, x, y) -> None: self.x_pos = float(x) self.y_pos = float(y) - @message(r'^([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+r$') + @message(r"([+-]?\d+(?:\.\d+)?)\s+([+-]?\d+(?:\.\d+)?)\s+r$") def set_rmove(self, x, y) -> None: self.x_pos += float(x) self.y_pos += float(y) - @message(r'^(1|2)\s+nrandmove$') + @message(r"(1|2)\s+nrandmove$") def set_nrandmove(self, axis) -> None: pos = random.uniform(0, 100) if axis == "1": @@ -123,11 +123,11 @@ def set_nrandmove(self, axis) -> None: elif axis == "2": self.y_pos = pos - @message(r'^(1|2)\s+(?:ncalibrate|ncal)$') + @message(r"(1|2)\s+(?:ncalibrate|ncal)$") def set_ncalibrate(self, axis) -> None: self.calibrate[axis] = 0x1 - @message(r'^(1|2)\s+(?:nrangemeasure|nrm)$') + @message(r"(1|2)\s+(?:nrangemeasure|nrm)$") def set_nrangemeasure(self, axis) -> None: self.calibrate[axis] |= 0x2 diff --git a/src/comet/emulator/keithley/k2400.py b/src/comet/emulator/keithley/k2400.py index 9ad4aab..fdbcf52 100644 --- a/src/comet/emulator/keithley/k2400.py +++ b/src/comet/emulator/keithley/k2400.py @@ -34,7 +34,7 @@ def __init__(self) -> None: self.format_elements = FormatElements() self.format_elements.update(["VOLT", "CURR", "RES", "TIME", "STAT"]) - @message(r'^\*RST$') + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() self.system_beeper_state = True @@ -56,15 +56,15 @@ def set_rst(self) -> None: self.format_elements.clear() self.format_elements.update(["VOLT", "CURR", "RES", "TIME", "STAT"]) - @message(r'^\*CLS$') + @message(r"\*CLS$") def set_cls(self) -> None: self.error_queue.clear() - @message(r'^:?SYST(?:em)?:ERR(?:or)?:COUN(?:t)?\?$') + @message(r":?SYST(?:em)?:ERR(?:or)?:COUN(?:t)?\?$") def get_system_error_count(self) -> str: return format(len(self.error_queue), "d") - @message(r'^:?SYST(?:em)?:ERR(?:or)?(?::NEXT)?\?$') + @message(r":?SYST(?:em)?:ERR(?:or)?(?::NEXT)?\?$") def get_system_error_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -74,41 +74,41 @@ def get_system_error_next(self) -> str: # Beeper - @message(r'^:?SYST(?:em)?:BEEP(?:er)?(?::STAT(?:e)?)?\?$') + @message(r":?SYST(?:em)?:BEEP(?:er)?(?::STAT(?:e)?)?\?$") def get_system_beeper_state(self) -> str: return format(self.system_beeper_state, "d") - @message(r'^:?SYST(?:em)?:BEEP(?:er)?(?::STAT(?:e)?)? (OFF|ON|0|1)$') + @message(r":?SYST(?:em)?:BEEP(?:er)?(?::STAT(?:e)?)?\s+(OFF|ON|0|1)$") def set_system_beeper_state(self, state) -> None: self.system_beeper_state = {"OFF": False, "ON": True, "0": False, "1": True}[state] # Remote sensing - @message(r'^:?SYST(?:em)?:RSEN(?:se)?\?$') + @message(r":?SYST(?:em)?:RSEN(?:se)?\?$") def get_system_rsense(self) -> str: return format(self.system_rsense, "d") - @message(r'^:?SYST(?:em)?:RSEN(?:se)?\s+(OFF|ON|0|1)$') + @message(r":?SYST(?:em)?:RSEN(?:se)?\s+(OFF|ON|0|1)$") def set_system_rsense(self, enabled) -> None: self.system_rsense = {'OFF': False, 'ON': True, '0': False, '1': True}[enabled] # Route terminal - @message(r'^:?ROUT(?:e)?:TERM(?:inal)?\?$') + @message(r":?ROUT(?:e)?:TERM(?:inal)?\?$") def get_route_terminals(self) -> str: return self.route_terminals - @message(r'^:?ROUT(?:e)?:TERM(?:inal)? (FRON|REAR)$') + @message(r":?ROUT(?:e)?:TERM(?:inal)?\s+(FRON|REAR)$") def set_route_terminals(self, terminal) -> None: self.route_terminals = terminal # Output state - @message(r'^:?OUTP(?:ut)?(?::STAT(?:e)?)?\?$') + @message(r":?OUTP(?:ut)?(?::STAT(?:e)?)?\?$") def get_output_state(self) -> str: return {False: '0', True: '1'}[self.output_state] - @message(r'^:?OUTP(?:ut)?(?::STAT(?:e)?)? (.+)$') + @message(r":?OUTP(?:ut)?(?::STAT(?:e)?)?\s+(.+)$") def set_output_state(self, state) -> None: try: self.output_state = {"ON": True, "OFF": False, "0": False, "1": True}[state] @@ -117,11 +117,11 @@ def set_output_state(self, state) -> None: # Source function mode - @message(r'^:?SOUR:FUNC(?::MODE)?\?$') + @message(r":?SOUR:FUNC(?::MODE)?\?$") def get_source_function_mode(self) -> str: return self.source_function_mode - @message(r'^:?SOUR:FUNC(?::MODE)? (VOLT|CURR)$') + @message(r":?SOUR:FUNC(?::MODE)?\s+(VOLT|CURR)$") def set_source_function_mode(self, function) -> None: try: self.source_function_mode = function @@ -130,11 +130,11 @@ def set_source_function_mode(self, function) -> None: # Source levels - @message(r'^:?SOUR:(VOLT|CURR)(?::LEV)?\?$') + @message(r":?SOUR:(VOLT|CURR)(?::LEV)?\?$") def get_source_level(self, function) -> str: return format(self.source_level[function], 'E') - @message(r'^:?SOUR:(VOLT|CURR)(?::LEV)? (.+)$') + @message(r":?SOUR:(VOLT|CURR)(?::LEV)?\s+(.+)$") def set_source_level(self, function, level) -> None: try: self.source_level[function] = float(level) @@ -143,11 +143,11 @@ def set_source_level(self, function, level) -> None: # Source range levels - @message(r'^:?SOUR:(VOLT|CURR):RANG\?$') + @message(r":?SOUR:(VOLT|CURR):RANG\?$") def get_source_range_level(self, function) -> str: return format(self.source_range[function], "E") - @message(r'^:?SOUR:(VOLT|CURR):RANG (.+)$') + @message(r":?SOUR:(VOLT|CURR):RANG\s+(.+)$") def set_source_range_level(self, function, level) -> None: try: self.source_range[function] = float(level) @@ -157,11 +157,11 @@ def set_source_range_level(self, function, level) -> None: # Source auto ranges - @message(r'^:?SOUR:(VOLT|CURR):RANG:AUTO\?$') + @message(r":?SOUR:(VOLT|CURR):RANG:AUTO\?$") def get_source_range_auto(self, function) -> str: return format(self.source_range_auto[function], "d") - @message(r'^:?SOUR:(VOLT|CURR):RANG:AUTO (.+)$') + @message(r":?SOUR:(VOLT|CURR):RANG:AUTO\s+(.+)$") def set_source_range_auto(self, function, state) -> None: try: self.source_range_auto[function] = {"ON": True, "OFF": False, "0": False, "1": True}[state] @@ -170,11 +170,11 @@ def set_source_range_auto(self, function, state) -> None: # Source voltage limit - @message(r'^:?SOUR:VOLT:PROT(?::LEV)?\?$') + @message(r":?SOUR:VOLT:PROT(?::LEV)?\?$") def get_source_voltage_protection_level(self) -> str: return format(self.source_voltage_protection_level, "E") - @message(r'^:?SOUR:VOLT:PROT(?::LEV)? (.+)$') + @message(r":?SOUR:VOLT:PROT(?::LEV)?\s+(.+)$") def set_source_voltage_protection_level(self, level) -> None: try: self.source_voltage_protection_level = float(level) @@ -183,105 +183,105 @@ def set_source_voltage_protection_level(self, level) -> None: # Source compliance - @message(r'^(?::?SENS)?:VOLT:PROT(?::LEV)?\?$') + @message(r"(?::?SENS)?:VOLT:PROT(?::LEV)?\?$") def get_sense_voltage_protection_level(self) -> str: return format(self.sense_voltage_protection_level, "E") - @message(r'^(?::?SENS)?:VOLT:PROT(?::LEV)? (.+)$') + @message(r"(?::?SENS)?:VOLT:PROT(?::LEV)?\s+(.+)$") def set_sense_voltage_protection_level(self, level) -> None: try: self.sense_voltage_protection_level = float(level) except ValueError: self.error_queue.append(Error(101, "malformed command")) - @message(r'^(?::?SENS)?:VOLT:PROT:TRIP\?$') + @message(r"(?::?SENS)?:VOLT:PROT:TRIP\?$") def get_sense_voltage_protection_tripped(self) -> str: return format(False, "d") # TODO - @message(r'^(?::?SENS)?:CURR:PROT(?::LEV)?\?$') + @message(r"(?::?SENS)?:CURR:PROT(?::LEV)?\?$") def get_sense_current_protection_level(self) -> str: return format(self.sense_current_protection_level, "E") - @message(r'^(?::?SENS)?:CURR:PROT(?::LEV)? (.+)$') + @message(r"(?::?SENS)?:CURR:PROT(?::LEV)?\s+(.+)$") def set_sense_current_protection_level(self, level) -> None: try: self.sense_current_protection_level = float(level) except ValueError: self.error_queue.append(Error(101, "malformed command")) - @message(r'^(?::?SENS)?:CURR:PROT:TRIP\?$') + @message(r"(?::?SENS)?:CURR:PROT:TRIP\?$") def get_sense_current_protection_tripped(self) -> str: return format(False, "d") # TODO # Sense function - @message(r'^(?::?SENS)?:FUNC(?::ON)?\?$') + @message(r"(?::?SENS)?:FUNC(?::ON)?\?$") def get_sense_function_on(self) -> str: return format(self.sense_function) - @message(r'^(?::?SENS)?:FUNC(?::ON)? \'(VOLT|CURR|RES|TIME|STAT)\'$') + @message(r"(?::?SENS)?:FUNC(?::ON)?\s+\'(VOLT|CURR|RES|TIME|STAT)\'$") def set_sense_function_on(self, value) -> None: self.sense_function.add(value) # TODO - @message(r'^(?::?SENS)?:FUNC(?::ON)? \'VOLT\',\s*\'CURR\'$') + @message(r"(?::?SENS)?:FUNC(?::ON)?\s+\'VOLT\',\s*\'CURR\'$") def set_sense_function_on_2(self) -> None: self.sense_function.add("VOLT") self.sense_function.add("CURR") # Concurrent function - @message(r'^(?::?SENS)?:FUNC:CONC\?$') + @message(r"(?::?SENS)?:FUNC:CONC\?$") def get_sense_function_concurrent(self) -> int: return int(self.sense_function_concurrent) - @message(r'^(?::?SENS)?:FUNC:CONC (OFF|ON|0|1)$') + @message(r"(?::?SENS)?:FUNC:CONC\s+(OFF|ON|0|1)$") def set_sense_function_concurrent(self, state) -> None: self.sense_function_concurrent = {"OFF": False, "ON": True, "0": False, "1": True}[state] # Average - @message(r'^(?::?SENS)?:AVER:TCON\?$') + @message(r"(?::?SENS)?:AVER:TCON\?$") def get_sense_average_tcontrol(self) -> str: return self.sense_average_tcontrol - @message(r'^(?::?SENS)?:AVER:TCON (REP|MOV)$') + @message(r"(?::?SENS)?:AVER:TCON\s+(REP|MOV)$") def set_sense_average_tcontrol(self, state: str) -> None: self.sense_average_tcontrol = state - @message(r'^(?::?SENS)?:AVER:COUN\?$') + @message(r"(?::?SENS)?:AVER:COUN\?$") def get_sense_average_count(self) -> int: return self.sense_average_count - @message(r'^(?::?SENS)?:AVER:COUN (\d+)$') + @message(r"(?::?SENS)?:AVER:COUN\s+(\d+)$") def set_sense_average_count(self, count: str) -> None: self.sense_average_count = int(count) - @message(r'^(?::?SENS)?:AVER(?::STAT)?\?$') + @message(r"(?::?SENS)?:AVER(?::STAT)?\?$") def get_sense_average_state(self) -> int: return int(self.sense_average_state) - @message(r'^(?::?SENS)?:AVER(?::STAT)? (OFF|ON|0|1)$') + @message(r"(?::?SENS)?:AVER(?::STAT)?\s+(OFF|ON|0|1)$") def set_sense_average_state(self, state) -> None: self.sense_average_state = {"OFF": False, "ON": True, "0": False, "1": True}[state] # Integration time - @message(r'^(?::?SENS)?:(?:VOLT|CURR|RES):NPLC\?$') + @message(r"(?::?SENS)?:(?:VOLT|CURR|RES):NPLC\?$") def get_sense_nplc(self) -> str: return format(self.sense_nplc, "E") - @message(r'^(?::?SENS)?:(?:VOLT|CURR|RES):NPLC (.+)$') + @message(r"(?::?SENS)?:(?:VOLT|CURR|RES):NPLC\s+(.+)$") def set_sense_nplc(self, nplc: str) -> None: self.sense_nplc = round(float(nplc), 2) # Format - @message(r'^:?FORM:ELEM\?$') + @message(r":?FORM:ELEM\?$") def get_format_elements(self) -> str: return format(self.format_elements) - @message(r'^:?FORM:ELEM (.+)$') + @message(r":?FORM:ELEM\s+(.+)$") def set_format_elements(self, elements) -> None: elements = [element.strip() for element in elements.split(",") if element.strip()] self.format_elements.clear() @@ -289,10 +289,10 @@ def set_format_elements(self, elements) -> None: # Measure - @message(r'^:?INIT(?:iate)?$') + @message(r":?INIT(?:iate)?$") def set_initiate(self) -> None: ... - @message(r'^:?READ\?$') + @message(r":?READ\?$") def get_read(self) -> str: result = [] if "VOLT" in self.format_elements._values: @@ -311,14 +311,14 @@ def get_read(self) -> str: result.append(format(0)) return ",".join(result) - @message(r'^:?FETC[H]?\?$') + @message(r":?FETC[H]?\?$") def get_fetch(self) -> str: curr_min = float(self.options.get("curr.min", 1e-6)) curr_max = float(self.options.get("curr.max", 1e-7)) return format(random.uniform(curr_min, curr_max), "E") - @message(r'^(.*)$') - def unknown_message(self, request) -> None: + @message(r".*") + def unknown_message(self) -> None: self.error_queue.append(Error(101, "malformed command")) diff --git a/src/comet/emulator/keithley/k2470.py b/src/comet/emulator/keithley/k2470.py index 6ca4a89..271aeb3 100644 --- a/src/comet/emulator/keithley/k2470.py +++ b/src/comet/emulator/keithley/k2470.py @@ -35,11 +35,11 @@ def __init__(self) -> None: def output_interlock_tripped(self) -> bool: return bool(self.options.get("interlock.tripped", True)) - @message(r'^\*LANG\?$') + @message(r"\*LANG\?$") def get_lang(self) -> str: return self.language - @message(r'^\*RST$') + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() self.route_terminals = "FRON" @@ -58,11 +58,11 @@ def set_rst(self) -> None: self.sense_nplc = 1.0 self.system_breakdown_protection = "OFF" - @message(r'^\*CLS$') + @message(r"\*CLS$") def set_cls(self) -> None: self.error_queue.clear() - @message(r'^:?SYST:ERR(?::NEXT)?\?$') + @message(r":?SYST:ERR(?::NEXT)?\?$") def get_system_error_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -70,48 +70,48 @@ def get_system_error_next(self) -> str: error = Error(0, "no error") return f'{error.code}, "{error.message}"' - @message(r'^:?SYST:BRE:PROT\?$') + @message(r":?SYST:BRE:PROT\?$") def get_system_breakdown_protection(self) -> str: return self.system_breakdown_protection - @message(r'^:?SYST:BRE:PROT (AUTO|OFF|ON)$') + @message(r":?SYST:BRE:PROT\s+(AUTO|OFF|ON)$") def set_system_breakdown_protection(self, state) -> None: self.system_breakdown_protection = state # Route terminal - @message(r'^:?ROUT:TERM\?$') + @message(r":?ROUT:TERM\?$") def get_route_terminals(self) -> str: return self.route_terminals - @message(r'^:?ROUT:TERM (FRON|REAR)$') + @message(r":?ROUT:TERM\s+(FRON|REAR)$") def set_route_terminals(self, terminal) -> None: self.route_terminals = terminal # Output state - @message(r'^:?OUTP(?::STAT)?\?$') + @message(r":?OUTP(?::STAT)?\?$") def get_output_state(self) -> str: return {False: "0", True: "1"}[self.output_state] - @message(r'^:?OUTP(?::STAT)? (.+)$') + @message(r":?OUTP(?::STAT)?\s+(.+)$") def set_output_state(self, state) -> None: try: self.output_state = {"ON": True, "OFF": False, "0": False, "1": True}[state] except KeyError: self.error_queue.append(Error(101, "malformed command")) - @message(r'^:?OUTP:INT:TRIP\?$') + @message(r":?OUTP:INT:TRIP\?$") def get_output_interlock_tripped(self) -> str: return {False: "0", True: "1"}[self.output_interlock_tripped] # Source function mode - @message(r'^:?SOUR:FUNC(?::MODE)?\?$') + @message(r":?SOUR:FUNC(?::MODE)?\?$") def get_source_function_mode(self) -> str: return self.source_function_mode - @message(r'^:?SOUR:FUNC(?::MODE)? (VOLT|CURR)$') + @message(r":?SOUR:FUNC(?::MODE)?\s+(VOLT|CURR)$") def set_source_function_mode(self, function) -> None: try: self.source_function_mode = function @@ -120,11 +120,11 @@ def set_source_function_mode(self, function) -> None: # Source levels - @message(r'^:?SOUR:(VOLT|CURR)(?::LEV)?\?$') + @message(r":?SOUR:(VOLT|CURR)(?::LEV)?\?$") def get_source_level(self, function) -> str: return format(self.source_level[function], "E") - @message(r'^:?SOUR:(VOLT|CURR)(?::LEV)? (.+)$') + @message(r":?SOUR:(VOLT|CURR)(?::LEV)?\s+(.+)$") def set_source_level(self, function, level) -> None: try: self.source_level[function] = float(level) @@ -133,11 +133,11 @@ def set_source_level(self, function, level) -> None: # Source range levels - @message(r'^:?SOUR:(VOLT|CURR):RANG\?$') + @message(r":?SOUR:(VOLT|CURR):RANG\?$") def get_source_range_level(self, function) -> str: return format(self.source_range[function], "E") - @message(r'^:?SOUR:(VOLT|CURR):RANG (.+)$') + @message(r":?SOUR:(VOLT|CURR):RANG\s+(.+)$") def set_source_range_level(self, function, level) -> None: try: self.source_range[function] = float(level) @@ -147,11 +147,11 @@ def set_source_range_level(self, function, level) -> None: # Source auto ranges - @message(r'^:?SOUR:(VOLT|CURR):RANG:AUTO\?$') + @message(r":?SOUR:(VOLT|CURR):RANG:AUTO\?$") def get_source_range_auto(self, function) -> int: return int(self.source_range_auto[function]) - @message(r'^:?SOUR:(VOLT|CURR):RANG:AUTO (.+)$') + @message(r":?SOUR:(VOLT|CURR):RANG:AUTO\s+(.+)$") def set_source_range_auto(self, function, state) -> None: try: self.source_range_auto[function] = {"ON": True, "OFF": False, "0": False, "1": True}[state] @@ -160,11 +160,11 @@ def set_source_range_auto(self, function, state) -> None: # Source voltage limit - @message(r'^:?SOUR:VOLT:PROT(?::LEV)?\?$') + @message(r":?SOUR:VOLT:PROT(?::LEV)?\?$") def get_source_voltage_protection_level(self) -> str: return format(self.source_voltage_protection_level, "E") - @message(r'^:?SOUR:VOLT:PROT(?::LEV)? (.+)$') + @message(r":?SOUR:VOLT:PROT(?::LEV)?\s+(.+)$") def set_source_voltage_protection_level(self, level) -> None: try: self.source_voltage_protection_level = float(level) @@ -173,147 +173,147 @@ def set_source_voltage_protection_level(self, level) -> None: # Source compliance - @message(r'^:?SOUR:VOLT:ILIM(?::LEV)?\?$') + @message(r":?SOUR:VOLT:ILIM(?::LEV)?\?$") def get_source_voltage_ilimit_level(self) -> str: return format(self.source_voltage_ilimit_level, "E") - @message(r'^:?SOUR:VOLT:ILIM(?::LEV)? (.+)$') + @message(r":?SOUR:VOLT:ILIM(?::LEV)?\s+(.+)$") def set_source_voltage_ilimit_level(self, level) -> None: try: self.source_voltage_ilimit_level = float(level) except ValueError: self.error_queue.append(Error(101, "malformed command")) - @message(r'^:?SOUR:VOLT:ILIM(?::LEV)?:TRIP\?$') + @message(r":?SOUR:VOLT:ILIM(?::LEV)?:TRIP\?$") def get_source_voltage_ilimit_level_tripped(self) -> str: return format(False, "E") # TODO - @message(r'^:?SOUR:CURR:VLIM(?::LEV)?\?$') + @message(r":?SOUR:CURR:VLIM(?::LEV)?\?$") def get_source_current_vlimit_level(self) -> str: return format(self.source_current_vlimit_level, "E") - @message(r'^:?SOUR:CURR:VLIM(?::LEV)? (.+)$') + @message(r":?SOUR:CURR:VLIM(?::LEV)?\s+(.+)$") def set_source_current_vlimit_level(self, level) -> None: try: self.source_current_vlimit_level = float(level) except ValueError: self.error_queue.append(Error(101, "malformed command")) - @message(r'^:?SOUR:CURR:VLIM(?::LEV)?:TRIP\?$') + @message(r":?SOUR:CURR:VLIM(?::LEV)?:TRIP\?$") def get_source_current_vlimit_level_tripped(self) -> str: return format(False, "E") # TODO - @message(r'^:?SENS:FUNC(?::ON)?\s+\"(CURR|RES|VOLT)\"$') + @message(r":?SENS:FUNC(?::ON)?\s+\"(CURR|RES|VOLT)\"$") def set_sense_function_on(self, function: str) -> None: self.sense_function_on = function # Average - @message(r'^:?SENS:(VOLT|CURR):AVER:TCON\?$') + @message(r":?SENS:(VOLT|CURR):AVER:TCON\?$") def get_sense_average_tcontrol(self, function: str) -> str: return format(self.sense_average_tcontrol[function], "E") - @message(r'^:?SENS:(VOLT|CURR):AVER:TCON (MOV|REP)$') + @message(r":?SENS:(VOLT|CURR):AVER:TCON\s+(MOV|REP)$") def set_sense_average_tcontrol(self, function: str, tcontrol: str) -> None: self.sense_average_tcontrol[function] = tcontrol - @message(r'^:?SENS:(VOLT|CURR):AVER:COUN[T]?\?$') + @message(r":?SENS:(VOLT|CURR):AVER:COUN[T]?\?$") def get_sense_average_count(self, function: str) -> str: return format(self.sense_average_count[function], "E") - @message(r':?SENS:(VOLT|CURR):AVER:COUN[T]? (\d+)$') + @message(r":?SENS:(VOLT|CURR):AVER:COUN[T]?\s+(\d+)$") def set_sense_average_count(self, function: str, count: str) -> None: self.sense_average_count[function] = int(count) - @message(r'^:?SENS:(VOLT|CURR):AVER:STAT[E]?\?$') + @message(r":?SENS:(VOLT|CURR):AVER:STAT[E]?\?$") def get_sense_average_state(self, function: str) -> str: return format(self.sense_average_state[function], "E") - @message(r'^:?SENS:(VOLT|CURR):AVER:STAT[E]? (OFF|ON|0|1)$') + @message(r":?SENS:(VOLT|CURR):AVER:STAT[E]?\s+(OFF|ON|0|1)$") def set_sense_average_state(self, function: str, state: str) -> None: self.sense_average_state[function] = {"OFF": False, "ON": True, "0": False, "1": True}[state] # Integration time - @message(r'^(?::?SENS)?:(?:VOLT|CURR|RES):NPLC\?$') + @message(r"(?::?SENS)?:(?:VOLT|CURR|RES):NPLC\?$") def get_sense_nplc(self) -> str: return format(self.sense_nplc, "E") - @message(r'^(?::?SENS)?:(?:VOLT|CURR|RES):NPLC (.+)$') + @message(r"(?::?SENS)?:(?:VOLT|CURR|RES):NPLC\s+(.+)$") def set_sense_nplc(self, nplc: str) -> None: self.sense_nplc = round(float(nplc), 2) # Measure - @message(r':?READ\?$') + @message(r":?READ\?$") def get_read(self) -> str: read = self._read_current() return format(read, "E") - @message(r':?READ\? \"[a-zA-Z0-9_]+\", SOUR, READ$') - def get_read_elements(self) -> str: + @message(r":?READ\?\s+\"([a-zA-Z0-9_]+)\",\s+SOUR,\s+READ$") + def get_read_elements(self, _buffer: str) -> str: sour = self._read_voltage() read = self._read_current() return f"{sour:E},{read:E}" - @message(r'^:?INIT$') + @message(r":?INIT(?::IMM)?$") def set_init(self) -> None: ... - @message(r'^:?MEAS:VOLT\?$') + @message(r":?MEAS:VOLT\?$") def get_measure_voltage(self) -> str: volt = self._read_voltage() return format(volt, "E") - @message(r'^:?MEAS:CURR\?$') + @message(r":?MEAS:CURR\?$") def get_measure_current(self) -> str: curr = self._read_current() return format(curr, "E") - @message(r'^:?TRAC:CLE \"[a-zA-Z0-9_]+\"$') - def set_trace_clear(self) -> None: + @message(r":?TRAC[E]?:CLE\s+\"([a-zA-Z0-9_]+)\"$") + def set_trace_clear(self, _buffer: str) -> None: ... - @message(r'^:?TRAC:TRIG \"[a-zA-Z0-9_]+\"$') - def set_trace_trigger(self) -> None: + @message(r":?TRAC[E]?:TRIG\s+\"([a-zA-Z0-9_]+)\"$") + def set_trace_trigger(self, _buffer: str) -> None: ... - @message(r'^:?TRAC:DATA\? 1, 1, \"[a-zA-Z0-9_]+\", SOUR, READ$') - def get_trace_data(self) -> str: + @message(r":?TRAC[E]?:DATA\?\s+1,\s+1,\s+\"([a-zA-Z0-9_]+)\",\s+SOUR,\s+READ$") + def get_trace_data(self, _buffer: str) -> str: sour = self._read_voltage() read = self._read_current() return f"{sour:E},{read:E}" # TSP - @message(r'^reset\(\)$') + @message(r"reset\(\)$") def write_reset(self) -> None: self.error_queue.clear() - @message(r'^clear\(\)$') + @message(r"clear\(\)$") def write_clear(self) -> None: self.error_queue.clear() - @message(r'^errorqueue\.clear\(\)$') + @message(r"errorqueue\.clear\(\)$") def set_errorqueue_clear(self) -> None: self.error_queue.clear() - @message(tsp_print(r'errorqueue\.count')) + @message(tsp_print(r"errorqueue\.count")) def get_errorqueue_count(self) -> int: return len(self.error_queue) - @message(tsp_print(r'errorqueue\.next\(\)')) + @message(tsp_print(r"errorqueue\.next\(\)")) def get_errorqueue_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) return f"{error.code}, \"{error.message}\", 0, 0" return "0, \"Queue is Empty\", 0, 0" - @message(tsp_print(r'smu\.source\.output')) + @message(tsp_print(r"smu\.source\.output")) def get_tsp_source_output(self) -> int: return {False: 0, True: 1}[self.smu_source_output] - @message(tsp_assign(r'smu\.source\.output')) + @message(tsp_assign(r"smu\.source\.output")) def set_tsp_source_output(self, value) -> None: try: self.smu_source_output = { @@ -323,7 +323,7 @@ def set_tsp_source_output(self, value) -> None: except KeyError: self.error_queue.append(Error(101, "malformed command")) - @message(r'^.*$') + @message(r".*") def unknown_message(self) -> None: self.error_queue.append(Error(101, "malformed command")) diff --git a/src/comet/emulator/keithley/k2657a.py b/src/comet/emulator/keithley/k2657a.py index f233d3c..9ba4651 100644 --- a/src/comet/emulator/keithley/k2657a.py +++ b/src/comet/emulator/keithley/k2657a.py @@ -25,7 +25,7 @@ def __init__(self) -> None: self.smua_measure_nplc: float = 1.0 self.source_protectv: float = 0. - @message(r'^reset\(\)$') + @message(r"reset\(\)$") def set_reset(self): self.error_queue.clear() self.beeper_enable = True @@ -42,23 +42,23 @@ def set_reset(self): self.smua_measure_nplc = 1.0 self.source_protectv = 0. - @message(r'^status.reset\(\)$') + @message(r"status.reset\(\)$") def set_status_reset(self) -> None: self.error_queue.clear() - @message(r'^clear\(\)$') + @message(r"clear\(\)$") def set_clear(self) -> None: self.error_queue.clear() - @message(r'^errorqueue\.clear\(\)$') + @message(r"errorqueue\.clear\(\)$") def set_errorqueue_clear(self) -> None: self.error_queue.clear() - @message(tsp_print(r'errorqueue\.count')) + @message(tsp_print(r"errorqueue\.count")) def get_errorqueue_count(self) -> str: return format(len(self.error_queue), "d") - @message(tsp_print(r'errorqueue\.next\(\)')) + @message(tsp_print(r"errorqueue\.next\(\)")) def get_errorqueue_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -68,11 +68,11 @@ def get_errorqueue_next(self) -> str: # Beeper - @message(tsp_print(r'beeper\.enabled')) + @message(tsp_print(r"beeper\.enabled")) def get_beeper_enabled(self) -> str: return format(self.beeper_enable, "d") - @message(tsp_assign(r'beeper\.enable')) + @message(tsp_assign(r"beeper\.enable")) def set_beeper_enable(self, enable: str) -> None: try: self.beeper_enable = { @@ -84,11 +84,11 @@ def set_beeper_enable(self, enable: str) -> None: # Display - @message(tsp_print(r'display\.smua\.measure\.func')) + @message(tsp_print(r"display\.smua\.measure\.func")) def get_display_measure_function(self) -> str: return format(self.display_measure_function, "E") - @message(tsp_assign(r'display\.smua\.measure\.func')) + @message(tsp_assign(r"display\.smua\.measure\.func")) def set_display_measure_function(self, func) -> None: try: self.display_measure_function = { @@ -102,11 +102,11 @@ def set_display_measure_function(self, func) -> None: # Source output - @message(tsp_print(r'smua\.source\.output')) + @message(tsp_print(r"smua\.source\.output")) def get_source_output(self) -> str: return format(self.smua_source_output, "E") - @message(tsp_assign(r'smua\.source\.output')) + @message(tsp_assign(r"smua\.source\.output")) def set_source_output(self, state) -> None: try: self.smua_source_output = { @@ -118,11 +118,11 @@ def set_source_output(self, state) -> None: # Source function - @message(tsp_print(r'smua\.source\.func')) + @message(tsp_print(r"smua\.source\.func")) def get_source_function(self) -> str: return format({"DCAMPS": 0, "DCVOLTS": 1}[self.smua_source_function], "E") - @message(tsp_assign(r'smua\.source\.func')) + @message(tsp_assign(r"smua\.source\.func")) def set_source_function(self, function) -> None: try: self.smua_source_function = { @@ -134,11 +134,11 @@ def set_source_function(self, function) -> None: # Source levels - @message(tsp_print(r'smua\.source\.level([iv])')) + @message(tsp_print(r"smua\.source\.level([iv])")) def get_source_level(self, function) -> str: return format(self.smua_source_level[function], "E") - @message(tsp_assign(r'smua\.source\.level([iv])')) + @message(tsp_assign(r"smua\.source\.level([iv])")) def set_source_level(self, function, level) -> None: try: self.smua_source_level[function] = float(level) @@ -147,11 +147,11 @@ def set_source_level(self, function, level) -> None: # Source ranges - @message(tsp_print(r'smua\.source\.range([iv])')) + @message(tsp_print(r"smua\.source\.range([iv])")) def get_source_range(self, function) -> str: return format(self.smua_source_range[function], "E") - @message(tsp_assign(r'smua\.source\.range([iv])')) + @message(tsp_assign(r"smua\.source\.range([iv])")) def set_source_range(self, function, level) -> None: try: self.smua_source_range[function] = float(level) @@ -160,11 +160,11 @@ def set_source_range(self, function, level) -> None: # Source autoranges - @message(tsp_print(r'smua\.source\.autorange([iv])')) + @message(tsp_print(r"smua\.source\.autorange([iv])")) def get_source_autorange(self, function) -> str: return format(self.smua_source_autorange[function], "E") - @message(tsp_assign(r'smua\.source\.autorange([iv])')) + @message(tsp_assign(r"smua\.source\.autorange([iv])")) def set_source_autorange(self, function, state) -> None: try: self.smua_source_autorange[function] = { @@ -176,11 +176,11 @@ def set_source_autorange(self, function, state) -> None: # Source voltage limit - @message(tsp_print(r'smua\.source\.protectv')) + @message(tsp_print(r"smua\.source\.protectv")) def get_source_protectv(self) -> str: return format(self.source_protectv, "E") - @message(tsp_assign(r'smua\.source\.protectv')) + @message(tsp_assign(r"smua\.source\.protectv")) def set_source_protectv(self, level) -> None: try: self.source_protectv = float(level) @@ -189,38 +189,38 @@ def set_source_protectv(self, level) -> None: # Compliance - @message(tsp_print(r'smua.source.compliance')) + @message(tsp_print(r"smua.source.compliance")) def get_source_compliance(self) -> str: return "false" - @message(tsp_print(r'smua\.source\.limit([iv])')) + @message(tsp_print(r"smua\.source\.limit([iv])")) def get_source_limit(self, function) -> str: return format(self.smua_source_limit[function], "E") - @message(tsp_assign(r'smua\.source\.limit([iv])')) + @message(tsp_assign(r"smua\.source\.limit([iv])")) def set_source_limit(self, function, level) -> None: try: self.smua_source_limit[function] = float(level) except ValueError: self.error_queue.append(Error(117, "malformed command")) - @message(tsp_print(r'smua\.measure\.i\(\)')) + @message(tsp_print(r"smua\.measure\.i\(\)")) def get_measure_i(self) -> str: curr_min = float(self.options.get("curr.min", 1e-6)) curr_max = float(self.options.get("curr.max", 1e-7)) return format(random.uniform(curr_min, curr_max), "E") - @message(tsp_print(r'smua\.measure\.v\(\)')) + @message(tsp_print(r"smua\.measure\.v\(\)")) def get_measure_v(self) -> str: return format(self.smua_source_level.get("v", 0) + random.uniform(-.25, +.25), "E") # Average - @message(tsp_print(r'smua\.measure\.filter\.enable')) + @message(tsp_print(r"smua\.measure\.filter\.enable")) def get_measure_filter_enable(self) -> str: return format(self.smua_measure_filter_enable, "d") - @message(tsp_assign(r'smua\.measure\.filter\.enable')) + @message(tsp_assign(r"smua\.measure\.filter\.enable")) def set_measure_filter_enable(self, enable: str) -> None: try: self.smua_measure_filter_enable = { @@ -230,22 +230,22 @@ def set_measure_filter_enable(self, enable: str) -> None: except KeyError: self.error_queue.append(Error(118, "malformed command")) - @message(tsp_print(r'smua\.measure\.filter\.count')) + @message(tsp_print(r"smua\.measure\.filter\.count")) def get_measure_filter_count(self) -> str: return format(self.smua_measure_filter_count, "d") - @message(tsp_assign(r'smua\.measure\.filter\.count')) + @message(tsp_assign(r"smua\.measure\.filter\.count")) def set_measure_filter_count(self, count: str) -> None: try: self.smua_measure_filter_count = int(count) except KeyError: self.error_queue.append(Error(119, "malformed command")) - @message(tsp_print(r'smua\.measure\.filter\.type')) + @message(tsp_print(r"smua\.measure\.filter\.type")) def get_measure_filter_type(self) -> str: return format(self.smua_measure_filter_type, "d") - @message(tsp_assign(r'smua\.measure\.filter\.type')) + @message(tsp_assign(r"smua\.measure\.filter\.type")) def set_measure_filter_type(self, enable: str) -> None: try: self.smua_measure_filter_type = { @@ -258,18 +258,18 @@ def set_measure_filter_type(self, enable: str) -> None: # Integration time - @message(tsp_print(r'smua\.measure\.nplc')) + @message(tsp_print(r"smua\.measure\.nplc")) def get_measure_nplc(self) -> str: return format(self.smua_measure_nplc, "E") - @message(tsp_assign(r'smua\.measure\.nplc')) + @message(tsp_assign(r"smua\.measure\.nplc")) def set_measure_nplc(self, nplc: str) -> None: try: self.smua_measure_nplc = round(float(nplc), 3) except KeyError: self.error_queue.append(Error(120, "malformed command")) - @message(r'^.*$') + @message(r".*") def unknown_message(self) -> None: self.error_queue.append(Error(100, "malformed command")) diff --git a/src/comet/emulator/keithley/k2700.py b/src/comet/emulator/keithley/k2700.py index 8625d7b..ce452fa 100644 --- a/src/comet/emulator/keithley/k2700.py +++ b/src/comet/emulator/keithley/k2700.py @@ -84,7 +84,7 @@ def __init__(self) -> None: self.trigger_delay_auto: bool = True self.trigger_delay: float = 0.001 - @message(r'^\*RST$') + @message(r"\*RST$") def set_rst(self) -> None: self.reading_number = 0 self.format_elements = FormatElements() @@ -97,17 +97,17 @@ def set_rst(self) -> None: self.trigger_delay = 0.001 self.error_queue.clear() - @message(r'^\*CLS$') + @message(r"\*CLS$") def set_cls(self) -> None: self.error_queue.clear() # Format - @message(r'^:?FORM(?:AT)?:ELEM(?:ENTS)?\?$') + @message(r":?FORM(?:AT)?:ELEM(?:ENTS)?\?$") def get_format_elements(self) -> str: return str(self.format_elements) - @message(r'^:?FORM(?:AT)?:ELEM(?:ENTS)?\s+(.*)$') + @message(r":?FORM(?:AT)?:ELEM(?:ENTS)?\s+(.*)$") def set_format_elements(self, elements) -> None: error = self.format_elements.from_text(elements) if error: @@ -115,39 +115,39 @@ def set_format_elements(self, elements) -> None: # Sense - @message(r'^:?SENS(?:E)?:FUNC(?:TION)?\?$') + @message(r":?SENS(?:E)?:FUNC(?:TION)?\?$") def get_sense_function(self) -> str: return f"\"{self.sense_function}\"" - @message(r'^:?SENS(?:E)?:FUNC(?:TION)\s+\"(VOLT|CURR|VOLT:DC|CURR:DC|TEMP)\"$') + @message(r":?SENS(?:E)?:FUNC(?:TION)\s+\"(VOLT|CURR|VOLT:DC|CURR:DC|TEMP)\"$") def set_sense_function(self, function: str) -> None: self.sense_function = {"VOLT": "VOLT:DC", "CURR": "CURR:DC"}.get(function, function) - @message(r'^:?SENS(?:E)?:VOLT:AVER:TCON\?$') + @message(r":?SENS(?:E)?:VOLT:AVER:TCON\?$") def get_sense_average_tcontrol(self) -> str: return self.sense_voltage_average_tcontrol - @message(r'^:?SENS(?:E)?:VOLT:AVER:TCON\s+(REP|MOV)$') + @message(r":?SENS(?:E)?:VOLT:AVER:TCON\s+(REP|MOV)$") def set_sense_average_tcontrol(self, name: str) -> None: self.sense_voltage_average_tcontrol = name - @message(r'^:?SENS(?:E)?:VOLT:AVER:COUN[T]?\?$') + @message(r":?SENS(?:E)?:VOLT:AVER:COUN[T]?\?$") def get_sense_average_count(self) -> str: return format(self.sense_voltage_average_count, "d") - @message(r'^:?SENS(?:E)?:VOLT:AVER:COUN[T]?\s+(\d+)$') + @message(r":?SENS(?:E)?:VOLT:AVER:COUN[T]?\s+(\d+)$") def set_sense_average_count(self, count: str) -> None: self.sense_voltage_average_count = int(count) - @message(r'^:?SENS(?:E)?:VOLT:AVER(?::STAT[E]?)?\?$') + @message(r":?SENS(?:E)?:VOLT:AVER(?::STAT[E]?)?\?$") def get_sense_voltage_average_state(self) -> str: return format(self.sense_voltage_average_state, "d") - @message(r'^:?SENS(?:E)?:VOLT:AVER(?::STAT[E]?)?\s+(OFF|ON|0|1)$') + @message(r":?SENS(?:E)?:VOLT:AVER(?::STAT[E]?)?\s+(OFF|ON|0|1)$") def set_sense_voltage_average_state(self, value) -> None: self.sense_voltage_average_state = {"0": False, "1": True, "OFF": False, "ON": True}[value] - @message(r'^:?SYST:ERR\?$') + @message(r":?SYST:ERR\?$") def get_system_error(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -155,65 +155,65 @@ def get_system_error(self) -> str: error = Error(0, "no error") return f'{error.code}, "{error.message}"' - @message(r'^:?SYST:BEEP(?::STAT)?\?$') + @message(r":?SYST:BEEP(?::STAT)?\?$") def get_beeper_state(self) -> str: return format(self.system_beeper_state, "d") - @message(r'^:?SYST:BEEP(?::STAT)? (OFF|ON|0|1)$') + @message(r":?SYST:BEEP(?::STAT)? (OFF|ON|0|1)$") def set_beeper_state(self, value) -> None: self.system_beeper_state = {'0': False, '1': True, 'OFF': False, 'ON': True}[value] - @message(r'^:?INIT(?::IMM)$') + @message(r":?INIT(?::IMM)$") def set_init(self) -> None: ... - @message(r'^:?READ\?$') + @message(r":?READ\?$") def get_read(self) -> None: return self._read() - @message(r'^:?FETC[H]?\?$') + @message(r":?FETC[H]?\?$") def get_fetch(self) -> str: return self._read() # Measure - @message(r'^:?MEAS:VOLT\?$') + @message(r":?MEAS:VOLT\?$") def get_measure_voltage(self) -> str: self.sense_function = "VOLT:DC" return self._read() - @message(r'^:?MEAS:CURR\?$') + @message(r":?MEAS:CURR\?$") def get_measure_current(self) -> str: self.sense_function = "CURR:DC" return self._read() - @message(r'^:?MEAS:TEMP\?$') + @message(r":?MEAS:TEMP\?$") def get_measure_temperature(self) -> str: self.sense_function = "TEMP" return self._read() # Trigger - @message(r':?TRIG:DEL:AUTO\?$') + @message(r":?TRIG:DEL:AUTO\?$") def get_trigger_delay_auto(self) -> str: return format(self.trigger_delay_auto, "d") - @message(r':?TRIG:DEL:AUTO\s+(OFF|ON|0|1)$') + @message(r":?TRIG:DEL:AUTO\s+(OFF|ON|0|1)$") def set_trigger_delay_auto(self, value) -> None: self.trigger_delay_auto = {'0': False, '1': True, 'OFF': False, 'ON': True}[value] - @message(r':?TRIG:DEL\?$') + @message(r":?TRIG:DEL\?$") def get_trigger_delay(self) -> str: return format(self.trigger_delay, "E") - @message(r':?TRIG:DEL\s+(.+)$') - def set_trigger_delay(self, value) -> None: + @message(r":?TRIG:DEL\s+(.+)$") + def set_trigger_delay(self, value: str) -> None: try: self.trigger_delay = min(999999.999, max(0.001, float(value))) except ValueError: self.error_queue.append(Error(-113, "undefined header")) - @message(r'^(.*)$') - def unknown_message(self, request) -> None: + @message(r".*") + def unknown_message(self) -> None: self.error_queue.append(Error(101, "malformed command")) def _read(self): diff --git a/src/comet/emulator/keithley/k4215cvu.py b/src/comet/emulator/keithley/k4215cvu.py index 918f4ac..ff53a7b 100644 --- a/src/comet/emulator/keithley/k4215cvu.py +++ b/src/comet/emulator/keithley/k4215cvu.py @@ -100,11 +100,11 @@ def _drift(self) -> float: def _effective_bias(self) -> float: return self.dcv + self.dcv_offset - @message(r'^\*IDN\?$') + @message(r"\*IDN\?$") def get_idn(self) -> str: return self.IDENTITY - @message(r'^\*RST$') + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() @@ -134,15 +134,15 @@ def set_rst(self) -> None: # keep drift seed but reset time reference self._t0 = time.time() - @message(r'^\*CLS$') + @message(r"\*CLS$") def set_cls(self) -> None: self.error_queue.clear() - @message(r"^BC\s*$") + @message(r"BC$") def set_buffer_clear(self) -> None: ... - @message(r'^:?ERROR:LAST:GET$') + @message(r":?ERROR:LAST:GET$") def get_last_error(self) -> str: if self.error_queue: error = self.error_queue[-1] @@ -150,12 +150,12 @@ def get_last_error(self) -> str: error = Error(0, "No error") return f'{error.code}, "{error.message}"' - @message(r'^:?ERROR:LAST:CLEAR$') + @message(r":?ERROR:LAST:CLEAR$") def clear_last_error(self) -> None: if self.error_queue: self.error_queue.pop() - @message(r'^:CVU:MODE\s(\d+)$') + @message(r":CVU:MODE\s+(\d+)$") def set_mode(self, mode: str) -> None: m = self._parse_int(mode) if m is None: @@ -166,27 +166,27 @@ def set_mode(self, mode: str) -> None: return self.cvu_mode = m - @message(r'^:CVU:MODE\?$') + @message(r":CVU:MODE\?$") def get_mode(self) -> str: return str(self.cvu_mode) - @message(r"^:CVU:CONFIG:ACVHI\s(1|2)$") + @message(r":CVU:CONFIG:ACVHI\s+(1|2)$") def set_config_acvhi(self, val: str) -> None: self.config_acvhi = int(val) - @message(r"^:CVU:CONFIG:ACVHI\?$") + @message(r":CVU:CONFIG:ACVHI\?$") def get_config_acvhi(self) -> str: return str(self.config_acvhi) - @message(r"^:CVU:CONFIG:DCVHI\s(1|2)$") + @message(r":CVU:CONFIG:DCVHI\s+(1|2)$") def set_config_dcvhi(self, val: str) -> None: self.config_dcvhi = int(val) - @message(r"^:CVU:CONFIG:DCVHI\?$") + @message(r":CVU:CONFIG:DCVHI\?$") def get_config_dcvhi(self) -> str: return str(self.config_dcvhi) - @message(r"^:CVU:ACZ:RANGE\s(.+)$") + @message(r":CVU:ACZ:RANGE\s+(.+)$") def set_acz_range(self, level: str) -> None: v = self._parse_float(level) if v is None: @@ -196,28 +196,28 @@ def set_acz_range(self, level: str) -> None: return self.cvu_acz_range = v - @message(r"^:CVU:ACZ:RANGE\?$") + @message(r":CVU:ACZ:RANGE\?$") def get_acz_range(self) -> str: return f"{self.cvu_acz_range:g}" - @message(r"^:CVU:OUTPUT\s(0|1)$") + @message(r":CVU:OUTPUT\s+(0|1)$") def set_cvu_output(self, state: str) -> None: self.cvu_output = bool(int(state)) - @message(r"^:CVU:OUTPUT\?$") + @message(r":CVU:OUTPUT\?$") def get_cvu_output(self) -> str: return "1" if self.cvu_output else "0" - @message(r"^:CVU:CORRECT\s(0|1),(0|1),(0|1)$") + @message(r":CVU:CORRECT\s+(0|1),(0|1),(0|1)$") def set_correction(self, open_val: str, short_val: str, load_val: str) -> None: self.cvu_correction = Correction(int(open_val), int(short_val), int(load_val)) - @message(r"^:CVU:CORRECT\?$") + @message(r":CVU:CORRECT\?$") def get_correction(self) -> str: open_val, short_val, load_val = astuple(self.cvu_correction) return f"{open_val},{short_val},{load_val}" - @message(r"^:CVU:LENGTH\s(.+)$") + @message(r":CVU:LENGTH\s+(.+)$") def set_length(self, length_m: str) -> None: v = self._parse_float(length_m) if v is None: @@ -230,12 +230,12 @@ def set_length(self, length_m: str) -> None: return self.cvu_length_m = v - @message(r"^:CVU:LENGTH\?$") + @message(r":CVU:LENGTH\?$") def get_length(self) -> str: # keep one decimal like typical instruments return f"{self.cvu_length_m:.1f}" - @message(r"^:CVU:ACV\s(.+)$") + @message(r":CVU:ACV\s+(.+)$") def set_acv(self, voltage: str) -> None: v = self._parse_float(voltage) if v is None: @@ -245,11 +245,11 @@ def set_acv(self, voltage: str) -> None: return self.acv = v - @message(r"^:CVU:ACV\?$") + @message(r":CVU:ACV\?$") def get_acv(self) -> str: return f"{self.acv:.6E}" - @message(r"^:CVU:FREQ\s(.+)$") + @message(r":CVU:FREQ\s+(.+)$") def set_freq(self, frequency: str) -> None: v = self._parse_float(frequency) if v is None: @@ -259,11 +259,11 @@ def set_freq(self, frequency: str) -> None: return self.freq_hz = int(round(v)) - @message(r"^:CVU:FREQ\?$") + @message(r":CVU:FREQ\?$") def get_freq(self) -> str: return str(self.freq_hz) - @message(r"^:CVU:DCV\s(.+)$") + @message(r":CVU:DCV\s+(.+)$") def set_dcv(self, level: str) -> None: v = self._parse_float(level) if v is None: @@ -273,11 +273,11 @@ def set_dcv(self, level: str) -> None: return self.dcv = v - @message(r"^:CVU:DCV\?$") + @message(r":CVU:DCV\?$") def get_dcv(self) -> str: return f"{self.dcv:.3E}" - @message(r"^:CVU:DCV:OFFSET\s(.+)$") + @message(r":CVU:DCV:OFFSET\s+(.+)$") def set_dcv_offset(self, offset: str) -> None: v = self._parse_float(offset) if v is None: @@ -287,11 +287,11 @@ def set_dcv_offset(self, offset: str) -> None: return self.dcv_offset = v - @message(r"^:CVU:DCV:OFFSET\?$") + @message(r":CVU:DCV:OFFSET\?$") def get_dcv_offset(self) -> str: return f"{self.dcv_offset:.3E}" - @message(r"^:CVU:SPEED\s(\d+),(.+),(.+),(.+)$") + @message(r":CVU:SPEED\s+(\d+),(.+),(.+),(.+)$") def set_speed(self, mode: str, delay_factor: str, filter_factor: str, aperture: str) -> None: m = self._parse_int(mode) d = self._parse_float(delay_factor) @@ -318,12 +318,12 @@ def set_speed(self, mode: str, delay_factor: str, filter_factor: str, aperture: self.filter_factor = f self.aperture_plc = a - @message(r"^:CVU:SPEED\?$") + @message(r":CVU:SPEED\?$") def get_speed(self) -> str: # Keep scientific formatting like typical SCPI return f"{self.speed_mode},{self.delay_factor:.3E},{self.filter_factor:.3E},{self.aperture_plc:.3E}" - @message(r"^:CVU:MODEL\s(.+)$") + @message(r":CVU:MODEL\s+(.+)$") def set_model(self, model: str) -> None: token = model.strip().upper() if token in self.MODEL_MAP: @@ -338,12 +338,12 @@ def set_model(self, model: str) -> None: return self.model_code = code - @message(r"^:CVU:MODEL\?$") + @message(r":CVU:MODEL\?$") def get_model(self) -> str: # return numeric code like many instruments do return str(self.model_code) - @message(r"^:CVU:MEASZ\?$") + @message(r":CVU:MEASZ\?$") def get_measz(self) -> str: """ Return two comma-separated values. @@ -457,7 +457,7 @@ def get_measz(self) -> str: return f"{a:.6E},{b:.6E}" - @message(r'^(.*)$') + @message(r"(.*)$") def unknown_message(self, request: str) -> None: # ignore empty lines quietly if request.strip() == "": diff --git a/src/comet/emulator/keithley/k6510.py b/src/comet/emulator/keithley/k6510.py index edf17f2..3eae475 100644 --- a/src/comet/emulator/keithley/k6510.py +++ b/src/comet/emulator/keithley/k6510.py @@ -14,19 +14,19 @@ def __init__(self) -> None: super().__init__() self.error_queue: list[Error] = [] - @message(r'^\*RST$') + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() - @message(r'^\*CLS$') + @message(r"\*CLS$") def set_cls(self) -> None: self.error_queue.clear() - @message(r'^:?SYST:ERR:COUN\?$') + @message(r":?SYST:ERR(?:or)?:COUN[T]?\?$") def get_system_error_count(self) -> str: return format(len(self.error_queue), "d") - @message(r'^:?SYST:ERR(?::NEXT)?\?$') + @message(r":?SYST:ERR(?:or)?(?::NEXT)?\?$") def get_system_error_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -36,7 +36,7 @@ def get_system_error_next(self) -> str: # Route terminal - @message(r'^:?ROUT(?:e)?:TERM(?:inal(?:s)?)?\?$') + @message(r":?ROUT(?:e)?:TERM(?:inal(?:s)?)?\?$") def get_route_terminals(self) -> str: value = self.options.get("route.terminals", "front") if value.lower().startswith("rear"): @@ -45,20 +45,20 @@ def get_route_terminals(self) -> str: # Measure - @message(r'^:?MEAS(?:ure)?:VOLT(?:age)?\?$') + @message(r":?MEAS(?:ure)?:VOLT(?:age)?\?$") def get_measure_voltage(self) -> str: volt_min = float(self.options.get("volt.min", 1e3)) volt_max = float(self.options.get("volt.max", 1e2)) return format(random.uniform(volt_min, volt_max), "E") - @message(r'^:?MEAS(?:ure)?:CURR(?:ent)?\?$') + @message(r":?MEAS(?:ure)?:CURR(?:ent)?\?$") def get_measure_current(self) -> str: curr_min = float(self.options.get("curr.min", 1e-6)) curr_max = float(self.options.get("curr.max", 1e-7)) return format(random.uniform(curr_min, curr_max), "E") - @message(r'^(.*)$') - def unknown_message(self, request) -> None: + @message(r".*") + def unknown_message(self) -> None: self.error_queue.append(Error(101, "malformed command")) diff --git a/src/comet/emulator/keithley/k6514.py b/src/comet/emulator/keithley/k6514.py index f16039b..0e6d1ba 100644 --- a/src/comet/emulator/keithley/k6514.py +++ b/src/comet/emulator/keithley/k6514.py @@ -22,7 +22,7 @@ def __init__(self) -> None: self.sense_current_range_auto: int = 1 self.sense_nplc: float = 5.0 - @message(r'^\*RST$') + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() self.zero_check = False @@ -35,15 +35,15 @@ def set_rst(self) -> None: self.sense_current_range_auto = 1 self.sense_nplc = 5.0 - @message(r'^\*CLS$') + @message(r"\*CLS$") def set_cls(self) -> None: self.error_queue.clear() - @message(r'^:?SYST:ERR:COUN\?$') + @message(r":?SYST:ERR:COUN[T]?\?$") def get_system_error_count(self) -> str: return format(len(self.error_queue), "d") - @message(r'^:?SYST:ERR(?::NEXT)?\?$') + @message(r":?SYST:ERR(?::NEXT)?\?$") def get_system_error_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -51,14 +51,14 @@ def get_system_error_next(self) -> str: error = Error(0, "no error") return f'{error.code}, "{error.message}"' - @message(r'^:?FORM:ELEM (READ)$') + @message(r":?FORM:ELEM\s+(READ)$") def set_format_elements(self, value) -> None: ... - @message(r'^:?FORM:ELEM\?$') + @message(r":?FORM:ELEM\?$") def get_format_elements(self) -> str: return "READ" - @message(r'^:?INIT$') + @message(r":?INIT(?::IMM)?$") def set_init(self) -> None: time.sleep(random.uniform(.5, 1.0)) @@ -73,109 +73,109 @@ def _reading(self) -> float: return random.uniform(volt_min, volt_max) return 0 - @message(r'^:?FETC[H]?\?$') + @message(r":?FETC[H]?\?$") def get_fetch(self) -> str: return format(self._reading(), "E") - @message(r'^:?READ\?$') + @message(r":?READ\?$") def get_read(self) -> str: time.sleep(random.uniform(.25, 1.0)) return format(self._reading(), "E") - @message(r'^:?SYST:ZCH\s+(0|1|ON|OFF)$') + @message(r":?SYST:ZCH\s+(0|1|ON|OFF)$") def set_zero_check(self, value) -> None: self.zero_check = {"0": False, "1": True, "OFF": False, "ON": True}[value] - @message(r'^:?SYST:ZCH\?$') + @message(r":?SYST:ZCH\?$") def get_zero_check(self) -> str: return {False: "0", True: "1"}[self.zero_check] - @message(r'^:?SYST:ZCOR\s+(0|1|ON|OFF)$') + @message(r":?SYST:ZCOR\s+(0|1|ON|OFF)$") def set_zero_correction(self, value) -> None: self.zero_correction = {"0": False, "1": True, "OFF": False, "ON": True}[value] - @message(r'^:?SYST:ZCOR\?$') + @message(r":?SYST:ZCOR\?$") def get_zero_correction(self) -> str: return {False: "0", True: "1"}[self.zero_correction] - @message(r'^:?SENS:FUNC \'(VOLT|CURR|RES|CHAR)(?:\:DC)?\'$') + @message(r":?SENS:FUNC\s+\'(VOLT|CURR|RES|CHAR)(?:\:DC)?\'$") def set_sense_function(self, value: str) -> None: self.sense_function = value - @message(r'^:?SENS:FUNC\?$') + @message(r":?SENS:FUNC\?$") def get_sense_function(self) -> str: return f"\"{self.sense_function}:DC\"" # Average - @message(r'^(?::?SENS)?:AVER:TCON\?$') + @message(r"(?::?SENS)?:AVER:TCON\?$") def get_sense_average_tcontrol(self) -> str: return self.sense_average_tcontrol - @message(r'^(?::?SENS)?:AVER:TCON (REP|MOV)$') + @message(r"(?::?SENS)?:AVER:TCON\s+(REP|MOV)$") def set_sense_average_tcontrol(self, state: str) -> None: self.sense_average_tcontrol = state - @message(r'^(?::?SENS)?:AVER:COUN\?$') + @message(r"(?::?SENS)?:AVER:COUN\?$") def get_sense_average_count(self) -> int: return self.sense_average_count - @message(r'^(?::?SENS)?:AVER:COUN (\d+)$') + @message(r"(?::?SENS)?:AVER:COUN\s+(\d+)$") def set_sense_average_count(self, count: str) -> None: self.sense_average_count = int(count) - @message(r'^(?::?SENS)?:AVER(?::STAT)?\?$') + @message(r"(?::?SENS)?:AVER(?::STAT)?\?$") def get_sense_average_state(self) -> int: return int(self.sense_average_state) - @message(r'^(?::?SENS)?:AVER(?::STAT)? (OFF|ON|0|1)$') + @message(r"(?::?SENS)?:AVER(?::STAT)?\s+(OFF|ON|0|1)$") def set_sense_average_state(self, state: str) -> None: self.sense_average_state = {"OFF": 0, "ON": 1, "0": 0, "1": 1}[state] # Current range - @message(r'^(?::?SENS)?:CURR:RANG\?$') + @message(r"(?::?SENS)?:CURR:RANG\?$") def get_sense_current_range(self) -> str: return format(self.sense_current_range, "E") - @message(r'^(?::?SENS)?:CURR:RANG(?::UPP)?\s+(.+)$') + @message(r"(?::?SENS)?:CURR:RANG(?::UPP)?\s+(.+)$") def set_sense_current_range(self, value: str) -> None: try: self.sense_current_range = float(value) except Exception: self.error_queue.append(Error(102, "invalid header")) - @message(r'^(?::?SENS)?:CURR:RANG:AUTO\?$') + @message(r"(?::?SENS)?:CURR:RANG:AUTO\?$") def get_sense_current_range_auto(self) -> int: return self.sense_current_range_auto - @message(r'^(?::?SENS)?:CURR:RANG:AUTO\s+(OFF|ON|0|1)$') + @message(r"(?::?SENS)?:CURR:RANG:AUTO\s+(OFF|ON|0|1)$") def set_sense_current_range_auto(self, state: str) -> None: self.sense_current_range_auto = {"OFF": 0, "ON": 1, "0": 0, "1": 1}[state] - @message(r'^(?::?SENS)?:CURR:RANG:AUTO:ULIM\s+(.+)$') + @message(r"(?::?SENS)?:CURR:RANG:AUTO:ULIM\s+(.+)$") def set_sense_current_range_auto_ulimit(self, value: str) -> None: ... # TODO - @message(r'^(?::?SENS)?:CURR:RANG:AUTO:LLIM\s+(.+)$') + @message(r"(?::?SENS)?:CURR:RANG:AUTO:LLIM\s+(.+)$") def set_sense_current_range_auto_llimit(self, value: str) -> None: ... # TODO # NPLC (coupled commands) - @message(r'^(?::?SENS)?:(:?CURR|VOLT|RES|CHAR):NPLC\?$') + @message(r"(?::?SENS)?:(?:CURR|VOLT|RES|CHAR):NPLC\?$") def get_sense_nplc(self) -> str: return format(self.sense_nplc, "f") # TODO precision? - @message(r'^(?::?SENS)?:(:?CURR|VOLT|RES|CHAR):NPLC\s+(.+)$') + @message(r"(?::?SENS)?:(CURR|VOLT|RES|CHAR):NPLC\s+(.+)$") def set_sense_nplc(self, mode: str, value: str) -> None: try: self.sense_nplc = max(0.01, min(10.0, float(value))) except Exception: self.error_queue.append(Error(102, "invalid header")) - @message(r'^(.*)$') - def unknown_message(self, request) -> None: + @message(r".*") + def unknown_message(self) -> None: self.error_queue.append(Error(101, "malformed command")) diff --git a/src/comet/emulator/keithley/k6517b.py b/src/comet/emulator/keithley/k6517b.py index 7f55498..5eee87c 100644 --- a/src/comet/emulator/keithley/k6517b.py +++ b/src/comet/emulator/keithley/k6517b.py @@ -25,7 +25,7 @@ def __init__(self) -> None: self.source_current_limit_state: bool = False self.sense_nplc: float = 5.0 - @message(r'^\*RST$') + @message(r"\*RST$") def set_reset(self) -> None: self.error_queue.clear() self.zero_check = False @@ -41,15 +41,15 @@ def set_reset(self) -> None: self.source_current_limit_state = False self.sense_nplc = 5.0 - @message(r'^\*CLS$') + @message(r"\*CLS$") def set_clear(self) -> None: self.error_queue.clear() - @message(r'^:?SYST:ERR:COUN\?$') + @message(r":?SYST:ERR:COUN\?$") def get_system_error_count(self) -> str: return format(len(self.error_queue), "d") - @message(r'^:?SYST:ERR(?::NEXT)?\?$') + @message(r":?SYST:ERR(?::NEXT)?\?$") def get_system_error_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -57,94 +57,94 @@ def get_system_error_next(self) -> str: error = Error(0, "no error") return f'{error.code}, "{error.message}"' - @message(r'^:?FORM:ELEM (READ)$') + @message(r":?FORM:ELEM\s+(READ)$") def set_format_elements(self, value) -> None: ... - @message(r'^:?FORM:ELEM\?$') + @message(r":?FORM:ELEM\?$") def get_format_elements(self) -> str: return "READ" - @message(r'^\*ESR\?$') + @message(r"\*ESR\?$") def get_esr(self) -> str: return format(random.randint(0, 1)) - @message(r'^:?SYST:ERR\?$') + @message(r":?SYST:ERR\?$") def get_system_error(self) -> str: return '0, "no error"' - @message(r'^:?SYST:ZCH\?$') + @message(r":?SYST:ZCH\?$") def get_system_zerocheck(self) -> str: return format(self.zero_check, "d") - @message(r'^:?SYST:ZCH\s+(OFF|ON)$') + @message(r":?SYST:ZCH\s+(OFF|ON)$") def set_system_zerocheck(self, value) -> None: self.zero_check = {"OFF": False, "ON": True}[value] - @message(r'^:?SYST:ZCOR(?:STAT)?\s+(0|1|ON|OFF)$') + @message(r":?SYST:ZCOR(?:STAT)?\s+(OFF|ON|0|1)$") def set_zero_correction(self, value) -> None: self.zero_correction = {"0": False, "1": True, "OFF": False, "ON": True}[value] - @message(r'^:?SYST:ZCOR(?:STAT)\?$') + @message(r":?SYST:ZCOR(?:STAT)\?$") def get_zero_correction(self) -> str: return {False: "0", True: "1"}[self.zero_correction] - @message(r'^:?SENS:FUNC \'(VOLT|CURR|RES|CHAR)(?:\:DC)?\'$') + @message(r":?SENS:FUNC\s+\'(VOLT|CURR|RES|CHAR)(?:\:DC)?\'$") def set_sense_function(self, value: str) -> None: self.sense_function = value - @message(r'^:?SENS:FUNC\?$') + @message(r":?SENS:FUNC\?$") def get_sense_function(self) -> str: return f'"{self.sense_function}:DC"' # Output - @message(r'^:?OUTP(?::STAT)?\s+(0|1|ON|OFF)$') + @message(r":?OUTP(?::STAT)?\s+(OFF|ON|0|1)$") def set_output_state(self, value) -> None: self.output_state = {"0": False, "1": True, "OFF": False, "ON": True}[value] - @message(r'^:?OUTP(?::STAT)?\?$') + @message(r":?OUTP(?::STAT)?\?$") def get_output_state(self) -> str: return {False: "0", True: "1"}[self.output_state] # Source - @message(r'^:?SOUR:VOLT(?::LEV(?::IMM(?::AMPL)?)?)?\s+(.+)$') + @message(r":?SOUR:VOLT(?::LEV(?::IMM(?::AMPL)?)?)?\s+(.+)$") def set_source_voltage_level_immediate_amplitude(self, level) -> None: try: self.source_voltage_level_immediate_amplitude = float(level) except ValueError: self.error_queue.append(Error(-171, "Invalid expression")) - @message(r'^:?SOUR:VOLT(?::LEV(?::IMM(?::AMPL)?)?)?\?$') + @message(r":?SOUR:VOLT(?::LEV(?::IMM(?::AMPL)?)?)?\?$") def get_source_voltage_level_immediate_amplitude(self) -> str: return format(self.source_voltage_level_immediate_amplitude, "E") - @message(r'^:?SOUR:VOLT:RANG\s+(.+)$') + @message(r":?SOUR:VOLT:RANG\s+(.+)$") def set_source_voltage_range(self, level) -> None: try: self.source_voltage_range = 100.0 if float(level) <= 100.0 else 1000.0 except ValueError: self.error_queue.append(Error(-171, "Invalid expression")) - @message(r'^:?SOUR:VOLT:RANG\?$') + @message(r":?SOUR:VOLT:RANG\?$") def get_source_voltage_range(self) -> str: return format(self.source_voltage_range, "E") - @message(r'^:?SOUR:CURR:LIM(?::STAT)?\?$') + @message(r":?SOUR:CURR:LIM(?::STAT)?\?$") def get_source_current_limit_state(self) -> str: return {False: "0", True: "1"}[self.source_current_limit_state] - @message(r'^:?SOUR:VOLT:MCON\s+(0|1|ON|OFF)$') - def set_source_voltage_mconnect(self, value) -> None: + @message(r":?SOUR:VOLT:MCON\s+(OFF|ON|0|1)$") + def set_source_voltage_mconnect(self, value: str) -> None: self.source_voltage_mconnect = {"0": False, "1": True, "OFF": False, "ON": True}[value] - @message(r'^:?SOUR:VOLT:MCON\?$') + @message(r":?SOUR:VOLT:MCON\?$") def get_source_voltage_mconnect(self) -> str: return {False: "0", True: "1"}[self.source_voltage_mconnect] # Measure - @message(r'^:?INIT$') + @message(r":?INIT(?::IMM)?$") def set_init(self) -> None: ... def _reading(self) -> float: @@ -158,22 +158,22 @@ def _reading(self) -> float: return random.uniform(volt_min, volt_max) return 0 - @message(r'^:?READ\?$') + @message(r":?READ\?$") def get_read(self) -> str: time.sleep(0.25) return format(self._reading(), "E") - @message(r'^:?FETC[H]?\?$') + @message(r":?FETC[H]?\?$") def get_fetch(self) -> str: return format(self._reading(), "E") - @message(r'^:?MEAS:CURR\?$') + @message(r":?MEAS:CURR\?$") def get_measure_current(self) -> str: time.sleep(random.uniform(0.25, 1.0)) vdc = random.uniform(0.000025, 0.0001) return format(vdc, "E") - @message(r'^:?MEAS:VOLT\?$') + @message(r":?MEAS:VOLT\?$") def get_measure_voltage(self) -> str: time.sleep(random.uniform(0.25, 1.0)) vdc = random.uniform(10.0, 100.0) @@ -181,74 +181,74 @@ def get_measure_voltage(self) -> str: # Average - @message(r'^:?SENS:(VOLT|CURR):AVER:TCON\?$') + @message(r":?SENS:(VOLT|CURR):AVER:TCON\?$") def get_sense_average_tcontrol(self, function: str) -> str: return format(self.sense_average_tcontrol[function], "E") - @message(r'^:?SENS:(VOLT|CURR):AVER:TCON (MOV|REP)$') + @message(r":?SENS:(VOLT|CURR):AVER:TCON\s+(MOV|REP)$") def set_sense_average_tcontrol(self, function: str, tcontrol: str) -> None: self.sense_average_tcontrol[function] = tcontrol - @message(r'^:?SENS:(VOLT|CURR):AVER:COUN[T]?\?$') + @message(r":?SENS:(VOLT|CURR):AVER:COUN[T]?\?$") def get_sense_average_count(self, function: str) -> str: return format(self.sense_average_count[function], "E") - @message(r'^:?SENS:(VOLT|CURR):AVER:COUN[T]? (\d+)$') + @message(r":?SENS:(VOLT|CURR):AVER:COUN[T]?\s+(\d+)$") def set_sense_average_count(self, function: str, count: str) -> None: self.sense_average_count[function] = int(count) - @message(r'^:?SENS:(VOLT|CURR):AVER:STAT[E]?\?$') + @message(r":?SENS:(VOLT|CURR):AVER:STAT[E]?\?$") def get_sense_average_state(self, function: str) -> str: return format(self.sense_average_state[function], "E") - @message(r'^:?SENS:(VOLT|CURR):AVER:STAT[E]? (OFF|ON|0|1)$') + @message(r":?SENS:(VOLT|CURR):AVER:STAT[E]?\s+(OFF|ON|0|1)$") def set_sense_average_state(self, function: str, state: str) -> None: self.sense_average_state[function] = {"OFF": 0, "ON": 1, "0": 0, "1": 1}[state] # Current range - @message(r'^(?::?SENS)?:CURR:RANG\?$') + @message(r"(?::?SENS)?:CURR:RANG\?$") def get_sense_current_range(self) -> str: return format(self.sense_current_range, "E") - @message(r'^(?::?SENS)?:CURR:RANG(?::UPP)?\s+(.+)$') + @message(r"(?::?SENS)?:CURR:RANG(?::UPP)?\s+(.+)$") def set_sense_current_range(self, value: str) -> None: try: self.sense_current_range = float(value) except Exception: self.error_queue.append(Error(102, "invalid header")) - @message(r'^(?::?SENS)?:CURR:RANG:AUTO\?$') + @message(r"(?::?SENS)?:CURR:RANG:AUTO\?$") def get_sense_current_range_auto(self) -> int: return self.sense_current_range_auto - @message(r'^(?::?SENS)?:CURR:RANG:AUTO\s+(OFF|ON|0|1)$') + @message(r"(?::?SENS)?:CURR:RANG:AUTO\s+(OFF|ON|0|1)$") def set_sense_current_range_auto(self, state: str) -> None: self.sense_current_range_auto = {"OFF": 0, "ON": 1, "0": 0, "1": 1}[state] - @message(r'^(?::?SENS)?:CURR:RANG:AUTO:ULIM\s+(.+)$') + @message(r"(?::?SENS)?:CURR:RANG:AUTO:ULIM\s+(.+)$") def set_sense_current_range_auto_ulimit(self, value: str) -> None: ... # TODO - @message(r'^(?::?SENS)?:CURR:RANG:AUTO:LLIM\s+(.+)$') + @message(r"(?::?SENS)?:CURR:RANG:AUTO:LLIM\s+(.+)$") def set_sense_current_range_auto_llimit(self, value: str) -> None: ... # TODO # NPLC (coupled commands) - @message(r'^(?::?SENS)?:(:?CURR|VOLT|RES|CHAR):NPLC\?$') + @message(r"(?::?SENS)?:(?:CURR|VOLT|RES|CHAR):NPLC\?$") def get_sense_nplc(self) -> str: return format(self.sense_nplc, "f") # TODO precision? - @message(r'^(?::?SENS)?:(:?CURR|VOLT|RES|CHAR):NPLC\s+(.+)$') + @message(r"(?::?SENS)?:(CURR|VOLT|RES|CHAR):NPLC\s+(.+)$") def set_sense_nplc(self, mode: str, value: str) -> None: try: self.sense_nplc = max(0.01, min(10.0, float(value))) except Exception: self.error_queue.append(Error(102, "invalid header")) - @message(r'^(.*)$') - def unknown_message(self, request) -> None: + @message(r".*") + def unknown_message(self) -> None: self.error_queue.append(Error(101, "malformed command")) diff --git a/src/comet/emulator/keithley/k707b.py b/src/comet/emulator/keithley/k707b.py index cec9eb1..7f5e393 100644 --- a/src/comet/emulator/keithley/k707b.py +++ b/src/comet/emulator/keithley/k707b.py @@ -13,23 +13,23 @@ def __init__(self) -> None: self.error_queue: list[Error] = [] self.closed_channels: set[str] = set() - @message(r'^reset\(\)|\*RST$') + @message(r"reset\(\)|\*RST$") def set_reset(self) -> None: self.error_queue.clear() - @message(r'^clear\(\)|\*CLS$') + @message(r"clear\(\)|\*CLS$") def set_clear(self) -> None: self.error_queue.clear() - @message(r'errorqueue\.clear\(\)') + @message(r"errorqueue\.clear\(\)$") def set_errorqueue_clear(self) -> None: self.error_queue.clear() - @message(tsp_print(r'errorqueue\.count')) + @message(tsp_print(r"errorqueue\.count")) def get_errorqueue_count(self) -> str: return format(len(self.error_queue), "d") - @message(tsp_print(r'errorqueue\.next\(\)')) + @message(tsp_print(r"errorqueue\.next\(\)")) def get_errorqueue_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -37,29 +37,29 @@ def get_errorqueue_next(self) -> str: error = Error(0, "Queue is Empty") return f"{error.code}\t\"{error.message}\"\t0\t0" - @message(tsp_print(r'channel\.getclose\(([^\)]+)\)')) - def get_channel_getclose(self, channels) -> str: + @message(tsp_print(r"channel\.getclose\(([^\)]+)\)")) + def get_channel_getclose(self, channels: str) -> str: if not self.closed_channels: return "nil" return ";".join(sorted(self.closed_channels)) - @message(r'channel\.close\(([^\)]+)\)') - def set_channel_close(self, channels) -> None: - channels = channels.strip('"').split(',') - self.closed_channels.update(channels) + @message(r"channel\.close\(([^\)]+)\)$") + def set_channel_close(self, channels: str) -> None: + channels_ = channels.strip('"').split(',') + self.closed_channels.update(channels_) - @message(r'channel\.open\(([^\)]+)\)') - def set_channel_open(self, channels) -> None: - channels = channels.strip('"').split(',') - if "allslots" in channels: + @message(r"channel\.open\(([^\)]+)\)$") + def set_channel_open(self, channels: str) -> None: + channels_ = channels.strip('"').split(',') + if "allslots" in channels_: self.closed_channels.clear() else: for channel in channels: if channel in self.closed_channels: self.closed_channels.remove(channel) - @message(r'^(.*)$') - def unknown_message(self, v) -> None: + @message(r".*") + def unknown_message(self) -> None: self.error_queue.append(Error(101, "malformed command")) diff --git a/src/comet/emulator/keysight/e4980a.py b/src/comet/emulator/keysight/e4980a.py index ca656cd..5c6ec7c 100644 --- a/src/comet/emulator/keysight/e4980a.py +++ b/src/comet/emulator/keysight/e4980a.py @@ -23,7 +23,7 @@ def __init__(self) -> None: self.bias_voltage_level: float = 0. self.bias_state: bool = False - @message(r'\*RST$') + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() self.function_impedance_type = "CPD" @@ -35,68 +35,68 @@ def set_rst(self) -> None: self.bias_voltage_level = 0. self.bias_state = False - @message(r'\*CLS$') + @message(r"\*CLS$") def set_cls(self) -> None: self.error_queue.clear() - @message(r':?SYST:ERR(?::NEXT)?\?$') + @message(r":?SYST:ERR(?::NEXT)?\?$") def get_system_error_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) return f"{error.code:+d},\"{error.message}\"" return '+0,"No error"' - @message(r'^:?FUNC:IMP:TYPE\?$') + @message(r":?FUNC:IMP:TYPE\?$") def get_function_impedance_type(self) -> str: return self.function_impedance_type - @message(r'^:?FUNC:IMP:TYPE\s+(CPD|CPRP|CSRS|LSRS)$') + @message(r":?FUNC:IMP:TYPE\s+(CPD|CPRP|CSRS|LSRS)$") def set_function_impedance_type(self, type) -> None: # TODO self.function_impedance_type = type - @message(r'^:?CORR:OPEN:STAT\?$') + @message(r":?CORR:OPEN:STAT\?$") def get_correction_open_state(self) -> str: return format(self.correction_open_state, "d") - @message(r'^:?CORR:OPEN:STAT\s+(OFF|ON|0|1)$') + @message(r":?CORR:OPEN:STAT\s+(OFF|ON|0|1)$") def set_correction_open_state(self, state: str) -> None: self.correction_open_state = {"0": False, "1": True, "OFF": False, "ON": True}[state] - @message(r'^:?CORR:OPEN$') + @message(r":?CORR:OPEN$") def get_correction_open(self) -> None: delay = self.options.get("correction_open_delay", self.CORRECTION_OPEN_DELAY) time.sleep(delay) - @message(r'^:?CORR:USE\?$') + @message(r":?CORR:USE\?$") def get_correction_use(self) -> str: return format(self.correction_use, "+d") - @message(r'^:?CORR:METH\?$') + @message(r":?CORR:METH\?$") def get_correction_method(self) -> str: return self.correction_method - @message(r'^:?CORR:METH\s+(SING|MULT)$') + @message(r":?CORR:METH\s+(SING|MULT)$") def set_correction_method_single(self, method: str) -> None: self.correction_method = method - @message(r'^:?CORR:USE:CHAN\?$') + @message(r":?CORR:USE:CHAN\?$") def get_correction_channel(self) -> str: return format(self.correction_channel, "d") - @message(r'^:?CORR:USE:CHAN\s+(\d+)$') + @message(r":?CORR:USE:CHAN\s+(\d+)$") def set_correction_channel(self, value): self.correction_channel = int(value) - @message(r'^:?CORR:LENG\?$') + @message(r":?CORR:LENG\?$") def get_correction_length(self) -> str: return format(self.correction_length, "+d") - @message(r'^:?CORR:LENG\s+(\d+)$') + @message(r":?CORR:LENG\s+(\d+)$") def set_correction_length(self, length: str) -> None: self.correction_length = int(length) - @message(r'^:?FETC[H]?(?:(?::IMP)?:FORM)?\?$') + @message(r":?FETC[H]?(?:(?::IMP)?:FORM)?\?$") def get_fetch(self) -> str: # TODO cp_min = float(self.options.get("cp.min", 2.5e-10)) @@ -107,32 +107,32 @@ def get_fetch(self) -> str: sec = random.uniform(rp_min, rp_max) return '{:E},{:E},{:+d}'.format(prim, sec, 0) - @message(r'^:?BIAS:POL:CURR(\::LEV)?\?$') + @message(r":?BIAS:POL:CURR(\::LEV)?\?$") def get_bias_polarity_current_level(self) -> str: return format(random.random() / 1000., "E") - @message(r'^:?BIAS:POL:VOLT(\::LEV)?\?$') + @message(r":?BIAS:POL:VOLT(\::LEV)?\?$") def get_bias_polarity_voltage_level(self) -> str: return format(random.random() / 100., "E") - @message(r'^:?BIAS:VOLT(?::LEV)?\?$') + @message(r":?BIAS:VOLT(?::LEV)?\?$") def get_bias_voltage_level(self) -> str: return format(self.bias_voltage_level, "E") - @message(r'^:?BIAS:VOLT(?::LEV)?\s+(.+)$') - def set_bias_voltage_level(self, value) -> None: + @message(r":?BIAS:VOLT(?::LEV)?\s+(.+)$") + def set_bias_voltage_level(self, value: str) -> None: self.bias_voltage_level = float(value) - @message(r'^:?BIAS:STAT\?$') + @message(r":?BIAS:STAT\?$") def get_bias_state(self) -> str: return format(self.bias_state, "d") - @message(r'^:?BIAS:STAT\s+(0|1|ON|OFF)$') - def set_bias_state(self, value) -> None: + @message(r":?BIAS:STAT\s+(OFF|ON|0|1)$") + def set_bias_state(self, value: str) -> None: self.bias_state = {"0": False, "1": True, "OFF": False, "ON": True}[value] - @message(r"(.*)$") - def undefined_header(self, _): + @message(r".*") + def undefined_header(self): self.error_queue.append(Error(-113, "Undefined header")) diff --git a/src/comet/emulator/marzhauser/tango.py b/src/comet/emulator/marzhauser/tango.py index e4530de..17ac6aa 100644 --- a/src/comet/emulator/marzhauser/tango.py +++ b/src/comet/emulator/marzhauser/tango.py @@ -22,25 +22,25 @@ def __init__(self) -> None: # Controller informations - @message(r'^\??version$') + @message(r"\??version$") def get_version(self) -> str: return format(self.options.get("version", self.VERSION)) - @message(r'^\?autostatus$') + @message(r"\?autostatus$") def get_autostatus(self) -> str: return format(self.autostatus, "d") - @message(r'^\!autostatus (0|1)$') + @message(r"\!autostatus (0|1)$") def set_autostatus(self, value) -> None: self.autostatus = {"0": False, "1": True}[value] - @message(r'^\??err$') + @message(r"\??err$") def get_error(self) -> str: return "0" # Calibration - @message(r'^\!?cal$') + @message(r"\!?cal$") def set_cal(self) -> Optional[str]: self.calst["x"] |= 0x1 self.calst["y"] |= 0x1 @@ -49,14 +49,14 @@ def set_cal(self) -> Optional[str]: return "AAA-." return None - @message(r'^\!?cal (x|y|z)$') + @message(r"\!?cal\s+(x|y|z)$") def set_cal_xyz(self, axes) -> Optional[str]: self.calst[axes] |= 0x1 if self.autostatus: return "A" return None - @message(r'^\!?rm$') + @message(r"\!?rm$") def set_rm(self) -> Optional[str]: self.calst["x"] |= 0x2 self.calst["y"] |= 0x2 @@ -65,40 +65,40 @@ def set_rm(self) -> Optional[str]: return "DDD-." return None - @message(r'^\!?rm (x|y|z)$') + @message(r"\!?rm\s+(x|y|z)$") def set_rm_xyz(self, axes) -> Optional[str]: self.calst[axes] &= 0x1 if self.autostatus: return "D" return None - @message(r'^\??calst$') + @message(r"\??calst$") def get_calst(self) -> str: x = self.calst.get("x", 0) y = self.calst.get("y", 0) z = self.calst.get("z", 0) return f"{x:d} {y:d} {z:d}" - @message(r'^\?calst (x|y|z)$') + @message(r"\?calst\s+(x|y|z)$") def get_calst_xyz(self, axis) -> str: value = self.calst.get(axis, 0) return f"{value:d}" # Positioning - @message(r'^\?pos$') + @message(r"\?pos$") def get_pos(self) -> str: x = self.position.get("x", 0) y = self.position.get("y", 0) z = self.position.get("z", 0) return f"{x:.3f} {y:.3f} {z:.3f}" - @message(r'^\?pos (x|y|z)$') + @message(r"\?pos\s+(x|y|z)$") def get_pos_xyz(self, axis) -> str: value = self.position.get(axis) return f"{value:.3f}" - @message(r'^\!?moa ([^\sxyza]+) ([^\s]+) ([^\s]+)$') + @message(r"\!?moa\s+([^\sxyza]+)\s+([^\s]+)\s+([^\s]+)$") def set_move_absolute(self, x, y, z) -> Optional[str]: self.position["x"] = float(x) self.position["y"] = float(y) @@ -107,14 +107,14 @@ def set_move_absolute(self, x, y, z) -> Optional[str]: return "@@@-." return None - @message(r'^\!?moa (x|y|z) ([^\s]+)$') + @message(r"\!?moa\s+(x|y|z)\s+([^\s]+)$") def set_move_absolute_xyz(self, axis, value) -> Optional[str]: self.position[axis] = float(value) if self.autostatus: return "@@@-." return None - @message(r'^\!?mor ([^\sxyza]+) ([^\s]+) ([^\s]+)$') + @message(r"\!?mor\s+([^\sxyza]+)\s+([^\s]+)\s+([^\s]+)$") def set_move_relative(self, x, y, z) -> Optional[str]: self.position["x"] = float(x) self.position["y"] = float(y) @@ -123,51 +123,51 @@ def set_move_relative(self, x, y, z) -> Optional[str]: return "@@@-." return None - @message(r'^\!?mor (x|y|z) ([^\s]+)$') + @message(r"\!?mor\s+(x|y|z)\s+([^\s]+)$") def set_move_relative_xyz(self, axis, value) -> Optional[str]: self.position[axis] = float(value) if self.autostatus: return "@@@-." return None - @message(r'^\?statusaxis$') + @message(r"\?statusaxis$") def get_statusaxis(self) -> str: x = self.statusaxis.get("x", "-") y = self.statusaxis.get("y", "-") z = self.statusaxis.get("z", "-") return f"{x}{y}{z}-.-" - @message(r'^\?statusaxis (x|y|z)$') + @message(r"\?statusaxis\s+(x|y|z)$") def get_statusaxis_xyz(self, axis) -> str: value = self.statusaxis.get(axis, "-") return f"{value}" - @message(r'^\?vel$') + @message(r"\?vel$") def get_vel(self) -> str: x = self.velocity.get("x") y = self.velocity.get("y") z = self.velocity.get("z") return f"{x:.3f} {y:.3f} {z:.3f}" - @message(r'^\?vel (x|y|z)$') + @message(r"\?vels\+(x|y|z)$") def get_vel_xyz(self, axis) -> str: value = self.velocity.get(axis) return f"{value:.3f}" - @message(r'^!vel (x|y|z) ([^\s]+)$') + @message(r"!vel\s+(x|y|z)\s+([^\s]+)$") def set_vel_xyz(self, axis, value) -> str: self.velocity[axis]= float(value) return "@@@-." # System configuration - @message(r'^save$') + @message(r"save$") def action_save(self) -> None: ... - @message(r'^restore$') + @message(r"restore$") def action_restore(self) -> None: ... - @message(r'^reset$') + @message(r"reset$") def action_reset(self) -> None: ... diff --git a/src/comet/emulator/nkt_photonics/pilas.py b/src/comet/emulator/nkt_photonics/pilas.py index 0a14e26..af88eaf 100644 --- a/src/comet/emulator/nkt_photonics/pilas.py +++ b/src/comet/emulator/nkt_photonics/pilas.py @@ -29,51 +29,51 @@ def __init__(self) -> None: self.laser_head_temperature: float = 25.0 self.laser_diode_temperature: bool = True - @message(r"^version\?$") + @message(r"version\?$") def get_version(self) -> str: return self.IDENTITY - @message(r"^ld\?$") + @message(r"ld\?$") def get_output(self) -> str: return "pulsed laser emission: " + ("on" if self.output else "off") - @message(r"^ld=(0|1)$") + @message(r"ld=(0|1)$") def set_output(self, output: str) -> str: self.output = bool(int(output)) return "done" - @message(r"^tm\?$") + @message(r"tm\?$") def get_tune_mode(self) -> str: return "tune mode:\t" + ("auto" if self.tune_mode else "manual") - @message(r"^tm=(0|1)$") + @message(r"tm=(0|1)$") def set_tune_mode(self, tune_mode: str) -> str: self.tune_mode = bool(int(tune_mode)) return "done" - @message(r"^tune\?$") + @message(r"tune\?$") def get_tune(self) -> str: return f"tune value:\t\t {self.tune:.2f} %" - @message(r"^tune=(\d{1,4})$") + @message(r"tune=(\d{1,4})$") def set_tune(self, tune: int) -> str: self.tune = int(tune) / 10 return "done" - @message(r"^f\?$") + @message(r"f\?$") def get_frequency(self) -> str: return f"int. frequency:\t {self.frequency} Hz" - @message(r"^f=(\d+)$") + @message(r"f=(\d+)$") def set_frequency(self, frequency: int) -> str: self.frequency = int(frequency) return "done" - @message(r"^lht\?$") + @message(r"lht\?$") def get_laser_head_temperature(self) -> TextResponse: return TextResponse(f"laser head temp.:\t {self.laser_head_temperature} °C", encoding="latin-1") - @message(r"^ldtemp\?$") + @message(r"ldtemp\?$") def get_laser_diode_temperature(self) -> str: return "LD temp.:\t\t" + ("good" if self.laser_diode_temperature else "bad") diff --git a/src/comet/emulator/photonic/f3000.py b/src/comet/emulator/photonic/f3000.py index 7ab8eec..689cd36 100644 --- a/src/comet/emulator/photonic/f3000.py +++ b/src/comet/emulator/photonic/f3000.py @@ -15,24 +15,24 @@ def __init__(self) -> None: self.current_brightness: int = 50 self.light_enabled: bool = False - @message(r"^V\?$") + @message(r"V\?$") def get_version(self) -> str: return self.IDENTITY - @message(r"^B(\d{1,3})$") + @message(r"B(\d{1,3})$") def set_brightness(self, brightness: int) -> None: brightness = max(0, min(int(brightness), 100)) self.current_brightness = int(brightness) - @message(r"^B\?$") + @message(r"B\?$") def get_brightness(self) -> str: return "B" + str(self.current_brightness) - @message(r"^S\?$") + @message(r"S\?$") def get_light_enabled(self) -> str: return "S0" if self.light_enabled else "S1" - @message(r"^S(0|1)$") + @message(r"S(0|1)$") def set_light_enabled(self, light_enabled: str) -> None: self.light_enabled = not bool(int(light_enabled)) diff --git a/src/comet/emulator/rohde_schwarz/nge100.py b/src/comet/emulator/rohde_schwarz/nge100.py index 56cf278..59e7fd1 100644 --- a/src/comet/emulator/rohde_schwarz/nge100.py +++ b/src/comet/emulator/rohde_schwarz/nge100.py @@ -39,19 +39,19 @@ def get_current(self) -> float: current_from_current_limit = self.current_limits[self.selected_channel] return min(current_from_voltage_level, current_from_current_limit) - @message(r"^\*IDN\?$") + @message(r"\*IDN\?$") def identify(self) -> str: return self.IDENTITY - @message(r"^INST(?:rument)? (\d)$") + @message(r"INST(?:rument)?\s+(\d)$") def set_channel(self, channel: int) -> None: self.selected_channel = int(channel) - 1 - @message(r"^INST(?:rument)?\?$") + @message(r"INST(?:rument)?\?$") def get_channel(self) -> str: return str(self.selected_channel + 1) - @message(r"^OUTP(?:ut)? (\d)$") + @message(r"^OUTP(?:ut)?\s+(\d)$") def set_enabled(self, enabled: int) -> None: self.enabled_channels[self.selected_channel] = bool(int(enabled)) @@ -59,7 +59,7 @@ def set_enabled(self, enabled: int) -> None: def get_enabled(self) -> str: return str(int(self.enabled_channels[self.selected_channel])) - @message(r"(?:SOUR(?:ce)?:)?VOLT(?:age)?(?::LEV(?:el)?)?(?::IMM(?:ediate)?)?(?::AMPL(?:itude)?)? (\d+\.?\d*)$") + @message(r"(?:SOUR(?:ce)?:)?VOLT(?:age)?(?::LEV(?:el)?)?(?::IMM(?:ediate)?)?(?::AMPL(?:itude)?)?\s+(.+)$") def set_voltage_level(self, voltage_level: float) -> None: voltage_level = min(max(0, float(voltage_level)), 32) self.voltage_levels[self.selected_channel] = voltage_level @@ -68,7 +68,7 @@ def set_voltage_level(self, voltage_level: float) -> None: def get_voltage_level(self) -> str: return str(self.voltage_levels[self.selected_channel]) - @message(r"(?:SOUR(?:ce)?:)?CURR(?:ent)?(?::LEV(?:el)?)?(?::IMM(?:ediate)?)?(?::AMPL(?:itude)?)? (\d+\.?\d*)$") + @message(r"(?:SOUR(?:ce)?:)?CURR(?:ent)?(?::LEV(?:el)?)?(?::IMM(?:ediate)?)?(?::AMPL(?:itude)?)?\s+(.+)$") def set_current_limit(self, current_limit: float) -> None: current_limit = min(max(0, float(current_limit)), 3) self.current_limits[self.selected_channel] = current_limit diff --git a/src/comet/emulator/rohde_schwarz/rtp164.py b/src/comet/emulator/rohde_schwarz/rtp164.py index b07ea7d..9f7d8bf 100644 --- a/src/comet/emulator/rohde_schwarz/rtp164.py +++ b/src/comet/emulator/rohde_schwarz/rtp164.py @@ -80,8 +80,8 @@ def get_channel_waveform_data(self, channel, waveform) -> BinaryResponse: big_endian = self.format_border == "MSBF" return BinaryResponse.pack_real32(y, big_endian=big_endian) - @message(r"(.*)$") - def undefined_header(self, command) -> None: + @message(r".*") + def undefined_header(self) -> None: self.error_queue.append(SCPIError(-113, "Undefined header")) diff --git a/src/comet/emulator/rohde_schwarz/sma100b.py b/src/comet/emulator/rohde_schwarz/sma100b.py index 8e357dc..c53473d 100644 --- a/src/comet/emulator/rohde_schwarz/sma100b.py +++ b/src/comet/emulator/rohde_schwarz/sma100b.py @@ -21,20 +21,20 @@ def __init__(self) -> None: self.power: float = 0 self.output: bool = False - @message(r"^\*IDN\?$") + @message(r"\*IDN\?$") def identify(self) -> str: return self.IDENTITY - @message(r"^\*RST$") + @message(r"\*RST$") def set_reset(self) -> None: self.average_count = 100 self.wavelength = 370 - @message(r"^\*CLS$") + @message(r"\*CLS$") def set_clear(self) -> None: self.error_queue.clear() - @message(r"^:?SYST(?:em)?:ERR(?::NEXT)?\?$") + @message(r":?SYST(?:em)?:ERR(?::NEXT)?\?$") def get_system_error_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -42,19 +42,19 @@ def get_system_error_next(self) -> str: error = Error(0, "no error") return f'{error.code}, "{error.message}"' - @message(r"^(?:SOUR(?:ce)?1)?:FREQ(?:uency)?:MODE\?$") + @message(r"(?:SOUR(?:ce)?1)?:FREQ(?:uency)?:MODE\?$") def get_frequency_mode(self) -> str: return self.frequency_mode - @message(r"^(?:SOUR(?:ce)?1)?:FREQ(?:uency)?:MODE (\w+)$") - def set_frequency_mode(self, mode) -> None: + @message(r"(?:SOUR(?:ce)?1)?:FREQ(?:uency)?:MODE\s+(\w+)$") + def set_frequency_mode(self, mode: str) -> None: self.frequency_mode = mode - @message(r"^(?:SOUR(?:ce)?1)?:FREQuency:(?:CW|FIX(?:ed)?)\?$") + @message(r"(?:SOUR(?:ce)?1)?:FREQuency:(?:CW|FIX(?:ed)?)\?$") def get_frequency(self) -> float: return self.frequency - @message(r"^(?:SOUR(?:ce)?1)?:FREQuency:(?:CW|FIX(?:ed)?) ([\d.]+(?:[eE][+-]?\d+)?)$") + @message(r"(?:SOUR(?:ce)?1)?:FREQuency:(?:CW|FIX(?:ed)?) ([\d.]+(?:[eE][+-]?\d+)?)$") def set_frequency(self, frequency) -> None: frequency = float(frequency) if frequency < 8e3 or frequency > 12.75e9: @@ -62,11 +62,11 @@ def set_frequency(self, frequency) -> None: else: self.frequency = frequency - @message(r"^(?:SOUR(?:ce)?1)?:POW(?:er)?:POW(?:er)?\?$") + @message(r"(?:SOUR(?:ce)?1)?:POW(?:er)?:POW(?:er)?\?$") def get_power(self) -> float: return self.power - @message(r"^(?:SOUR(?:ce)?1)?:POW(?:er)?:POW(?:er)? ([+-]?[\d.]+(?:[eE][+-]?\d+)?)$") + @message(r"(?:SOUR(?:ce)?1)?:POW(?:er)?:POW(?:er)? ([+-]?[\d.]+(?:[eE][+-]?\d+)?)$") def set_power(self, power) -> None: power = float(power) if power < -145 or power > 40: @@ -74,11 +74,11 @@ def set_power(self, power) -> None: else: self.power = power - @message(r"^(?:SOUR(?:ce)?1:)?OUTP(?:ut)?:STAT(?:e)?\?$") + @message(r"(?:SOUR(?:ce)?1:)?OUTP(?:ut)?:STAT(?:e)?\?$") def get_output(self) -> str: return "1" if self.output else "0" - @message(r"^(?:SOUR(?:ce)?1:)?OUTP(?:ut)?:STAT(?:e)? (ON|OFF)$") + @message(r"(?:SOUR(?:ce)?1:)?OUTP(?:ut)?:STAT(?:e)?\s+(ON|OFF)$") def set_output(self, state) -> None: self.output = True if state == "ON" else False diff --git a/src/comet/emulator/thorlabs/pm100.py b/src/comet/emulator/thorlabs/pm100.py index c62f6bc..84555fe 100644 --- a/src/comet/emulator/thorlabs/pm100.py +++ b/src/comet/emulator/thorlabs/pm100.py @@ -20,24 +20,24 @@ def __init__(self) -> None: self.average_count: int = 100 self.wavelength: int = 370 - @message(r"^\*IDN\?$") + @message(r"\*IDN\?$") def identify(self) -> str: return self.IDENTITY - @message(r"^\*RST$") + @message(r"\*RST$") def set_reset(self) -> None: self.average_count = 100 self.wavelength = 370 - @message(r"^\*CLS$") + @message(r"\*CLS$") def set_clear(self) -> None: self.error_queue.clear() - @message(r"^\*OPC\?$") + @message(r"\*OPC\?$") def get_opc(self) -> int: return 1 - @message(r"^:?SYST:ERR(?::NEXT)?\?$") + @message(r":?SYST:ERR(?::NEXT)?\?$") def get_system_error_next(self) -> str: if self.error_queue: error = self.error_queue.pop(0) @@ -45,23 +45,23 @@ def get_system_error_next(self) -> str: error = Error(0, "no error") return f'{error.code}, "{error.message}"' - @message(r"^(?:SENS(?:e)?)?:AVER(?:age)?:COUN(?:t)?\?$") + @message(r"(?:SENS(?:e)?)?:AVER(?:age)?:COUN(?:t)?\?$") def get_average_count(self) -> int: return self.average_count - @message(r"^(?:SENS(?:e)?)?:AVER(?:age)?:COUN(?:t)? (\d+)$") + @message(r"(?:SENS(?:e)?)?:AVER(?:age)?:COUN(?:t)?\s+(\d+)$") def set_average_count(self, average_count) -> None: self.average_count = average_count - @message(r"^(?:SENS(?:e)?)?:CORR(?:ection)?:WAV(?:elength)?\?$") + @message(r"(?:SENS(?:e)?)?:CORR(?:ection)?:WAV(?:elength)?\?$") def get_wavelength(self) -> int: return self.wavelength - @message(r"^(?:SENS(?:e)?)?:CORR(?:ection)?:WAV(?:elength)? (\d+)$") + @message(r"(?:SENS(?:e)?)?:CORR(?:ection)?:WAV(?:elength)?\s+(\d+)$") def set_wavelength(self, wavelength) -> None: self.wavelength = wavelength - @message(r"^MEAS(?:ure)?(?::SCAL(?:ar)?)?(?::POW(?:er)?)?") + @message(r"MEAS(?:ure)?(?::SCAL(?:ar)?)?(?::POW(?:er)?)?$") def measure_power(self) -> str: power = random.uniform(1e-9, 2e-9) return format(power, "E")