Skip to content

Normalize emulator routes #120

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
arnobaer merged 2 commits into from
May 4, 2026
Merged

Normalize emulator routes #120

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3e78d09
refactor(routes): normalized emulator message routes
arnobaer May 4, 2026
6a0908a
refactor(routes): fixed regex routes
arnobaer May 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 21 additions & 21 deletions src/comet/emulator/cts/itc.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}"
Expand Down
24 changes: 12 additions & 12 deletions src/comet/emulator/ers/ac3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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"

Expand Down
48 changes: 24 additions & 24 deletions src/comet/emulator/hephy/brandbox.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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):
Expand All @@ -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":
Expand All @@ -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":
Expand All @@ -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":
Expand All @@ -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)
Expand All @@ -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

Expand Down
Loading
Loading