Improvements to Galil driver

docs/changes/newsfragments/8677.breaking

@@ -0,0 +1,2 @@
+``GalilDMC4133Controller.absolute_position`` is now read-only. Setting it previously only
+updated the cached value without moving the motors, and now raises an error.

docs/changes/newsfragments/8677.improved_driver

@@ -0,0 +1,7 @@
+All queries of the Galil DMC-41x3 driver now retry the empty or bare ``:`` replies that gclib
+can return under rapid polling, instead of only ``is_in_motion`` and ``absolute_position``.
+Non-empty replies that cannot be parsed, such as an absolute position with fewer than three
+values, now raise immediately instead of being retried. ``GalilDMC4133Motor.on_off_status`` and
+``off_when_error_occurs`` now parse the whole reply instead of only its first character.
+``GalilMotionController.close`` now also closes the QCoDeS instrument, so a closed controller
+can be created again with the same name.

src/qcodes/instrument_drivers/Galil/dmc_41x3.py

@@ -36,15 +36,22 @@
 _REPLY_RETRIES = 10
 
 
+def _strip_prompt(response: str) -> str:
+    """
+    Remove whitespace and the ``:`` prompt that gclib can leave around a reply.
+    """
+    return response.strip().strip(":").strip()
+
+
 def _parse_motion_flag(response: str) -> int | None:
     """
-    Parse the reply to ``MG _BG<axis>``.
+    Parse a single reply to ``MG _BG<axis>`` while polling for motion.
 
     Under rapid polling gclib can return the bare ``:`` prompt or an empty
     string instead of a number. Return ``None`` for such transient or
-    otherwise unparsable replies so that the caller can retry.
+    otherwise unparsable replies so that the caller can poll again.
     """
-    cleaned = response.strip().strip(":").strip()
+    cleaned = _strip_prompt(response)
     if not cleaned:
         return None
     try:
@@ -53,24 +60,22 @@ def _parse_motion_flag(response: str) -> int | None:
         return None
 
 
-def _parse_position_reply(response: str) -> dict[str, int] | None:
+def _parse_position_reply(response: str) -> dict[str, int]:
     """
     Parse the reply to ``PA ?,?,?`` into a dictionary with keys ``"A"``,
     ``"B"`` and ``"C"``.
 
-    The controller returns exactly three numeric positions separated by commas and
-    sometimes terminated by the ``:`` prompt, e.g. ``"100, 200, 300"``.
-    Return ``None`` for a bare ``:`` prompt, an empty reply or a malformed
-    reply so that the caller can retry.
+    The controller returns exactly three numeric positions separated by
+    commas, e.g. ``"100, 200, 300"``.
+
+    Raises:
+        ValueError: if the reply does not contain three numbers.
+
     """
-    cleaned = response.strip().removesuffix(":").strip()
-    parts = cleaned.split(",")
+    parts = _strip_prompt(response).split(",")
     if len(parts) != 3:
-        return None
-    try:
-        a, b, c = (int(float(part)) for part in parts)
-    except (ValueError, OverflowError):
-        return None
+        raise ValueError(f"Expected three positions, got {response!r}")
+    a, b, c = (int(float(part)) for part in parts)
     return {"A": a, "B": b, "C": c}
 
 
@@ -158,8 +163,30 @@ def write_raw(self, cmd: str) -> None:
     def ask_raw(self, cmd: str) -> str:
         """
         Asks/Reads data from Galil motion controller
+
+        Under rapid polling gclib can return the bare ``:`` prompt or an
+        empty string instead of the reply. Such replies are retried a few
+        times before giving up. The prompt is removed from the reply.
+
+        Raises:
+            RuntimeError: if the controller keeps returning empty replies.
+
         """
-        return self.g.GCommand(cmd + "\r")
+        for _ in range(_REPLY_RETRIES):
+            response = self._ask_once(cmd)
+            if response:
+                return response
+            time.sleep(_POLL_INTERVAL_S)
+        raise RuntimeError(
+            f"Got only empty replies to {cmd!r} after {_REPLY_RETRIES} attempts"
+        )
+
+    def _ask_once(self, cmd: str) -> str:
+        """
+        Ask once without retrying an empty reply. The ``:`` prompt is removed
+        from the reply.
+        """
+        return _strip_prompt(self.g.GCommand(cmd + "\r"))
 
     def timeout(self, val: int) -> None:
         """
@@ -178,7 +205,13 @@ def close(self) -> None:
         """
         Close connection to the instrument
         """
-        self.g.GClose()
+        try:
+            # the connection is removed by Instrument.close, so a second close
+            # has nothing to close
+            if hasattr(self, "g"):
+                self.g.GClose()
+        finally:
+            super().close()
 
     def motion_complete(self, axes: str) -> None:
         """
