diff --git a/uc2rest/galvo.py b/uc2rest/galvo.py index 46b06d8..43c9fe6 100644 --- a/uc2rest/galvo.py +++ b/uc2rest/galvo.py @@ -37,16 +37,17 @@ def set_dac(self, channel=1, frequency=1, offset=0, amplitude=1, clk_div=0, phas SCANNER ############################################################################################################################## ''' - def set_galvo_scan(self, nx=256, ny=256, x_min=500, x_max=3500, - y_min=500, y_max=3500, sample_period_us=1, - frame_count=0, bidirectional=False, - pre_samples=0, fly_samples=0, trig_delay_us=0, - trig_width_us=0, line_settle_samples=0, + def set_galvo_scan(self, nx=256, ny=256, x_min=500, x_max=3500, + y_min=500, y_max=3500, sample_period_us=1, + frame_count=0, bidirectional=False, + pre_samples=0, fly_samples=0, trig_delay_us=0, + trig_width_us=0, line_settle_samples=0, enable_trigger=1, apply_x_lut=0, + overscan_samples=0, laser_blanking=0, hw_pixel_clock=0, timeout=1): """ Start galvo scanner with new API (HighSpeedScannerCore) - + Args: nx: Number of X samples per line (default: 256) ny: Number of Y lines (default: 256) @@ -56,14 +57,36 @@ def set_galvo_scan(self, nx=256, ny=256, x_min=500, x_max=3500, y_max: Max Y position 0-4095 (default: 3500) sample_period_us: Microseconds per sample, 0=max speed (default: 1) frame_count: Number of frames, 0=infinite (default: 0) - bidirectional: Enable bidirectional scanning (default: False) + bidirectional: Enable bidirectional scanning (default: False). + Odd lines are scanned with a mirrored profile; the trigger + window stays identical, so pixel counts remain monotonic and + equidistant. The host must flip odd lines when reassembling. + trig_delay_us: Gap between the frame marker and the line marker + at frame start (both fire during pre-blanking, ahead of the + first pixel). + trig_width_us: Marker/pixel trigger pulse width in microseconds + (pixel width is capped at half the dwell time; 0 = fastest + possible pulse). + overscan_samples: Linear ramp extension (same per-pixel slope) on + both sides of the imaging window. Compensates galvo lag: the + mirror is already moving at constant velocity when triggers + and laser start (default: 0) + laser_blanking: 1 = gate the galvo laser pin HIGH only during the + imaging window (off during pre/overscan/flyback/settle) + (default: 0) + hw_pixel_clock: 1 = generate the pixel clock with the RMT + peripheral (hardware-equidistant, decoupled from the DAC/SPI + loop; ESP32-S3 only, silently falls back to software pulses + elsewhere) (default: 0) timeout: Request timeout in seconds (default: 1) - + sends: - {"task": "/galvo_act", "config": {"nx":512,"ny":512,"x_min":500,"x_max":3500,"y_min":500,"y_max":3500,"pre_samples":0,"fly_samples":0,"sample_period_us":0,"trig_delay_us":0,"trig_width_us":0,"line_settle_samples":0,"enable_trigger":1,"apply_x_lut":0,"frame_count":0,"bidirectional":true}} - + {"task": "/galvo_act", "config": {"nx":512,"ny":512,"x_min":500,"x_max":3500,"y_min":500,"y_max":3500,"pre_samples":0,"fly_samples":0,"sample_period_us":0,"trig_delay_us":0,"trig_width_us":0,"line_settle_samples":0,"enable_trigger":1,"apply_x_lut":0,"frame_count":0,"bidirectional":true,"overscan_samples":0,"laser_blanking":0,"hw_pixel_clock":0}} + Example: - >>> galvo.set_galvo_scan(nx=64, ny=64, frame_count=10, bidirectional=True) + >>> galvo.set_galvo_scan(nx=64, ny=64, frame_count=10, + ... overscan_samples=8, laser_blanking=1, + ... hw_pixel_clock=1) """ path = '/galvo_act' payload = { @@ -84,10 +107,13 @@ def set_galvo_scan(self, nx=256, ny=256, x_min=500, x_max=3500, "trig_width_us": trig_width_us, "line_settle_samples": line_settle_samples, "enable_trigger": enable_trigger, - "apply_x_lut": apply_x_lut + "apply_x_lut": apply_x_lut, + "overscan_samples": overscan_samples, + "laser_blanking": 1 if laser_blanking else 0, + "hw_pixel_clock": 1 if hw_pixel_clock else 0 } } - + return self._parent.post_json(path, payload, timeout=timeout) def stop_galvo_scan(self, timeout=1): @@ -108,6 +134,32 @@ def stop_galvo_scan(self, timeout=1): return self._parent.post_json(path, payload, timeout=timeout) + def set_position(self, x, y, timeout=1): + """ + Move the galvo to a static XY position (DAC counts 0-4095) without scanning. + + Sends the direct-positioning form of /galvo_act, which the firmware maps + to a "goto" command (single-point dwell). Useful e.g. for parking the + beam at a defined position when a scan is stopped. + + Args: + x: X DAC coordinate (0-4095) + y: Y DAC coordinate (0-4095) + timeout: Request timeout in seconds (default: 1) + + Example: + >>> galvo.set_position(2048, 2048) # park at center + """ + path = '/galvo_act' + payload = { + "task": path, + "galvo": { + "x": int(x), + "y": int(y) + } + } + return self._parent.post_json(path, payload, timeout=timeout) + def get_galvo_status(self, timeout=1): """ Get galvo scanner status @@ -178,10 +230,10 @@ def set_arbitrary_points(self, points, laser_trigger="AUTO", timeout=1): path = '/galvo_act' payload = { "task": path, - #"laser_trigger": laser_trigger, + "laser_trigger": laser_trigger, "points": points } - + return self._parent.post_json(path, payload, timeout=timeout) def stop_arbitrary_points(self, timeout=1): diff --git a/uc2rest/motor.py b/uc2rest/motor.py index 8653dfb..6c62584 100644 --- a/uc2rest/motor.py +++ b/uc2rest/motor.py @@ -55,6 +55,9 @@ def __init__(self, parent=None): self._parent.serial.register_callback(self._callback_motor_status, pattern="steppers") # Register callback for stagescan completion signal: {"stagescan":{},"qid":0,"success":1} self._parent.serial.register_callback(self._callback_stagescan_complete, pattern="stagescan") + # Register callback for async closed-loop axis fault events (design v2): + # {"axisEvent":{"axis":n,"fault":"STALL","posErrSteps":-142,...}} + self._parent.serial.register_callback(self._callback_axis_event, pattern="axisEvent") # announce a function that is called when we receive a position update through the callback self._callbackPerKey = {} self.nCallbacks = 10 @@ -64,6 +67,8 @@ def __init__(self, parent=None): # Stage scan completion state self._stagescan_complete = False self._stagescan_callbacks = [] # List of callbacks to call when stagescan completes + # Closed-loop axis fault event listeners (design v2, WP9) + self._axis_event_callbacks = [] # called with the fault dict on STALL/LOST_STEPS/… # move motor to wake them up #FIXME: Should not be necessary! #self.move_stepper(steps=(1,1,1,1), speed=(1000,1000,1000,1000), is_absolute=(False,False,False,False)) #self.move_stepper(steps=(-1,-1,-1,-1), speed=(1000,1000,1000,1000), is_absolute=(False,False,False,False)) @@ -192,8 +197,189 @@ def wait_for_stagescan_complete(self, timeout=300): def register_callback(self, key, callbackfct): ''' register a callback function for a specific key ''' self._callbackPerKey[key] = callbackfct - - + + # ======================================================================== + # Closed-loop / encoder feedback (design v2, WP9) + # + # All values are in STEPS — do NOT convert to µm here; µm conversion stays + # in the higher-level ImSwitch code. Every method is safe when the firmware + # lacks encoder support: it returns sensible defaults and never raises. + # + # The firmware exposes per-axis feedback in the CANopen AXIS OD block + # (0x2040-0x204B) on each motor slave, addressed as (node, sub-index). The + # sub-index is the slave's motor axis id (usually 1). Reach it through the + # master's generic SDO bridge (/can_act {"sdo":{...}}). + # ======================================================================== + + # AXIS OD indices (mirror lib/uc2_od + tools/canopen registry, base 0x2040) + _AXIS_MEASURED_STEPS = 0x2040 + _AXIS_POSITION_ERROR_STEPS = 0x2041 + _AXIS_MODE = 0x2042 + _AXIS_HEALTH = 0x2043 + _AXIS_FAULT = 0x2044 + _AXIS_RESET = 0x2045 + _AXIS_CALIBRATED = 0x2046 + _AXIS_REFERENCED = 0x2047 + _AXIS_CALIBRATE = 0x2048 + _AXIS_COUNTS_PER_STEP_Q16 = 0x2049 + _AXIS_BACKLASH_STEPS = 0x204A + _AXIS_RAW_COUNTS = 0x204B + + _AXIS_MODE_NAMES = {0: "OPEN_LOOP", 1: "MONITOR", 2: "CORRECT", 3: "SERVO"} + _AXIS_HEALTH_NAMES = {0: "OK", 1: "DEGRADED", 2: "FAULT"} + _AXIS_FAULT_NAMES = {0: "NONE", 1: "STALL", 2: "LOST_STEPS", 3: "DIVERGENCE", + 4: "TIMEOUT", 5: "CAL_INVALID", 6: "CAL_FAILED", 7: "ENC_NOISE"} + _AXIS_RESET_POLICIES = {"TRUST_ENCODER": 1, "TRUST_STEPS": 2, "FORCE_REHOME": 3} + + def _axis_sdo(self, node, index, sub, op, ctype="i32", value=None, timeout=2): + """Low-level SDO read/write to a slave AXIS OD entry via the master. + + Returns the integer value on a successful read, True on a successful + write, and None on any failure (never raises), so callers can degrade + gracefully when the firmware has no encoder/axis support. + """ + sdo = {"node": int(node), "index": int(index), "sub": int(sub), + "op": op, "type": ctype} + if op == "w": + sdo["value"] = int(value) + payload = {"task": "/can_act", "sdo": sdo} + try: + resp = self._parent.post_json("/can_act", payload, getReturn=True, + timeout=timeout, nResponses=1) + except Exception as e: + self._parent.logger.debug(f"axis SDO transport error: {e}") + return None + if not isinstance(resp, dict) or resp.get("status") != "ok": + return None + return True if op == "w" else resp.get("value", None) + + def _callback_axis_event(self, data): + """Serial callback for async axis fault events from firmware: + {"axisEvent":{"axis":n,"fault":"STALL","posErrSteps":-142, + "commandedSteps":..,"measuredSteps":..,"node":..}} + Forwards the inner fault dict to every registered listener so ImSwitch + can react to STALL / LOST_STEPS / DIVERGENCE during long acquisitions. + """ + try: + event = data.get("axisEvent", data) if isinstance(data, dict) else data + for callback in list(self._axis_event_callbacks): + try: + callback(event) + except Exception as e: + self._parent.logger.error(f"Error in axis event callback: {e}") + except Exception as e: + print(f"Error in _callback_axis_event: {e}") + + def register_axis_event_callback(self, callback): + """Register a callback invoked with the fault dict on every axis event.""" + if callback not in self._axis_event_callbacks: + self._axis_event_callbacks.append(callback) + + def unregister_axis_event_callback(self, callback): + """Unregister a previously registered axis-event callback.""" + if callback in self._axis_event_callbacks: + self._axis_event_callbacks.remove(callback) + + def setAxisMode(self, node, mode, axis=1): + """Set the closed-loop mode of a motor-slave axis. + + :param node: CAN node id of the motor slave + :param mode: 0/1/2/3 or "OPEN_LOOP"/"MONITOR"/"CORRECT"/"SERVO" + :param axis: OD sub-index (the slave's motor axis id, usually 1) + :return: True on success, False otherwise (never raises) + """ + if isinstance(mode, str): + inv = {v: k for k, v in self._AXIS_MODE_NAMES.items()} + mode = inv.get(mode.upper(), 0) + ok = self._axis_sdo(node, self._AXIS_MODE, axis, "w", "u8", int(mode)) + return ok is True + + def getAxisFeedback(self, node, axis=1): + """Read a motor-slave axis's live feedback (all values in STEPS). + + :return: dict with commandedSteps, measuredSteps, positionErrorSteps, + mode, health, fault (ints + *_name), calibrated, referenced, + rawCounts. When the firmware has no encoder support the read fails + and OPEN_LOOP / zero-error defaults are returned; never raises. + """ + measured = self._axis_sdo(node, self._AXIS_MEASURED_STEPS, axis, "r", "i32") + if measured is None: + # No encoder / no axis support — uniform, non-raising default. + return {"commandedSteps": 0, "measuredSteps": 0, "positionErrorSteps": 0, + "mode": 0, "mode_name": "OPEN_LOOP", "health": 0, "health_name": "OK", + "fault": 0, "fault_name": "NONE", "calibrated": False, + "referenced": False, "rawCounts": 0} + posErr = self._axis_sdo(node, self._AXIS_POSITION_ERROR_STEPS, axis, "r", "i32") or 0 + mode = self._axis_sdo(node, self._AXIS_MODE, axis, "r", "u8") or 0 + health = self._axis_sdo(node, self._AXIS_HEALTH, axis, "r", "u8") or 0 + fault = self._axis_sdo(node, self._AXIS_FAULT, axis, "r", "u8") or 0 + calibrated = self._axis_sdo(node, self._AXIS_CALIBRATED, axis, "r", "u8") or 0 + referenced = self._axis_sdo(node, self._AXIS_REFERENCED, axis, "r", "u8") or 0 + rawCounts = self._axis_sdo(node, self._AXIS_RAW_COUNTS, axis, "r", "i32") or 0 + return { + "commandedSteps": int(measured) - int(posErr), + "measuredSteps": int(measured), + "positionErrorSteps": int(posErr), + "mode": int(mode), "mode_name": self._AXIS_MODE_NAMES.get(int(mode), "?"), + "health": int(health), "health_name": self._AXIS_HEALTH_NAMES.get(int(health), "?"), + "fault": int(fault), "fault_name": self._AXIS_FAULT_NAMES.get(int(fault), "?"), + "calibrated": bool(calibrated), "referenced": bool(referenced), + "rawCounts": int(rawCounts), + } + + def resetAxis(self, node, policy="TRUST_ENCODER", axis=1): + """Clear a latched fault and re-establish the axis origin. + + :param policy: "TRUST_ENCODER"|"TRUST_STEPS"|"FORCE_REHOME" or 1/2/3 + :return: True on success (never raises) + """ + if isinstance(policy, str): + policy = self._AXIS_RESET_POLICIES.get(policy.upper(), 1) + ok = self._axis_sdo(node, self._AXIS_RESET, axis, "w", "u8", int(policy)) + return ok is True + + def getAxisCalibration(self, node, axis=1): + """Read the persisted calibration of a motor-slave axis. + + :return: dict with countsPerStepQ16 (signed Q16.16), backlashSteps, + calibrated. None fields default to 0/False; never raises. + """ + cps = self._axis_sdo(node, self._AXIS_COUNTS_PER_STEP_Q16, axis, "r", "i32") + backlash = self._axis_sdo(node, self._AXIS_BACKLASH_STEPS, axis, "r", "i32") + calibrated = self._axis_sdo(node, self._AXIS_CALIBRATED, axis, "r", "u8") + return { + "countsPerStepQ16": int(cps) if cps is not None else 0, + "countsPerStep": (int(cps) / 65536.0) if cps is not None else 0.0, + "backlashSteps": int(backlash) if backlash is not None else 0, + "calibrated": bool(calibrated) if calibrated is not None else False, + } + + def calibrateAxis(self, node, axis=1, wait=True, timeout=30, **params): + """Trigger the firmware sign/scale/backlash calibration routine. + + The routine runs ON the slave (it moves the motor through probe moves); + `params` are accepted for API forward-compatibility but the firmware + currently uses its built-in defaults. With ``wait=True`` this polls until + the axis reports calibrated (or ``timeout`` s) and returns the resulting + calibration dict; otherwise returns {"status": "triggered"}. + """ + ok = self._axis_sdo(node, self._AXIS_CALIBRATE, axis, "w", "u8", 1) + if ok is not True: + return {"status": "error", "error": "calibration trigger failed (no encoder?)"} + if not wait: + return {"status": "triggered"} + start = time.time() + while time.time() - start < timeout: + cal = self.getAxisCalibration(node, axis) + fb = self.getAxisFeedback(node, axis) + if cal["calibrated"] or fb["fault_name"] in ("CAL_FAILED",): + cal["status"] = "ok" if cal["calibrated"] else "failed" + cal["fault"] = fb["fault_name"] + return cal + time.sleep(0.5) + return {"status": "timeout"} + + def setTrigger(self, axis="X", pin=1, offset=0, period=1): # {"task": "/motor_act", "setTrig": {"steppers": [{"stepperid": 1, "trigPin": 1, "trigOff":0, "trigPer":1}]}} if type(axis) is not int: