diff --git a/LICENSE b/LICENSE index b3adf73..f84d4c8 100644 --- a/LICENSE +++ b/LICENSE @@ -25,4 +25,3 @@ SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - diff --git a/README.md b/README.md index f197118..482b044 100644 --- a/README.md +++ b/README.md @@ -26,4 +26,4 @@ -Development of new IOC for pulling and reading real time data streaming from Hiden RGA \ No newline at end of file +Development of new IOC for pulling and reading real time data streaming from Hiden RGA diff --git a/pyproject.toml b/pyproject.toml index bfc9d7a..0bc67a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -125,3 +125,4 @@ isort.required-imports = ["from __future__ import annotations"] [tool.ruff.lint.per-file-ignores] "tests/**" = ["T20"] "noxfile.py" = ["T20"] +"src/hiden/cap2.py" = ["ARG002"] # caproto putter signature requires `instance` diff --git a/src/hiden/cap2.py b/src/hiden/cap2.py index 40aa2b8..0a39752 100644 --- a/src/hiden/cap2.py +++ b/src/hiden/cap2.py @@ -1,197 +1,536 @@ +""" +Caproto IOC for Hiden RGA via MASsoft sockets (merged). + +Combines the working version's reliability patterns (eager connect, polling +fallback with reconnect retry) with the rewrite's protocol-correct hot-link +threads and diagnostic PVs. + +Provides both working-version PVs (RunExp, AbortExp, CloseExp) and rewrite +PVs (Go, Abort, Close) for backward compatibility. +""" + +from __future__ import annotations + +import argparse import asyncio +import contextlib import logging -from pathlib import PureWindowsPath +import sys +import time + +from caproto.server import PVGroup, ioc_arg_parser, pvproperty, run + +from .massoft_client import MASsoftClient, MASsoftTimeout, load_runtime_config + +if not logging.getLogger().handlers: + logging.basicConfig(level=logging.INFO) +logging.getLogger("caproto").propagate = False +logging.getLogger("caproto.ctx").propagate = False +LOG = logging.getLogger(__name__) -from caproto import ChannelType -from caproto.server import PVGroup, pvproperty, ioc_arg_parser, run +# Runtime config for IOC defaults +_RUNTIME_CFG = load_runtime_config() +_IOC_CFG = _RUNTIME_CFG.get("ioc", {}) if isinstance(_RUNTIME_CFG, dict) else {} +if not isinstance(_IOC_CFG, dict): + _IOC_CFG = {} -from .massoft_client import MASsoftClient, EXPERIMENT_DIRECTORY, MOST_RECENT_FILE -logging.basicConfig(level=logging.INFO) +def _ioc_default(name: str, default): + val = _IOC_CFG.get(name, default) + return default if val is None else val + class RGAIOC(PVGroup): - # — Control / Configuration PVs — + # ----------------------------------------------------------------- + # Control PVs (working-version names) + # ----------------------------------------------------------------- + open_exp = pvproperty( - name='XF:08IDB-SE{{RGA:1}}:OpenExp', - value=0, dtype=int, - doc='Write 1 to open the experiment file' + name="XF:08IDB-SE{{RGA:1}}:OpenExp", + value=0, + dtype=int, + doc="Write 1 to connect + open/associate the experiment + fetch legends", ) - experiment_name = pvproperty( - name='XF:08IDB-SE{{RGA:1}}:ExpName', - value='file1.exp', - dtype=ChannelType.STRING, max_length=64, - doc='Name of the .exp file in MASsoft folder' + experiment = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:ExpName", + value=str(_ioc_default("default_experiment", "file56.exp")), + dtype=str, + max_length=256, + doc="Experiment file name (relative to MASsoft experiment directory) or full path", ) acquire = pvproperty( - name='XF:08IDB-SE{{RGA:1}}:Acquire', - value=0, dtype=int, - doc='Start/stop the acquisition loop', + name="XF:08IDB-SE{{RGA:1}}:Acquire", + value=0, + dtype=int, + doc="Start/stop the acquisition loop", ) run_exp = pvproperty( - name='XF:08IDB-SE{{RGA:1}}:RunExp', - value=0, dtype=int, - doc='Write 1 to start the experiment' + name="XF:08IDB-SE{{RGA:1}}:RunExp", + value=0, + dtype=int, + doc="Write 1 to start the experiment", ) abort_exp = pvproperty( - name='XF:08IDB-SE{{RGA:1}}:AbortExp', - value=0, dtype=int, - doc='Write 1 to abort the running experiment' + name="XF:08IDB-SE{{RGA:1}}:AbortExp", + value=0, + dtype=int, + doc="Write 1 to abort the running experiment", ) close_exp = pvproperty( - name='XF:08IDB-SE{{RGA:1}}:CloseExp', - value=0, dtype=int, - doc='Write 1 to close the experiment file' + name="XF:08IDB-SE{{RGA:1}}:CloseExp", + value=0, + dtype=int, + doc="Write 1 to close the experiment file", + ) + + go = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:Go", + value=0, + dtype=int, + doc="Momentary: -xGo (run experiment). Uses GoOD/GoOT and GoFilename.", + ) + + abort = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:Abort", + value=0, + dtype=int, + doc="Momentary: -xAbort and wait for Stopped* status.", + ) + + close = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:Close", + value=0, + dtype=int, + doc="Momentary: safe close (abort->wait for Stopped*->-xClose).", + ) + + # ----------------------------------------------------------------- + # Configuration PVs (from rewrite) + # ----------------------------------------------------------------- + + view = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:View", + value=max(1, int(_ioc_default("default_view", 1))), + dtype=int, + doc="MASsoft view number (used for -lStatus/-lData/-lLegends)", + ) + + go_od = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:GoOD", + value=1 if int(_ioc_default("default_go_od", 1)) else 0, + dtype=int, + doc="When 1, include 'd' in -O flags (create date directory).", + ) + + go_ot = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:GoOT", + value=1 if int(_ioc_default("default_go_ot", 1)) else 0, + dtype=int, + doc="When 1, include 't' in -O flags (time-based filename).", + ) + + go_filename = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:GoFilename", + value=str(_ioc_default("default_go_filename", "")), + dtype=str, + max_length=256, + doc="Optional filename argument to -xGo. If blank, MASsoft uses its defaults.", + ) + + # ----------------------------------------------------------------- + # Diagnostic PVs (from rewrite) + # ----------------------------------------------------------------- + + connected = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:Connected", + value=0, + dtype=int, + doc="1 if client has connected sockets and associated a file; 0 otherwise.", ) - # — MID-I & Mass PVs (1–10) — + status = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:Status", + value="Disconnected", + dtype=str, + max_length=32, + doc="Last MASsoft status received from the status hot-link.", + ) + + last_error = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:LastError", + value="", + dtype=str, + max_length=256, + doc="Last client-side socket/protocol error for diagnostics.", + ) + + data_age = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:DataAge", + value=-1.0, + dtype=float, + doc="Seconds since last data row received. -1 means unknown.", + ) + + status_age = pvproperty( + name="XF:08IDB-SE{{RGA:1}}:StatusAge", + value=-1.0, + dtype=float, + doc="Seconds since last status update received. -1 means unknown.", + ) + + # ----------------------------------------------------------------- + # MID-I readbacks (1-10) + # ----------------------------------------------------------------- + for idx in range(1, 11): - locals()[f'mid{idx}'] = pvproperty( - name=f'XF:08IDB-SE{{{{RGA:1}}}}P:MID{idx}-I', - value=0.0, dtype=float, - doc=f'RGA reading for MID{idx}', + locals()[f"mid{idx}"] = pvproperty( + name=f"XF:08IDB-SE{{{{RGA:1}}}}P:MID{idx}-I", + value=0.0, + dtype=float, + doc=f"RGA MID{idx} intensity", ) - locals()[f'mass{idx}'] = pvproperty( - name=f'XF:08IDB-VA{{{{RGA:1}}}}Mass:MID{idx}', - value=0.0, dtype=float, - doc=f'RGA mass for MID{idx}', + del idx + + # ----------------------------------------------------------------- + # Mass readbacks (1-10) + # ----------------------------------------------------------------- + + for idx in range(1, 11): + locals()[f"mass{idx}"] = pvproperty( + name=f"XF:08IDB-VA{{{{RGA:1}}}}Mass:MID{idx}", + value=0.0, + dtype=float, + doc=f"MID{idx} mass value", ) del idx - def __init__(self, *args, **kwargs): + # ----------------------------------------------------------------- + # Constructor (eager connect) + # ----------------------------------------------------------------- + + def __init__(self, *args, mas_host=None, mas_port=None, **kwargs): super().__init__(*args, **kwargs) - self.client = MASsoftClient() - self.client.initialize() - self._running = False - self._task = None - self._mass_vals = [] # store legends + self.client = MASsoftClient(host=mas_host, port=mas_port) + self.client.connect() + self._running = False + self._acq_task: asyncio.Task | None = None + self._links_started = False + self._mass_vals: list[float] = [] + self._last_pub_status_ts = 0.0 + self._last_pub_row_ts = 0.0 + + # ----------------------------------------------------------------- + # PV put handlers + # ----------------------------------------------------------------- @open_exp.putter async def open_exp(self, instance, value): - want = bool(int(value)) - if want: - fn = self.experiment_name.value - if isinstance(fn, (list, tuple)): - fn = fn[0] - loop = asyncio.get_running_loop() - await loop.run_in_executor(None, - self.client.open_experiment_commands, fn + """Connect if needed, open/associate experiment, fetch legends.""" + if not int(value): + return value + + fname = self.experiment.value + if isinstance(fname, (list, tuple)): + fname = fname[0] + fname = (fname or "").strip() or "file56.exp" + view = max(1, int(self.view.value or 1)) + + LOG.info("Opening experiment: %s (view=%d)", fname, view) + + try: + if not self.client.command.is_connected(): + await asyncio.to_thread(self.client.connect) + + await asyncio.to_thread(self.client.open_experiment, fname) + + legends = await asyncio.to_thread(self.client.fetch_legends, view=view) + masses: list[float] = [] + for item in legends: + if "mass" in item.lower(): + try: + masses.append(float(item.split()[-1])) + except Exception: + continue + self._mass_vals = masses[:10] + + for i, m in enumerate(self._mass_vals, start=1): + await getattr(self, f"mass{i}").write(m) + + await self.connected.write(1) + await self.status.write("Connected") + except Exception as exc: + LOG.exception("OpenExp failed") + await self.connected.write(0) + await self.status.write("Error") + await self.last_error.write(repr(exc)) + + return 0 + + @run_exp.putter + async def run_exp(self, instance, value): + """Write 1 to start the experiment.""" + if int(value): + try: + await asyncio.to_thread(self.client.run_experiment) + except Exception as exc: + LOG.exception("RunExp failed") + await self.last_error.write(repr(exc)) + return 0 + + @abort_exp.putter + async def abort_exp(self, instance, value): + """Write 1 to abort the running experiment.""" + if int(value): + try: + await asyncio.to_thread(self.client.abort_experiment) + except Exception as exc: + LOG.exception("AbortExp failed") + await self.last_error.write(repr(exc)) + return 0 + + @close_exp.putter + async def close_exp(self, instance, value): + """Write 1 to close the experiment file.""" + if not int(value): + return 0 + + try: + self._running = False + if self._acq_task is not None: + self._acq_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._acq_task + self._acq_task = None + + if self._links_started: + self.client.stop_links() + self._links_started = False + + await asyncio.to_thread(self.client.close_experiment) + await self.connected.write(0) + await self.status.write("Disconnected") + except Exception as exc: + LOG.exception("CloseExp failed") + await self.last_error.write(repr(exc)) + + return 0 + + @go.putter + async def go(self, instance, value): + """Momentary: start experiment using GoOD/GoOT/GoFilename config PVs.""" + if not int(value): + return value + try: + od = bool(int(self.go_od.value)) + ot = bool(int(self.go_ot.value)) + fn = (self.go_filename.value or "").strip() or None + await asyncio.to_thread(self.client.x_go, filename=fn, od=od, ot=ot) + except Exception as exc: + LOG.exception("Go failed") + await self.last_error.write(repr(exc)) + return 0 + + @abort.putter + async def abort(self, instance, value): + """Momentary: abort and wait for Stopped* status.""" + if not int(value): + return value + try: + final = await asyncio.to_thread( + self.client.safe_abort_and_wait, + timeout_s=30.0, ) - return value + await self.status.write(final) + except MASsoftTimeout as exc: + await self.last_error.write(repr(exc)) + except Exception as exc: + LOG.exception("Abort failed") + await self.last_error.write(repr(exc)) + return 0 - @experiment_name.putter - async def experiment_name(self, instance, value): - return value + @close.putter + async def close(self, instance, value): + """Momentary: safe abort->wait->close. MASsoft drops all file sockets.""" + if not int(value): + return value + try: + self._running = False + if self._acq_task is not None: + self._acq_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._acq_task + self._acq_task = None + + await asyncio.to_thread( + self.client.safe_abort_and_close, + abort_timeout_s=30.0, + reconnect=False, + ) + self._links_started = False + await self.connected.write(0) + await self.status.write("Disconnected") + except Exception as exc: + LOG.exception("Close failed") + await self.last_error.write(repr(exc)) + return 0 @acquire.putter async def acquire(self, instance, value): - """Triggered when someone writes to the START PV.""" - want_acquire = bool(int(value)) - if want_acquire and not self._running: - logging.info("Starting acquisition loop") + """Start/stop acquisition loop.""" + want = bool(int(value)) + + if want and not self._running: + LOG.info("Starting acquisition loop") self._running = True - # spawn background task - self._task = asyncio.create_task(self._acquire_loop()) - elif not want_acquire and self._running: - logging.info("Stopping acquisition loop") + self._acq_task = asyncio.create_task(self._acquire_loop()) + elif not want and self._running: + LOG.info("Stopping acquisition loop") self._running = False - if self._task: - self._task.cancel() + if self._acq_task is not None: + self._acq_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._acq_task + self._acq_task = None + return value + # ----------------------------------------------------------------- + # Acquisition loops + # ----------------------------------------------------------------- + async def _acquire_loop(self): - """Read all channels once per second until stopped.""" + """Hot-link with polling fallback.""" try: - # Parses RGA data headers into PVs - headers, path = self.client.get_legends(1) - mass_values = [ - float(h.split()[-1]) - for h in headers - if 'mass' in h + view = max(1, int(self.view.value or 1)) + + # Fetch legends and populate mass PVs + legends, path = await asyncio.to_thread(self.client.get_legends, view) + mass_vals = [float(h.split()[-1]) for h in legends if "mass" in h.lower()][ + :10 ] - # print('Mass values: {}'.format(mass_values)) - for idx, mass_val in enumerate(mass_values[:10], start=1): - # print(f'Mass {idx}: {mass_val}') - pv = getattr(self, f'mass{idx}') - await pv.write(mass_val) - logging.debug(f"Wrote {mass_val:.2f} to {pv.name}") - - self.client.open_experiment_data(path) - while self._running: - print(self._running) - # print('Trying......') - try: - raw_data = self.client.data_socket.send_command(f"-lData -v1") - if raw_data != '0': - lines = raw_data.strip().split('\r\n') - # print(f'Lines: {lines}') - for line in lines: - if line.strip() == '0': - # print("Ignoring first line with '0'.") - continue - values = line.split()[2:] - # print(f'Values: {values}') - # print(f'Length of values: {len(values)}') - # print('Length of mass values: {}'.format(len(mass_values))) - if len(values) == len(mass_values): - # print('updating PVs') - for idx, val in enumerate(values): - # print(f'Index {idx}: {val}') - pv = getattr(self, f'mid{idx+1}') - print(pv) - await pv.write(float(val)) - logging.debug(f'Wrote {val} to {pv.name}') - except Exception as e: - print(f"Caught an exception: {e}") - if str(e).startswith('[Errno 32]'): - self._running = 0 - - await asyncio.sleep(1.0) + self._mass_vals = mass_vals + for i, m in enumerate(mass_vals, start=1): + await getattr(self, f"mass{i}").write(m) + + # Try hot-link threads first + try: + await asyncio.to_thread(self.client.start_status_link, view=view) + await asyncio.to_thread(self.client.start_data_link, view=view) + self._links_started = True + LOG.info("Hot-links started, using hotlink update loop") + await self._hotlink_update_loop() + except Exception as exc: + LOG.warning("Hot-links failed (%s), falling back to polling", exc) + self._links_started = False + await self._polling_acquire_loop(view, mass_vals, path) + except asyncio.CancelledError: - logging.info("Acquisition loop cancelled") - return + LOG.info("Acquisition loop cancelled") + except Exception as exc: + LOG.error("Error in acquisition loop: %s", exc) + with contextlib.suppress(Exception): + await self.last_error.write(repr(exc)) - @run_exp.putter - async def run_exp(self, instance, value): - want = bool(int(value)) - if want: - loop = asyncio.get_running_loop() - await loop.run_in_executor(None, self.client.run_experiment) + async def _hotlink_update_loop(self): + """Read thread-safe accessors at ~1 Hz and update PVs.""" + while self._running: + now = time.monotonic() - return value + # Status + st = self.client.get_latest_status() + st_ts = self.client.get_latest_status_timestamp() + if st and st_ts > self._last_pub_status_ts: + await self.status.write(st) + self._last_pub_status_ts = st_ts + if st_ts > 0: + await self.status_age.write(max(0.0, now - st_ts)) + else: + await self.status_age.write(-1.0) - @abort_exp.putter - async def abort_exp(self, instance, value): - """Write 1 to abort the running experiment, always resets to 0.""" - self.client.shutdown() - self.client.initialize() - self.client.command_socket.send_command('-f"%HIDEN_LastFile%"') - path = self.client.command_socket.send_command('-xFilename') - # print(f'Aborting experiment: {path}') - want = bool(int(value)) - if want: - loop = asyncio.get_running_loop() - await loop.run_in_executor(None, self.client.abort_experiment) - return value + # Data + row = self.client.get_latest_row() + row_ts = self.client.get_latest_row_timestamp() + if row and row_ts > self._last_pub_row_ts: + for i, val in enumerate(row[:10], start=1): + await getattr(self, f"mid{i}").write(val) + self._last_pub_row_ts = row_ts + if row_ts > 0: + await self.data_age.write(max(0.0, now - row_ts)) + else: + await self.data_age.write(-1.0) - @close_exp.putter - async def close_exp(self, instance, value): - self.client.command_socket.send_command('-f"%HIDEN_LastFile%"') - want = bool(int(value)) - if want: - loop = asyncio.get_running_loop() - await loop.run_in_executor(None, self.client.close_experiment) - return value + # Error + err = self.client.get_last_error() + if err: + await self.last_error.write(err) + await asyncio.sleep(1.0) + + async def _polling_acquire_loop(self, view, mass_vals, path): + """Working-version direct -lData polling with 5-attempt reconnect retry.""" + # Associate data socket with the experiment + try: + await asyncio.to_thread(self.client.open_experiment_data, path) + except Exception as exc: + LOG.error("Failed to associate data socket: %s", exc) + return -if __name__ == '__main__': + while self._running: + try: + raw_data = await asyncio.to_thread( + self.client.data_socket.send_command, + f"-lData -v{view}", + ) + if raw_data and raw_data != "0": + values = raw_data.split()[2:] # Skip time columns + if len(values) >= len(mass_vals): + for i, val in enumerate(values[: len(mass_vals)], start=1): + await getattr(self, f"mid{i}").write(float(val)) + except Exception as exc: + LOG.error("Socket error during polling: %s", exc) + self.client.data_socket.close() + for attempt in range(5): + try: + await asyncio.to_thread(self.client.data_socket.connect) + LOG.info("Reconnected data socket") + break + except Exception as reconnect_err: + LOG.error( + "Reconnect attempt %d failed: %s", + attempt + 1, + reconnect_err, + ) + await asyncio.sleep(5) + await asyncio.sleep(1.0) + + +if __name__ == "__main__": + # Parse MASsoft connection arguments + parser = argparse.ArgumentParser(description="RGA MASsoft IOC") + parser.add_argument( + "--mas-host", + default=None, + help="MASsoft host address (overrides config file)", + ) + parser.add_argument( + "--mas-port", + type=int, + default=None, + help="MASsoft port number (overrides config file)", + ) + args, remaining = parser.parse_known_args(sys.argv[1:]) + + # Let caproto parse its own arguments from remaining args + sys.argv = [sys.argv[0], *remaining] ioc_opts, run_opts = ioc_arg_parser( - default_prefix='', # PV names include the {{RGA:1}} macro literally - desc='RGA MASsoft IOC' + default_prefix="", + desc="RGA MASsoft IOC", ) - ioc = RGAIOC(**ioc_opts) + + ioc = RGAIOC(mas_host=args.mas_host, mas_port=args.mas_port, **ioc_opts) run(ioc.pvdb, **run_opts) diff --git a/src/hiden/hiden_config.json b/src/hiden/hiden_config.json new file mode 100644 index 0000000..8d7aa6a --- /dev/null +++ b/src/hiden/hiden_config.json @@ -0,0 +1,38 @@ +{ + "beamline_name": "08IDB", + "massoft": { + "host": "10.66.58.225", + "port": 5026, + "experiment_directory": "C:\\Users\\08id-user\\Documents\\Hiden Analytical\\MASsoft\\11", + "retry_s": 15, + "command_timeout_s": 20.0, + "link_chunk_timeout_s": 1.0, + "link_burst_gap_s": 0.1, + "enable_keepalive": true + }, + "ioc": { + "default_experiment": "file56.exp", + "default_view": 1, + "update_period_s": 1.0, + "default_go_od": 1, + "default_go_ot": 1, + "default_go_filename": "", + "start_links_on_open_exp": 0, + "default_data_cycles": 1, + "default_data_time_fmt": 0, + "default_data_ms_fmt": 0 + }, + "archiver": { + "sampling_mode": "Monitor", + "sampling_period_s": 1.0, + "condition_pv": "XF:08IDB-SE{RGA:1}:Acquire" + }, + "epics": { + "ca_server_port": 5064, + "ca_repeater_port": 5065, + "ca_addr_list": "10.66.59.255", + "ca_auto_addr_list": "NO", + "cas_auto_beacon_addr_list": "NO", + "cas_beacon_addr_list": "10.66.59.255" + } +} diff --git a/src/hiden/massoft_client.py b/src/hiden/massoft_client.py index 724e1a9..f2ebf8f 100644 --- a/src/hiden/massoft_client.py +++ b/src/hiden/massoft_client.py @@ -1,291 +1,1165 @@ -from pathlib import PureWindowsPath -import socket -import time +""" +MASsoft sockets client (merged). + +Combines the protocol-correct CRLF framing, hot-link threads, and diagnostics from the +rewrite with backward-compatible module constants and method aliases from the working +version. + +MASsoft Sockets protocol rules: + +- Command and response strings are CRLF terminated. +- Read a response for every command before sending the next command. +- Use dedicated sockets for hot links; once a link is established, do not send further + commands on that socket. +- Do not close a socket while a command is in progress; the worst-case duration is + bounded by the "-d" retry window you request. +""" + +from __future__ import annotations + +import contextlib +import json import logging -from contextlib import closing import os +import socket +import threading +import time +from dataclasses import dataclass, replace +from pathlib import Path, PureWindowsPath +from typing import Any, Callable + +LOG = logging.getLogger(__name__) +CRLF = b"\r\n" +DEFAULT_CONFIG_FILE_NAME = "hiden_config.json" -# Logger setup -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(message)s", - handlers=[logging.StreamHandler(), logging.FileHandler("massoft_client.log")] -) +# --------------------------------------------------------------------------- +# Module-level constants (backward compat for IPython workflows) +# --------------------------------------------------------------------------- -# System Configuration -MAS_HOST = '10.66.58.225' +MAS_HOST = "10.66.58.225" MAS_PORT = 5026 EXPERIMENT_DIRECTORY = r"C:\Users\08id-user\Documents\Hiden Analytical\MASsoft\11" -EXPERIMENT_DIRECTORY_ENV = "%HIDEN_FilePath%" # Environment variable name for the experiment directory -MOST_RECENT_FILE = "%HIDEN_LastFile%" # This environment variable name already includes the path -TIME_PERSISTANCE = 20 # Time in seconds for the messages to keep trying waiting for success -MESSAGE_TERMINATOR = "\r\n" +EXPERIMENT_DIRECTORY_ENV = "HIDEN_FilePath" +MOST_RECENT_FILE = "HIDEN_LastFile" +TEMPLATE_DICT = { + "exp1": "HIDEN_1.exp", + "exp2": "HIDEN_2.exp", + "exp3": "HIDEN_3.exp", + "exp4": "HIDEN_4.exp", +} + + +# --------------------------------------------------------------------------- +# Runtime config helpers +# --------------------------------------------------------------------------- + + +def get_runtime_config_path(config_path: str | None = None) -> Path: + """ + Resolve runtime config path. + + Priority: + 1) explicit ``config_path`` + 2) ``HIDEN_CONFIG`` environment variable + 3) ``/hiden_config.json`` + """ + if config_path: + return Path(config_path).expanduser() + + env_path = os.getenv("HIDEN_CONFIG") + if env_path: + return Path(env_path).expanduser() + + return Path(__file__).with_name(DEFAULT_CONFIG_FILE_NAME) + + +def load_runtime_config(config_path: str | None = None) -> dict[str, Any]: + """Load runtime JSON config. Returns empty dict when file is missing.""" + path = get_runtime_config_path(config_path=config_path) + if not path.exists(): + return {} + + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + msg = f"Invalid JSON in runtime config: {path}" + raise ValueError(msg) from exc + + if not isinstance(raw, dict): + msg = f"Runtime config must be a JSON object: {path}" + raise ValueError(msg) + + return raw + + +# --------------------------------------------------------------------------- +# Exceptions +# --------------------------------------------------------------------------- + + +class MASsoftError(Exception): + """Base exception for MASsoft socket client errors.""" + + +class MASsoftDisconnected(MASsoftError, ConnectionError): + """Raised when the remote side closes the socket or the socket is unusable.""" + -class MASsoftSocket: - def __init__(self, host, port, name="GenericSocket", timeout=20): +class MASsoftTimeout(MASsoftError, TimeoutError): + """Raised when a read/write exceeds the requested timeout.""" + + +class MASsoftProtocolError(MASsoftError): + """Raised when MASsoft returns '0' (failure) or protocol framing is violated.""" + + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class MASsoftConfig: + beamline_name: str = "08IDB" + host: str = "10.66.58.225" + port: int = 5026 + + # Directory used when ``open_experiment("file56.exp")`` is called. + experiment_directory: str = ( + r"C:\Users\08id-user\Documents\Hiden Analytical\MASsoft\11" + ) + + # Default MASsoft "-d" retry window for commands that support it. + retry_s: int = 15 + + # IMPORTANT: must be >= retry_s (+ margin), otherwise you can time out locally + # while MASsoft continues retrying and later sends a "late" response that + # desynchronizes the stream. + command_timeout_s: float = 20.0 + + # Per-recv timeout used by link reader threads. + link_chunk_timeout_s: float = 1.0 + + # Group multiple CRLF lines that arrive in a short burst into one callback. + link_burst_gap_s: float = 0.10 + + enable_keepalive: bool = True + + @classmethod + def from_runtime_config(cls, config_path: str | None = None) -> MASsoftConfig: + """Build config from runtime JSON.""" + raw = load_runtime_config(config_path=config_path) + massoft = raw.get("massoft", {}) + if not isinstance(massoft, dict): + msg = "runtime config key 'massoft' must be an object" + raise ValueError(msg) + + def _get(key: str, default: Any) -> Any: + if key in massoft: + return massoft[key] + return raw.get(key, default) + + return cls( + beamline_name=str(raw.get("beamline_name", cls.beamline_name)), + host=str(_get("host", cls.host)), + port=int(_get("port", cls.port)), + experiment_directory=str( + _get("experiment_directory", cls.experiment_directory) + ), + retry_s=int(_get("retry_s", cls.retry_s)), + command_timeout_s=float(_get("command_timeout_s", cls.command_timeout_s)), + link_chunk_timeout_s=float( + _get("link_chunk_timeout_s", cls.link_chunk_timeout_s) + ), + link_burst_gap_s=float(_get("link_burst_gap_s", cls.link_burst_gap_s)), + enable_keepalive=bool(_get("enable_keepalive", cls.enable_keepalive)), + ) + + +# --------------------------------------------------------------------------- +# Low-level CRLF framed socket +# --------------------------------------------------------------------------- + + +class _CRLFSocket: + """ + A single TCP connection to MASsoft with CRLF framed reads. + + - ``request(...)`` is serialized by a lock so command/response pairs cannot + interleave. + - For hot-links, call ``send(...)`` once, then only call ``read_line(...)`` + from one thread. + """ + + def __init__(self, host: str, port: int, *, name: str, timeout_s: float): self.host = host self.port = port self.name = name - self.timeout = timeout - self.sock = None + self._timeout_s = float(timeout_s) - def connect(self): - """Establish or re-establish the socket connection.""" - if self.sock: - try: - # Test current socket - self.sock.sendall(b'') - return - except Exception: - self.close() - self.sock = socket.create_connection((self.host, self.port)) - self.sock.settimeout(self.timeout) - logging.info(f"{self.name} connected to {self.host}:{self.port}") + self._sock: socket.socket | None = None + self._buf = bytearray() + self._req_lock = threading.Lock() + + def connect(self, *, enable_keepalive: bool = True) -> None: + self.close() + + sock = socket.create_connection((self.host, self.port), timeout=self._timeout_s) + sock.settimeout(self._timeout_s) + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + + if enable_keepalive: + with contextlib.suppress(OSError): + sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) + + self._sock = sock + self._buf.clear() + + # Best-effort: consume the greeting if it arrives quickly. try: - _ = self.sock.recv(4096) - except socket.timeout: + _ = self.read_line(timeout_s=1.0) + except MASsoftTimeout: pass + except MASsoftDisconnected: + raise - def send_command(self, command, expect_response=True): - if not self.sock: - raise RuntimeError(f"{self.name} not connected.") - # Append retry delay and CRLF - message = command.strip() + f' -d{TIME_PERSISTANCE}{MESSAGE_TERMINATOR}' - self.sock.sendall(message.encode('utf-8')) - if expect_response: - try: - resp = self.sock.recv(4096).decode('utf-8').strip() - except socket.timeout: - logging.warning(f"{self.name} response timeout for: {message.strip()}") - return '' - logging.info(f"{self.name} | {message.strip()} => {resp}") - return resp - return '' - - def receive(self): - if not self.sock: - raise RuntimeError(f"{self.name} not connected.") + LOG.info("%s connected to %s:%d", self.name, self.host, self.port) + + def close(self) -> None: + sock = self._sock + self._sock = None + self._buf.clear() + if sock is None: + return + + with contextlib.suppress(OSError): + sock.shutdown(socket.SHUT_RDWR) + with contextlib.suppress(OSError): + sock.close() + LOG.info("%s closed", self.name) + + def is_connected(self) -> bool: + return self._sock is not None + + def send(self, command: str, *, retry_s: int | None = None) -> None: + if self._sock is None: + msg = f"{self.name}: not connected" + raise MASsoftDisconnected(msg) + + cmd = command.strip() + + # Append -d only if requested and not already present. + if retry_s is not None and " -d" not in cmd: + cmd = f"{cmd} -d{int(retry_s)}" + + wire = (cmd + "\r\n").encode("utf-8") try: - return self.sock.recv(4096).decode('utf-8').strip() - except socket.timeout: - return '' + self._sock.sendall(wire) + except OSError as exc: + msg = f"{self.name}: send failed: {exc!r}" + raise MASsoftDisconnected(msg) from exc + + def read_line(self, *, timeout_s: float | None = None) -> str: + if self._sock is None: + msg = f"{self.name}: not connected" + raise MASsoftDisconnected(msg) + + sock = self._sock + old_timeout = sock.gettimeout() + if timeout_s is not None: + sock.settimeout(float(timeout_s)) + + try: + while True: + idx = self._buf.find(CRLF) + if idx != -1: + raw = bytes(self._buf[:idx]) + del self._buf[: idx + 2] + return raw.decode("utf-8", errors="replace") + + try: + chunk = sock.recv(4096) + except socket.timeout as exc: + msg = f"{self.name}: read timed out" + raise MASsoftTimeout(msg) from exc + except OSError as exc: + msg = f"{self.name}: read failed: {exc!r}" + raise MASsoftDisconnected( + msg + ) from exc + + if chunk == b"": + msg = f"{self.name}: remote closed connection" + raise MASsoftDisconnected(msg) + + self._buf.extend(chunk) + finally: + if timeout_s is not None: + sock.settimeout(old_timeout) + + def request( + self, + command: str, + *, + retry_s: int | None = None, + timeout_s: float | None = None, + ) -> str: + """Send one command and read one CRLF-terminated response line.""" + with self._req_lock: + self.send(command, retry_s=retry_s) + return self.read_line(timeout_s=timeout_s) + + # ------------------------------------------------------------------- + # Compat methods (working-version API) + # ------------------------------------------------------------------- + + def send_command(self, command: str, expect_response: bool = True) -> str: + """Compat: send command with ``-d20`` retry and return response. + + Returns ``""`` on timeout. Raises on disconnect so callers can + trigger reconnection logic. + """ + if not expect_response: + self.send(command, retry_s=20) + return "" + try: + return self.request(command, retry_s=20, timeout_s=25.0) + except MASsoftTimeout: + LOG.warning("%s: timeout for: %s", self.name, command.strip()) + return "" + + def receive(self) -> str: + """Compat: read one CRLF line, return ``""`` on timeout/disconnect.""" + try: + return self.read_line() + except (MASsoftTimeout, MASsoftDisconnected): + return "" + + +# --------------------------------------------------------------------------- +# Hot-link subscription helper +# --------------------------------------------------------------------------- + + +class MASsoftHotlink: + """ + A dedicated listening socket reading a MASsoft ``-l`` hot link. + + The callback is invoked with a *burst* of one or more lines. Bursts are + separated by idle gaps > ``burst_gap_s``. + """ + + def __init__( + self, + sock: _CRLFSocket, + *, + chunk_timeout_s: float, + burst_gap_s: float, + on_burst: Callable[[list[str]], None], + name: str, + ): + self._sock = sock + self._chunk_timeout_s = float(chunk_timeout_s) + self._burst_gap_s = float(burst_gap_s) + self._on_burst = on_burst + self._name = name + + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + def start(self) -> None: + self._stop.clear() + t = threading.Thread(target=self._run, name=self._name, daemon=True) + self._thread = t + t.start() + + def stop(self, *, join_timeout_s: float = 2.0) -> None: + self._stop.set() + t = self._thread + if t is not None and t.is_alive(): + t.join(timeout=float(join_timeout_s)) + self._thread = None + + def _run(self) -> None: + pending: list[str] = [] + last_rx: float | None = None + + def _emit_pending() -> None: + nonlocal pending, last_rx + if not pending: + return + try: + self._on_burst(pending) + except Exception: + LOG.exception("%s callback error", self._name) + pending = [] + last_rx = None + + while not self._stop.is_set(): + try: + line = self._sock.read_line(timeout_s=self._chunk_timeout_s) + s = line.strip() + if s: + now = time.monotonic() + if ( + pending + and last_rx is not None + and (now - last_rx) >= self._burst_gap_s + ): + _emit_pending() + pending.append(s) + last_rx = now + except MASsoftTimeout: + if pending and last_rx is not None and (time.monotonic() - last_rx) >= self._burst_gap_s: + _emit_pending() + except MASsoftDisconnected as exc: + LOG.warning("%s disconnected: %s", self._name, exc) + break + except Exception: + LOG.exception("%s reader error", self._name) + break + + if pending and last_rx is not None and self._burst_gap_s <= 0: + _emit_pending() + + # Flush any remaining lines + if pending: + _emit_pending() + + +# --------------------------------------------------------------------------- +# High-level MASsoft client +# --------------------------------------------------------------------------- - def close(self): - if self.sock: - self.sock.close() - logging.info(f"{self.name} closed.") class MASsoftClient: - def __init__(self, host=MAS_HOST, port=MAS_PORT): - self.command_socket = MASsoftSocket(host, port, name="CommandSocket") - self.status_socket = MASsoftSocket(host, port, name="StatusSocket") - self.data_socket = MASsoftSocket(host, port, name="DataSocket") - self.current_file = MOST_RECENT_FILE - - def initialize(self): - """Connect all sockets.""" - self.command_socket.connect() - self.status_socket.connect() - self.data_socket.connect() - - def open_experiment_commands(self, file_name=None): - """Open and associate an experiment file. - If file_name is None, query MASsoft for the current filename.""" - # 1) Determine the filename string - if file_name is None: - full_path = self.query_filename() + """ + High-level MASsoft client. + + Accepts multiple constructor styles for backward compatibility:: + + MASsoftClient() # load from config file + MASsoftClient(cfg) # MASsoftConfig instance + MASsoftClient(host="...", port=5026) # working-version keyword style + """ + + def __init__( + self, + cfg: MASsoftConfig | None = None, + *, + host: str | None = None, + port: int | None = None, + config_path: str | None = None, + ): + if isinstance(cfg, MASsoftConfig): + self.cfg = cfg + elif cfg is None: + base = MASsoftConfig.from_runtime_config(config_path=config_path) + if host is not None or port is not None: + overrides: dict[str, Any] = {} + if host is not None: + overrides["host"] = host + if port is not None: + overrides["port"] = port + self.cfg = replace(base, **overrides) + else: + self.cfg = base else: - # Coerce list or tuple into a single string - if isinstance(file_name, (list, tuple)): - file_name = file_name[0] - # Build a pure-Windows path - full_path = str(PureWindowsPath(EXPERIMENT_DIRECTORY) / file_name) - - # 2) Send to MASsoft - resp = self.command_socket.send_command(f'-f"{full_path}"') - if resp =='0': - raise RuntimeError(f"Failed to open experiment file: {full_path}") - - # 3) Remember it for future operations - self.current_file = full_path - return full_path - - def open_experiment_data(self, file_name=None): - """Open and associate an experiment file. - If file_name is None, query MASsoft for the current filename.""" - # 1) Determine the filename string + msg = f"Expected MASsoftConfig, None, or keyword host=/port=; got {type(cfg)}" + raise TypeError( + msg + ) + + # Use self.cfg (not cfg) -- fixes rewrite bug where cfg could be None. + self.command = _CRLFSocket( + self.cfg.host, + self.cfg.port, + name="MASsoftCommand", + timeout_s=self.cfg.command_timeout_s, + ) + self.status_sock = _CRLFSocket( + self.cfg.host, + self.cfg.port, + name="MASsoftStatus", + timeout_s=self.cfg.link_chunk_timeout_s, + ) + self.data_sock = _CRLFSocket( + self.cfg.host, + self.cfg.port, + name="MASsoftData", + timeout_s=self.cfg.link_chunk_timeout_s, + ) + + self.current_file: str | None = None + self._command_assoc_file: str | None = None + self._status_assoc_file: str | None = None + self._data_assoc_file: str | None = None + self._assoc_fallback_logged: set[str] = set() + + self._status_link: MASsoftHotlink | None = None + self._data_link: MASsoftHotlink | None = None + + self._latest_status_lock = threading.Lock() + self._latest_status: str | None = None + self._latest_status_ts = 0.0 + + self._latest_row_lock = threading.Lock() + self._latest_row: list[float] | None = None + self._latest_row_ts = 0.0 + self._latest_raw_row: str | None = None + self._latest_raw_row_ts = 0.0 + + self._last_error_lock = threading.Lock() + self._last_error: str | None = None + + # ------------------------------------------------------------------- + # Socket attribute aliases (working-version names) + # ------------------------------------------------------------------- + + @property + def command_socket(self) -> _CRLFSocket: + return self.command + + @property + def status_socket(self) -> _CRLFSocket: + return self.status_sock + + @property + def data_socket(self) -> _CRLFSocket: + return self.data_sock + + # ------------------------------------------------------------------- + # Connection lifecycle + # ------------------------------------------------------------------- + + def connect(self) -> None: + """Connect all sockets (command + link sockets).""" + self.command.connect(enable_keepalive=self.cfg.enable_keepalive) + self.status_sock.connect(enable_keepalive=self.cfg.enable_keepalive) + self.data_sock.connect(enable_keepalive=self.cfg.enable_keepalive) + + def disconnect(self) -> None: + """Stop hot-links and close sockets.""" + self.stop_links() + self.command.close() + self.status_sock.close() + self.data_sock.close() + self.current_file = None + self._command_assoc_file = None + self._status_assoc_file = None + self._data_assoc_file = None + self._assoc_fallback_logged.clear() + + def initialize(self) -> None: + """Compat: connect all sockets. Delegates to ``connect()``.""" + self.connect() + + def shutdown(self) -> None: + """Compat: close all sockets. Delegates to ``disconnect()``.""" + self.disconnect() + + # ------------------------------------------------------------------- + # File open / association + # ------------------------------------------------------------------- + + def _resolve_path(self, file_name_or_path: str) -> str: + if ( + file_name_or_path.startswith(("%", "\\\\")) or ":" in file_name_or_path + ): + return file_name_or_path + return str(PureWindowsPath(self.cfg.experiment_directory) / file_name_or_path) + + def open_experiment( + self, file_name_or_path: str | None = None, *, retry_s: int | None = None + ) -> str: + """ + Open/associate an experiment file on all sockets. + + When *file_name_or_path* is ``None``, queries MASsoft for the current + filename (like the working version). Lists/tuples are unpacked for + backward compatibility. + """ + if retry_s is None: + retry_s = self.cfg.retry_s + + if file_name_or_path is None: + path = self.query_filename(retry_s=retry_s, update_current=True) + else: + if isinstance(file_name_or_path, (list, tuple)): + file_name_or_path = file_name_or_path[0] + path = self._resolve_path(file_name_or_path) + + r = self.command.request( + f'-f"{path}"', + retry_s=retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if r == "0": + msg = f"MASsoft failed to open/associate: {path}" + raise MASsoftProtocolError(msg) + self._command_assoc_file = path + + for s in (self.status_sock, self.data_sock): + rr = s.request( + f'-f"{path}"', + retry_s=retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if rr == "0": + msg = f"MASsoft failed to associate {s.name} with: {path}" + raise MASsoftProtocolError( + msg + ) + if s is self.status_sock: + self._status_assoc_file = path + elif s is self.data_sock: + self._data_assoc_file = path + + self.current_file = path + return path + + # Compat alias used by the IOC (cap2.py) + open_experiment_commands = open_experiment + + def open_experiment_data(self, file_name: str | None = None) -> str: + """Compat: associate an experiment file on the data socket only.""" if file_name is None: - full_path = self.query_filename_data() + path = self.query_filename_data() else: - # Coerce list or tuple into a single string if isinstance(file_name, (list, tuple)): file_name = file_name[0] - # Build a pure-Windows path - full_path = str(PureWindowsPath(EXPERIMENT_DIRECTORY) / file_name) - - # 2) Send to MASsoft - resp = self.data_socket.send_command(f'-f"{full_path}"') - if resp =='0': - raise RuntimeError(f"Failed to open experiment file: {full_path}") - - # 3) Remember it for future operations - self.current_file = full_path - return full_path - - def open_experiment_status(self, file_name=None): - """Open and associate an experiment file. - If file_name is None, query MASsoft for the current filename.""" - # 1) Determine the filename string + path = self._resolve_path(file_name) + r = self.data_sock.request( + f'-f"{path}"', + retry_s=self.cfg.retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if r == "0": + msg = f"Failed to associate data socket: {path}" + raise MASsoftProtocolError(msg) + self._data_assoc_file = path + self.current_file = path + return path + + def open_experiment_status(self, file_name: str | None = None) -> str: + """Compat: associate an experiment file on the status socket only.""" if file_name is None: - full_path = self.query_filename() + path = self.query_filename(update_current=False) else: - # Coerce list or tuple into a single string if isinstance(file_name, (list, tuple)): file_name = file_name[0] - # Build a pure-Windows path - full_path = str(PureWindowsPath(EXPERIMENT_DIRECTORY) / file_name) - - # 2) Send to MASsoft - resp = self.status_socket_socket.send_command(f'-f"{full_path}"') - if resp =='0': - raise RuntimeError(f"Failed to open experiment file: {full_path}") - - # 3) Remember it for future operations - self.current_file = full_path - return full_path - - def run_experiment(self, new_file_name = None, mode = "-Odt"): - """Start the experiment.""" - resp = self.command_socket.send_command(f'-xGo {mode}') - if resp == '0': - raise RuntimeError("Experiment failed to start.") - if not resp: - logging.warning("Assuming experiment started despite no response.") - - def associate_status_link(self, view=1): - """Set up a hot-link for status updates.""" - if not self.current_file: - raise RuntimeError("No file opened.") - self.open_experiment_status() - self.status_socket.send_command(f'-lStatus -v{view}') - - def monitor_until_stopped(self, timeout=120): - """ - Listen for status updates until a 'Stopped...' status arrives. - Requires associate_status_link() to be called first. - """ - if not self.current_file: - raise RuntimeError("No file opened.") - self.associate_status_link() - start = time.time() - while time.time() - start < timeout: - status = self.status_socket.receive() - if status: - logging.info(f"Status: {status}") - if status.lower().startswith('stopped'): - return True - time.sleep(1) - raise TimeoutError(f"Did not stop within {timeout}s.") - - def get_data(self, view=1): - """Retrieve scan data via a new data socket.""" - if not self.current_file: - raise RuntimeError("No file opened.") - headers = self.get_legends(view=view) - data = [] - while True: - raw_data = self.data_socket.send_command(f"-lData -v{view}") - if raw_data != '0': - lines = raw_data.strip().split('\r\n') - # print(f'Lines: {lines}') - for line in lines: - if line.strip() == '0': - # print("Ignoring first line with '0'.") - continue - values = line.split() - # print(f'Values: {values}') - if len(values) < len(headers): - # print(f"Line skipped due to insufficient values: {line.strip()}") - continue - data.append(values) - # print(f"Data appended: {values}") - time.sleep(1) + path = self._resolve_path(file_name) + r = self.status_sock.request( + f'-f"{path}"', + retry_s=self.cfg.retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if r == "0": + msg = f"Failed to associate status socket: {path}" + raise MASsoftProtocolError(msg) + self._status_assoc_file = path + self.current_file = path + return path + + def query_filename( + self, *, retry_s: int | None = None, update_current: bool = True + ) -> str: + """Return the filename currently associated with the command socket.""" + if retry_s is None: + retry_s = self.cfg.retry_s + path = self.command.request( + "-xFilename", + retry_s=retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if path == "0" or not path: + msg = "MASsoft refused -xFilename (returned 0/empty)" + raise MASsoftProtocolError(msg) - def get_legends(self, view=1): - """Retrieve column legends via a temporary socket.""" - path = self.command_socket.send_command("-xFilename") - time.sleep(1) - path = self.command_socket.send_command("-xFilename") - self.command_socket.send_command(f'-f"{path}"') + path = path.strip().strip('"') + if not path: + msg = "MASsoft returned an empty filename for -xFilename" + raise MASsoftProtocolError( + msg + ) + + if update_current: + self.current_file = path + + return path + + def query_filename_data(self) -> str: + """Compat: return the filename currently associated with the data socket.""" + resp = self.data_sock.request( + "-xFilename", + retry_s=self.cfg.retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if resp in ("0", ""): + msg = "Failed to retrieve filename from data socket" + raise MASsoftProtocolError(msg) + return resp.strip('"') + + def _associate_socket_with_active_file( + self, + sock: _CRLFSocket, + *, + retry_s: int | None = None, + ) -> str: + """Associate a socket with the currently active experiment file.""" + if retry_s is None: + retry_s = self.cfg.retry_s + + if sock is self.command: + fallback_assoc = self._command_assoc_file + elif sock is self.status_sock: + fallback_assoc = self._status_assoc_file + elif sock is self.data_sock: + fallback_assoc = self._data_assoc_file + else: + fallback_assoc = None + + active_file = self.current_file try: - while True: - raw_data = self.command_socket.send_command(f"-lLegends -v{view}") - if raw_data != '0': - legend = raw_data.replace("\r\n", "\t").split("\t") - break - else: - time.sleep(1) - except KeyboardInterrupt: - print("Done.") - return legend, path - - def get_legends_data(self, view=1): - """Retrieve column legends via a temporary socket.""" - path = self.data_socket.send_command("-xFilename") - time.sleep(1) - path = self.data_socket.send_command("-xFilename") - self.command_socket.send_command(f'-f"{path}"') + active_file = self.query_filename(retry_s=retry_s, update_current=True) + except Exception: + if not active_file: + raise + + rr = sock.request( + f'-f"{active_file}"', + retry_s=retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if rr != "0": + if sock is self.command: + self._command_assoc_file = active_file + elif sock is self.status_sock: + self._status_assoc_file = active_file + elif sock is self.data_sock: + self._data_assoc_file = active_file + self._assoc_fallback_logged.discard(sock.name) + return active_file + + if fallback_assoc: + msg = ( + f"{sock.name} association to active file failed; " + f"keeping existing association: {fallback_assoc}" + ) + if sock.name not in self._assoc_fallback_logged: + LOG.info(msg) + self._assoc_fallback_logged.add(sock.name) + return fallback_assoc + + msg = f"MASsoft failed to associate {sock.name} with: {active_file}" + raise MASsoftProtocolError( + msg + ) + + # ------------------------------------------------------------------- + # Execute (-x*) commands + # ------------------------------------------------------------------- + + def x_status(self, *, retry_s: int | None = None) -> str: + """Return current MSIU status via ``-xStatus``.""" + if retry_s is None: + retry_s = self.cfg.retry_s + return self.command.request( + "-xStatus", + retry_s=retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + + def x_go( + self, + *, + filename: str | None = None, + od: bool = True, + ot: bool = True, + retry_s: int | None = None, + ) -> None: + """Start the experiment via ``-xGo``.""" + if retry_s is None: + retry_s = self.cfg.retry_s + + o_flags = "" + if od or ot: + o_flags = "-O" + ("d" if od else "") + ("t" if ot else "") + + parts = ["-xGo"] + if filename: + parts.append(filename) + if o_flags: + parts.append(o_flags) + + cmd = " ".join(parts) + r = self.command.request( + cmd, + retry_s=retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if r == "0": + msg = "MASsoft refused -xGo (returned 0)" + raise MASsoftProtocolError(msg) + + with contextlib.suppress(Exception): + self.query_filename(retry_s=retry_s, update_current=True) + + def x_abort(self, *, retry_s: int | None = None) -> None: + """Abort acquisition via ``-xAbort``.""" + if retry_s is None: + retry_s = self.cfg.retry_s + r = self.command.request( + "-xAbort", + retry_s=retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if r == "0": + msg = "MASsoft refused -xAbort (returned 0)" + raise MASsoftProtocolError(msg) + + def x_close(self, *, retry_s: int | None = None) -> None: + """Close the experiment via ``-xClose``.""" + if retry_s is None: + retry_s = self.cfg.retry_s + + r = self.command.request( + "-xClose", + retry_s=retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if r == "0": + msg = "MASsoft refused -xClose (returned 0)" + raise MASsoftProtocolError(msg) + + self.stop_links() + self.current_file = None + self._command_assoc_file = None + self._status_assoc_file = None + self._data_assoc_file = None + self._assoc_fallback_logged.clear() + + def safe_abort_and_wait(self, *, timeout_s: float = 30.0) -> str: + """Abort and wait until status becomes Stopped*.""" + self.x_abort() + + deadline = time.monotonic() + float(timeout_s) + while time.monotonic() < deadline: + st = self.get_latest_status() + if st and st.lower().startswith("stopped"): + return st + try: + st2 = self.x_status() + if st2 and st2.lower().startswith("stopped"): + self._set_latest_status(st2) + return st2 + except Exception: + pass + time.sleep(0.2) + + msg = f"Timed out waiting for stopped status after abort ({timeout_s}s)" + raise MASsoftTimeout( + msg + ) + + def safe_abort_and_close( + self, + *, + abort_timeout_s: float = 30.0, + close_retry_s: int | None = None, + reconnect: bool = False, + ) -> str: + """Safely stop (if running), then close the experiment file.""" + if close_retry_s is None: + close_retry_s = self.cfg.retry_s + + st = self.get_latest_status() + if not st: + try: + st = self.x_status() + except Exception: + st = None + + def _is_running(s: str) -> bool: + ss = s.strip().lower() + return ( + ss.startswith(("starting", "scanning", "stopping")) or ss == "degas" + ) + + final_status = st or "" + + if st and _is_running(st): + final_status = self.safe_abort_and_wait(timeout_s=abort_timeout_s) + elif st is None: + with contextlib.suppress(Exception): + final_status = self.safe_abort_and_wait(timeout_s=abort_timeout_s) + try: - while True: - raw_data = self.command_socket.send_command(f"-lLegends -v{view}") - if raw_data != '0': - legend = raw_data.replace("\r\n", "\t").split("\t") - break - else: - time.sleep(1) - except KeyboardInterrupt: - print("Done.") - return legend_data - - def query_filename(self): - """Return the filename currently associated with the command socket.""" - resp = self.command_socket.send_command('-xFilename') - if resp == '0': - raise RuntimeError("Failed querying filename.") - return resp + self.x_close(retry_s=close_retry_s) + finally: + with contextlib.suppress(Exception): + self.disconnect() - def query_filename_data(self): - """Return the filename currently associated with the command socket.""" - resp = self.data_socket.send_command('-xFilename') - if resp == '0': - raise RuntimeError("Failed querying filename.") - return resp - - def close_experiment(self): - """Close the experiment file.""" - resp = self.command_socket.send_command('-xClose') - if resp == '0': - raise RuntimeError("Close failed.") - - def abort_experiment(self): - """Abort the experiment.""" - resp = self.command_socket.send_command('-xAbort') - if resp == '0': - raise RuntimeError("Abort failed.") - - def shutdown(self): - """Close all sockets.""" - self.command_socket.close() - self.status_socket.close() - self.data_socket.close() - -# Example IPython Usage: -# from massoft_client import MASsoftClient -# client = MASsoftClient(); client.initialize() -# client.open_experiment('file56.exp') -# client.run_experiment(); client.associate_status_link() -# client.monitor_until_stopped(timeout=300) -# print(client.query_filename()) -# data = client.get_data(); legends = client.get_legends() -# client.close_experiment(); client.shutdown() + if reconnect: + self.connect() + + return final_status or (self.get_latest_status() or "") + + # ------------------------------------------------------------------- + # Compat execute methods (working-version API) + # ------------------------------------------------------------------- + + def run_experiment(self, mode: str = "-Odt") -> None: + """Compat: start the experiment. Delegates to ``x_go()``.""" + od = "d" in mode if mode else False + ot = "t" in mode if mode else False + self.x_go(od=od, ot=ot) + + def abort_experiment(self) -> None: + """Compat: abort the experiment. Delegates to ``x_abort()``.""" + self.x_abort() + + def close_experiment(self) -> None: + """Compat: close the experiment. Delegates to ``x_close()``.""" + self.x_close() + + # ------------------------------------------------------------------- + # Links (-l*) for status/data + # ------------------------------------------------------------------- + + def start_status_link(self, *, view: int = 1) -> None: + """Start a ``-lStatus`` hot-link on the dedicated status socket.""" + if self.current_file is None: + msg = "No experiment is associated. Call open_experiment(...) first." + raise RuntimeError( + msg + ) + + self._associate_socket_with_active_file( + self.status_sock, retry_s=self.cfg.retry_s + ) + + self.status_sock.send(f"-lStatus -v{int(view)}", retry_s=self.cfg.retry_s) + + def _on(lines: list[str]) -> None: + for line in lines: + if line == "0": + self._set_last_error( + "Status link returned '0' (view missing or MASsoft busy)" + ) + continue + self._set_latest_status(line) + + self._status_link = MASsoftHotlink( + self.status_sock, + chunk_timeout_s=self.cfg.link_chunk_timeout_s, + burst_gap_s=self.cfg.link_burst_gap_s, + on_burst=_on, + name="MASsoftStatusHotlink", + ) + self._status_link.start() + + def start_data_link( + self, + *, + view: int = 1, + mid_cycles: int = 1, + include_time: bool = False, + include_ms: bool = False, + ) -> None: + """Start a ``-lData`` hot-link on the dedicated data socket.""" + if self.current_file is None: + msg = "No experiment is associated. Call open_experiment(...) first." + raise RuntimeError( + msg + ) + + self._associate_socket_with_active_file( + self.data_sock, retry_s=self.cfg.retry_s + ) + + c = max(1, int(mid_cycles)) + t = 1 if include_time else 0 + m = 1 if include_ms else 0 + + cmd = f"-lData -v{int(view)}" + if c != 1 or t != 0 or m != 0: + cmd += f" -c{c} -t{t} -m{m}" + self.data_sock.send(cmd, retry_s=self.cfg.retry_s) + + drop = (1 if include_time else 0) + (1 if include_ms else 0) + + def _to_float(tok: str) -> float | None: + try: + return float(tok) + except Exception: + return None + + def _on(lines: list[str]) -> None: + last_row: list[float] | None = None + last_raw_line: str | None = None + for line in lines: + if not line: + continue + if line == "0": + self._set_last_error( + "Data link returned '0' (view missing or MASsoft busy)" + ) + continue + last_raw_line = line + parts = line.split() + + start = 0 + if parts and ":" in parts[0]: + start = 1 + start += drop + if start >= len(parts): + continue + + numeric: list[float] = [] + for p in parts[start:]: + v = _to_float(p) + if v is None: + continue + numeric.append(v) + + if not numeric: + self._set_last_error(f"Unparseable MID data line: {line!r}") + continue + + if not include_ms and len(numeric) >= 2: + first_raw = parts[start] + second_raw = parts[start + 1] if (start + 1) < len(parts) else "" + first_is_intish = first_raw.isdigit() + second_is_scan_like = ("e" in second_raw.lower()) or ( + "." in second_raw + ) + if first_is_intish and second_is_scan_like: + numeric = numeric[1:] + + if not numeric: + continue + last_row = numeric + + if last_row is not None: + with self._latest_row_lock: + self._latest_row = last_row + self._latest_row_ts = time.monotonic() + if last_raw_line is not None: + with self._latest_row_lock: + self._latest_raw_row = last_raw_line + self._latest_raw_row_ts = time.monotonic() + + self._data_link = MASsoftHotlink( + self.data_sock, + chunk_timeout_s=self.cfg.link_chunk_timeout_s, + burst_gap_s=0.0, + on_burst=_on, + name="MASsoftDataHotlink", + ) + self._data_link.start() + + def stop_links(self) -> None: + """Stop status/data hot-links (does not close sockets).""" + if self._status_link is not None: + self._status_link.stop() + self._status_link = None + if self._data_link is not None: + self._data_link.stop() + self._data_link = None + + # ------------------------------------------------------------------- + # One-shot metadata + # ------------------------------------------------------------------- + + def fetch_legends(self, *, view: int = 1) -> list[str]: + """Fetch ``-lLegends`` once on a temporary socket.""" + if self.current_file is None: + msg = "No experiment is associated. Call open_experiment(...) first." + raise RuntimeError( + msg + ) + + active_file = self.query_filename(update_current=True) + + tmp = _CRLFSocket( + self.cfg.host, + self.cfg.port, + name="MASsoftLegendsTmp", + timeout_s=self.cfg.command_timeout_s, + ) + try: + tmp.connect(enable_keepalive=self.cfg.enable_keepalive) + r = tmp.request( + f'-f"{active_file}"', + retry_s=self.cfg.retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if r == "0": + msg = "Failed to associate temp legends socket with experiment" + raise MASsoftProtocolError( + msg + ) + + line = tmp.request( + f"-lLegends -v{int(view)}", + retry_s=self.cfg.retry_s, + timeout_s=self.cfg.command_timeout_s, + ).strip() + if line == "0": + msg = "MASsoft refused -lLegends (returned 0)" + raise MASsoftProtocolError(msg) + + if "\t" in line: + return [p.strip().strip('"') for p in line.split("\t") if p.strip()] + return [line] + finally: + tmp.close() + + def get_legends(self, view: int = 1) -> tuple[list[str], str]: + """Compat: return ``(legend_list, path)`` tuple.""" + path = self.query_filename(update_current=True) + legends = self.fetch_legends(view=view) + return legends, path + + # ------------------------------------------------------------------- + # Readback helpers (thread-safe accessors) + # ------------------------------------------------------------------- + + def _set_latest_status(self, status: str) -> None: + with self._latest_status_lock: + self._latest_status = status + self._latest_status_ts = time.monotonic() + + def get_latest_status(self) -> str | None: + with self._latest_status_lock: + return self._latest_status + + def get_latest_status_timestamp(self) -> float: + with self._latest_status_lock: + return self._latest_status_ts + + def get_latest_row(self) -> list[float] | None: + with self._latest_row_lock: + return list(self._latest_row) if self._latest_row is not None else None + + def get_latest_row_timestamp(self) -> float: + with self._latest_row_lock: + return self._latest_row_ts + + def get_latest_raw_line(self) -> str | None: + with self._latest_row_lock: + return self._latest_raw_row + + def get_latest_raw_line_timestamp(self) -> float: + with self._latest_row_lock: + return self._latest_raw_row_ts + + def _set_last_error(self, msg: str) -> None: + with self._last_error_lock: + self._last_error = msg + + def get_last_error(self) -> str | None: + with self._last_error_lock: + return self._last_error