@@ -446,35 +479,33 @@ def _get_off_when_error_occurs(self) -> int:
         Gets the status if motor is automatically set to turn off when error
         occurs.
         """
-        val = self.ask(f"MG _OE{self._axis}")
-
-        return int(val[0])
+        return int(float(self.ask(f"MG _OE{self._axis}")))
 
     def _set_off_when_error_occurs(self, val: int) -> None:
         """
         Sets the motor to turn off automatically when the error occurs
         """
         self.write(f"OE{self._axis}={val}")
 
-    def _set_deceleration(self, val: str) -> None:
+    def _set_deceleration(self, val: int) -> None:
         """
         Set deceleration for the motor's motion
         """
         self.write(f"DC{self._axis}={val}")
 
-    def _set_acceleration(self, val: str) -> None:
+    def _set_acceleration(self, val: int) -> None:
         """
         Set acceleration for the motor's motion
         """
         self.write(f"AC{self._axis}={val}")
 
-    def _set_speed(self, val: str) -> None:
+    def _set_speed(self, val: int) -> None:
         """
         Sets speed for motor's motion
         """
         self.write(f"SP{self._axis}={val}")
 
-    def _set_relative_position(self, val: str) -> None:
+    def _set_relative_position(self, val: int) -> None:
         """
         Sets relative position
         """
@@ -490,8 +521,7 @@ def on_off_status(self) -> str:
         """
         Tells motor on off status
         """
-        val = self.ask(f"MG _MO{self._axis}")
-        if val[0] == "1":
+        if int(float(self.ask(f"MG _MO{self._axis}"))) == 1:
             return "off"
         else:
             return "on"
@@ -512,20 +542,8 @@ def is_in_motion(self) -> int:
         """
         Checks if the motor is in motion or not. return 1, if motor is in
         motion otherwise 0
-
-        Transient empty replies of the controller are retried.
         """
-        response = ""
-        for _ in range(_REPLY_RETRIES):
-            response = self.ask(f"MG _BG{self._axis}")
-            flag = _parse_motion_flag(response)
-            if flag is not None:
-                return flag
-            time.sleep(_POLL_INTERVAL_S)
-        raise RuntimeError(
-            f"Could not parse motion flag of axis {self._axis} after retries: "
-            f"{response!r}"
-        )
+        return int(float(self.ask(f"MG _BG{self._axis}")))
 
     def wait_till_motor_motion_complete(
         self, timeout_s: float = _DEFAULT_MOTION_TIMEOUT_S
@@ -577,10 +595,10 @@ def __init__(
         )
         """sets number of decimals in the format of the position"""
 
-        self.absolute_position: Parameter = self.add_parameter(
+        self.absolute_position: Parameter[dict[str, int], Self] = self.add_parameter(
             "absolute_position",
             get_cmd=self._get_absolute_position,
-            set_cmd=None,
+            set_cmd=False,
             unit="quadrature counts",
             docstring="gets absolute position of the motors from the set origin",
         )
@@ -635,19 +653,8 @@ def _set_default_update_time(self) -> None:
     def _get_absolute_position(self) -> dict[str, int]:
         """
         Gets absolution position of the motors from the defined origin
-
-        Transient empty replies of the controller are retried.
         """
-        response = ""
-        for _ in range(_REPLY_RETRIES):
-            response = self.ask("PA ?,?,?")
-            parsed = _parse_position_reply(response)
-            if parsed is not None:
-                return parsed
-            time.sleep(_POLL_INTERVAL_S)
-        raise RuntimeError(
-            f"Could not parse absolute position reply after retries: {response!r}"
-        )
+        return _parse_position_reply(self.ask("PA ?,?,?"))
 
     def end_program(self) -> None:
         """
@@ -730,7 +737,9 @@ def wait_till_axes_motion_complete(
                         query_timeout_ms = min(query_timeout_ms, original_timeout)
                     try:
                         self.timeout(query_timeout_ms)
-                        response = self.ask(f"MG _BG{axis}")
+                        # poll once per query; an empty reply is treated as
+                        # "not yet finished" and polled again until the deadline
+                        response = self._ask_once(f"MG _BG{axis}")
                     except gclib.GclibError as e:
                         if time.monotonic() >= deadline:
                             raise TimeoutError(