diff --git a/.gitignore b/.gitignore index 15201ac..9dda065 100644 --- a/.gitignore +++ b/.gitignore @@ -160,12 +160,9 @@ dmypy.json # Cython debug symbols cython_debug/ -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ +# IDE +.idea/ +.vscode # PyPI configuration file .pypirc diff --git a/README.md b/README.md index 3431485..a9b5a63 100644 --- a/README.md +++ b/README.md @@ -39,19 +39,19 @@ for device_id, desc in devices: device_key = ("192.168.29.100", 10005, 2) print("Now on device 2:", controller.get_idn(device_key)) -# Check if axis '1' is referenced -is_referenced = controller.is_controller_referenced(device_key, '1') +# Check if axis '1' is referenced (homed) +is_referenced = controller.is_homed(device_key, '1') print("Axis 1 referenced:", is_referenced) -# Perform a reference move (home) on axis '1' -success = controller.reference_move(device_key, '1', method="FRF", blocking=True, timeout=30) +# Perform a home (reference move) on axis '1' +success = controller.home(device_key, '1', method="FRF", blocking=True, timeout=30) if success: - print("Reference move completed successfully.") + print("Home completed successfully.") else: - print("Reference move failed or timed out.") + print("Home failed or timed out.") # Move axis 1 to position 12.0 -controller.set_position(device_key, '1', 12.0) +controller.set_pos(12.0, device_key, '1') # Save current position as "home" controller.set_named_position(device_key, '1', 'home') @@ -72,11 +72,11 @@ controller.disconnect_all() | `get_idn(device_key)` | Get the controller identification string | | `get_serial_number(device_key)` | Get the serial number from the IDN | | `get_axes(device_key)` | Return available axes | -| `get_position(device_key, axis_index)` | Get current position of axis by index | -| `servo_status(device_key, axis)` | Check if the servo on an axis is enabled | +| `get_pos(device_key, axis)` | Get current position of axis | +| `is_loop_closed(device_key, axis)` | Check if the servo on an axis is enabled | | `get_error_code(device_key)` | Get the controller's last error code | | `halt_motion(device_key)` | Stop all motion on the controller | -| `set_position(device_key, axis, position)` | Move an axis to a position | +| `set_pos(pos, device_key, axis)` | Move an axis to a position | | `set_named_position(device_key, axis, name)` | Save a position under a named label | | `go_to_named_position(name)` | Move to a previously saved named position | | `disconnect_device(device_key)` | Disconnect a single device | @@ -84,9 +84,9 @@ controller.disconnect_all() ## Logging -By default, the controller logs info and error messages to the console. You can suppress logs (except warnings/errors) by passing quiet=True: +By default, the controller logs info and error messages to the console. You can disable logging by passing log=False: ```python -controller = PIControllerBase(quiet=True) +controller = PIControllerBase(log=False) ``` ## 🧪 Testing diff --git a/pi_controller.py b/pi_controller.py index 0815a9e..1d3baa7 100644 --- a/pi_controller.py +++ b/pi_controller.py @@ -4,42 +4,56 @@ import json import os import time -import logging from pipython import GCSDevice, GCSError - - -class PIControllerBase: # pylint: disable=too-many-public-methods +from hardware_device_base import HardwareMotionBase + + +# Status codes +class PIStatus: + """Status codes for PI controller operations.""" + OK = 0 + CONNECTED = 1 + MOVING = 2 + HOMED = 3 + + # Error codes (negative) + ERROR_NOT_CONNECTED = -1 + ERROR_CONNECTION_FAILED = -2 + ERROR_DEVICE_NOT_FOUND = -3 + ERROR_POSITION = -4 + ERROR_SERVO = -5 + ERROR_REFERENCE = -6 + ERROR_LIMITS = -7 + ERROR_TIMEOUT = -8 + ERROR_COMMAND = -9 + ERROR_INVALID_METHOD = -10 + + +class PIControllerBase(HardwareMotionBase): """ Base class for communicating with PI (Physik Instrumente) motion controllers daisy-chained over TCP/IP via a terminal server. """ - def __init__(self, quiet=False): + def __init__(self, log: bool = True, logfile: str = None): """ - Initialize the controller, set up logging, and prepare device storage. + Initialize the controller and set up logging. """ + # set up logging + if logfile is None: + logfile = self.__class__.__module__.rsplit(".", 1)[-1] + super().__init__(log, logfile) + self.devices = {} # {(ip, port, device_id): GCSDevice instance} self.daisy_chains = {} # {(ip, port): [(device_id, desc)]} - self.connected = False self.named_position_file = "config/pi_named_positions.json" - - # Logging - logfile = __name__.rsplit('.', 1)[-1] + '.log' - self.logger = logging.getLogger(logfile) - self.logger.setLevel(logging.INFO) - formatter = logging.Formatter( - '%(asctime)s - %(name)s - %(levelname)s - %(message)s' - ) - if not quiet: - file_handler = logging.FileHandler(logfile) - file_handler.setFormatter(formatter) - self.logger.addHandler(file_handler) def _require_connection(self): """ Raise an error if not connected to any device. """ - if not self.connected: + if not self.is_connected(): + self.report_error("Controller is not connected", PIStatus.ERROR_NOT_CONNECTED) raise RuntimeError("Controller is not connected") def connect_tcp(self, ip_address, port=50000): @@ -49,8 +63,8 @@ def connect_tcp(self, ip_address, port=50000): device = GCSDevice() device.ConnectTCPIP(ip_address, port) self.devices[(ip_address, port, 1)] = device - self.connected = True - self.logger.info("Connected to single PI controller at %s:%s", ip_address, port) + self._set_connected(True) + self.report_info(f"Connected to single PI controller at {ip_address}:{port}", PIStatus.CONNECTED) def connect_tcpip_daisy_chain(self, ip_address, port, blocking=True): """ @@ -67,6 +81,7 @@ def connect_tcpip_daisy_chain(self, ip_address, port, blocking=True): available.append((index, desc)) if not available: + self.report_error(f"No connected devices found at {ip_address}:{port}", PIStatus.ERROR_DEVICE_NOT_FOUND) raise RuntimeError(f"No connected devices found at {ip_address}:{port}") self.daisy_chains[(ip_address, port)] = available @@ -79,9 +94,9 @@ def connect_tcpip_daisy_chain(self, ip_address, port, blocking=True): dev.ConnectDaisyChainDevice(device_id, dcid) self.devices[(ip_address, port, device_id)] = dev - self.logger.info("[{ip}:{port}] Connected to device %s: %s", device_id, desc) + self.report_info(f"[{ip_address}:{port}] Connected to device {device_id}: {desc}", PIStatus.CONNECTED) - self.connected = True + self._set_connected(True) if blocking: # Wait until all devices are ready @@ -95,9 +110,9 @@ def disconnect_device(self, device_key): if device_key in self.devices: self.devices[device_key].CloseConnection() del self.devices[device_key] - self.logger.info("Disconnected device %s", device_key) + self.report_info(f"Disconnected device {device_key}", PIStatus.OK) if not self.devices: - self.connected = False + self._set_connected(False) def disconnect_all(self): """ @@ -105,10 +120,10 @@ def disconnect_all(self): """ for device_key in list(self.devices.keys()): self.devices[device_key].CloseConnection() - self.logger.info("Disconnected device %s", device_key) + self.report_info(f"Disconnected device {device_key}", PIStatus.OK) self.devices.clear() - self.connected = False - self.logger.info("Disconnected from all PI controllers") + self._set_connected(False) + self.report_info("Disconnected from all PI controllers", PIStatus.OK) def list_devices_on_chain(self, ip_address, port): """ @@ -118,12 +133,6 @@ def list_devices_on_chain(self, ip_address, port): raise ValueError(f"No daisy chain found at {ip_address}:{port}") return self.daisy_chains[(ip_address, port)] - def is_connected(self) -> bool: - """ - Check if the controller is connected to any device. - """ - return self.connected - def get_idn(self, device_key) -> str: """ Return the identification string for the specified device. @@ -145,27 +154,49 @@ def get_axes(self, device_key): self._require_connection() return self.devices[device_key].axes - def get_position(self, device_key, axis): + def get_pos(self, device_key=None, axis=None): """ - Return the position of the specified axis for the given device. + Get the position of the hardware motion device. + + Args: + device_key: Tuple (ip, port, device_id) identifying the device + axis: Axis to get position for + + Returns: + Position value or None if error """ + if device_key is None or axis is None: + self.report_error("device_key and axis are required for get_pos", PIStatus.ERROR_POSITION) + return None + self._require_connection() device = self.devices[device_key] try: return device.qPOS(axis)[axis] except (GCSError, IndexError) as ex: - self.logger.error("Error getting position: %s", ex) + self.report_error(f"Error getting position: {ex}", PIStatus.ERROR_POSITION) return None - def servo_status(self, device_key, axis): + def is_loop_closed(self, device_key=None, axis=None) -> bool: """ - Return True if the servo for the given axis is enabled, False otherwise. + Check if the hardware motion device servo is enabled (closed loop). + + Args: + device_key: Tuple (ip, port, device_id) identifying the device + axis: Axis to check + + Returns: + True if servo is enabled, False otherwise """ + if device_key is None or axis is None: + self.report_error("device_key and axis are required for is_loop_closed", PIStatus.ERROR_SERVO) + return False + self._require_connection() try: return bool(self.devices[device_key].qSVO(axis)[axis]) except GCSError as ex: - self.logger.error("Error checking servo status: %s", ex) + self.report_error(f"Error checking servo status: {ex}", PIStatus.ERROR_SERVO) return False def get_error_code(self, device_key): @@ -176,7 +207,7 @@ def get_error_code(self, device_key): try: return self.devices[device_key].qERR() except GCSError as ex: - self.logger.error("Error getting error code: %s", ex) + self.report_error(f"Error getting error code: {ex}", PIStatus.ERROR_COMMAND) return None def halt_motion(self, device_key): @@ -187,22 +218,43 @@ def halt_motion(self, device_key): try: self.devices[device_key].HLT() except GCSError as ex: - self.logger.error("Error halting motion: %s", ex) + self.report_error(f"Error halting motion: {ex}", PIStatus.ERROR_POSITION) - def set_position(self, device_key, axis, position, blocking=True): + def set_pos(self, pos, device_key=None, axis=None, blocking=True, timeout=20) -> bool: """ - Move the specified axis to the given position for the specified device. - If blocking=True, wait until move is complete. + Set the position. + + Args: + pos: Target position + device_key: Tuple (ip, port, device_id) identifying the device + axis: Axis to move + blocking: If True, wait until move is complete + timeout: Timeout in seconds for blocking operation + + Returns: + True if successful, False otherwise """ + if device_key is None or axis is None: + self.report_error("device_key and axis are required for set_pos", PIStatus.ERROR_POSITION) + return False + self._require_connection() try: - self.devices[device_key].MOV(axis, position) + self.devices[device_key].MOV(axis, pos) if blocking: + start_time = time.time() while self.is_moving(device_key, axis): + if time.time() - start_time > timeout: + self.report_error( + f"Move to position {pos} timed out after {timeout} seconds on axis {axis}", + PIStatus.ERROR_TIMEOUT + ) + return False time.sleep(0.1) - + return True except GCSError as ex: - self.logger.error("Error setting position: %s", ex) + self.report_error(f"Error setting position: {ex}", PIStatus.ERROR_POSITION) + return False def set_named_position(self, device_key, axis, name): """ @@ -213,10 +265,10 @@ def set_named_position(self, device_key, axis, name): try: pos = device.qMOV(axis)[axis] except (GCSError, OSError, ValueError): - pos = self.get_position(device_key, axis) + pos = self.get_pos(device_key, axis) if pos is None: - self.logger.warning("Could not get position for axis %s", axis) + self.report_warning(f"Could not get position for axis {axis}", PIStatus.ERROR_POSITION) return serial = self.get_serial_number(device_key) @@ -227,8 +279,8 @@ def set_named_position(self, device_key, axis, name): try: positions = json.load(file) except json.JSONDecodeError: - self.logger.warning( - "Could not parse JSON from %s", self.named_position_file + self.report_warning( + f"Could not parse JSON from {self.named_position_file}", PIStatus.ERROR_POSITION ) if serial not in positions: @@ -239,19 +291,25 @@ def set_named_position(self, device_key, axis, name): with open(self.named_position_file, "w") as file: json.dump(positions, file, indent=2) - self.logger.info( - "Saved position '%s' for controller %s, axis %s: %s", name, serial, axis, pos + self.report_info( + f"Saved position '{name}' for controller {serial}, axis {axis}: {pos}", PIStatus.OK ) - def go_to_named_position(self, device_key, name, blocking=True): + def go_to_named_position(self, device_key, name, blocking=True, timeout=20): """ Move the specified device's axis to a previously saved named position. + + Args: + device_key: Tuple (ip, port, device_id) identifying the device + name: Name of the saved position + blocking: If True, wait until move is complete + timeout: Timeout in seconds for blocking operation """ serial = self.get_serial_number(device_key) if not os.path.exists(self.named_position_file): - self.logger.warning( - "Named positions file not found: %s", self.named_position_file + self.report_warning( + f"Named positions file not found: {self.named_position_file}", PIStatus.ERROR_POSITION ) return @@ -259,25 +317,25 @@ def go_to_named_position(self, device_key, name, blocking=True): with open(self.named_position_file, "r") as file: positions = json.load(file) except json.JSONDecodeError: - self.logger.warning( - "Failed to read positions from %s", self.named_position_file + self.report_warning( + f"Failed to read positions from {self.named_position_file}", PIStatus.ERROR_POSITION ) return if serial not in positions: - self.logger.warning("No named positions found for controller %s", serial) + self.report_warning(f"No named positions found for controller {serial}", PIStatus.ERROR_POSITION) return if name not in positions[serial]: - self.logger.warning( - "Named position '%s' not found for controller %s", name, serial + self.report_warning( + f"Named position '{name}' not found for controller {serial}", PIStatus.ERROR_POSITION ) return axis, pos = positions[serial][name] - self.set_position(device_key, axis, pos, blocking) - self.logger.info( - "Moved axis %s to named position '%s' for controller %s: %s", axis, name, serial, pos + self.set_pos(pos, device_key, axis, blocking, timeout) + self.report_info( + f"Moved axis {axis} to named position '{name}' for controller {serial}: {pos}", PIStatus.OK ) def is_moving(self, device_key, axis): @@ -285,43 +343,102 @@ def is_moving(self, device_key, axis): self._require_connection() return self.devices[device_key].IsMoving(axis)[axis] - def set_servo(self, device_key, axis, enable=True): - """Open (enable) or close (disable) servo loop.""" - self._require_connection() - return self.devices[device_key].SVO(axis, int(enable)) + def close_loop(self, device_key=None, axis=None, enable=True) -> bool: + """ + Close the loop (enable servo) for the hardware motion device. + Can also be used to open the loop (disable servo) by setting enable=False. + + Args: + device_key: Tuple (ip, port, device_id) identifying the device + axis: Axis to enable servo for + enable: True to close loop (enable servo), False to open loop (disable servo) + + Returns: + True if successful, False otherwise + """ + if device_key is None or axis is None: + self.report_error("device_key and axis are required for close_loop", PIStatus.ERROR_SERVO) + return False - def get_limit_min(self, device_key, axis): - """Query stage minimum limit.""" self._require_connection() - return self.devices[device_key].qTMN(axis)[axis] + try: + self.devices[device_key].SVO(axis, int(enable)) + self.report_info(f"{'Closed (enabled servo)' if enable else 'Opened (disabled servo)'} loop for device {device_key}, axis {axis}", PIStatus.OK) + return True + except (GCSError, RuntimeError) as ex: + self.report_error(f"Error {'closing' if enable else 'opening'} loop: {ex}", PIStatus.ERROR_SERVO) + return False + + def get_limits(self, device_key=None, axis=None): + """ + Get the limits of the hardware motion device. + + Args: + device_key: Tuple (ip, port, device_id) identifying the device + axis: Axis to get limits for + + Returns: + Dictionary with axis as key and (min, max) tuple as value, or None if error + """ + if device_key is None or axis is None: + self.report_error("device_key and axis are required for get_limits", PIStatus.ERROR_LIMITS) + return None - def get_limit_max(self, device_key, axis): - """Query stage maximum limit.""" self._require_connection() - return self.devices[device_key].qTMX(axis)[axis] + try: + min_limit = self.devices[device_key].qTMN(axis)[axis] + max_limit = self.devices[device_key].qTMX(axis)[axis] + return {axis: (min_limit, max_limit)} + except (GCSError, RuntimeError) as ex: + self.report_error(f"Error getting limits: {ex}", PIStatus.ERROR_LIMITS) + return None def is_controller_ready(self, device_key): """Check if stage/controller is ready.""" self._require_connection() return self.devices[device_key].IsControllerReady() - def is_controller_referenced(self, device_key, axis): - """Check reference/home state for axis.""" + def is_homed(self, device_key=None, axis=None) -> bool: + """ + Check if the hardware motion device is homed (referenced). + + Args: + device_key: Tuple (ip, port, device_id) identifying the device + axis: Axis to check + + Returns: + True if axis is referenced, False otherwise + """ + if device_key is None or axis is None: + self.report_error("device_key and axis are required for is_homed", PIStatus.ERROR_REFERENCE) + return False + self._require_connection() return self.devices[device_key].qFRF(axis)[axis] - def reference_move(self, device_key, axis, method="FRF", blocking=True, timeout=30): # pylint:disable=too-many-arguments + def home(self, device_key=None, axis=None, method="FRF", blocking=True, timeout=20) -> bool: # pylint:disable=too-many-arguments """ - Execute a reference/home move (FRF, FNL, FPL). - method: which command to use ("FRF", "FNL", "FPL") - blocking: if True, wait until move is complete - Returns True if successful, False otherwise. + Home the hardware motion device. + + Args: + device_key: Tuple (ip, port, device_id) identifying the device + axis: Axis to home + method: Reference method to use ("FRF", "FNL", "FPL") + blocking: If True, wait until homing is complete + timeout: Timeout in seconds for blocking operation + + Returns: + True if successful, False otherwise """ + if device_key is None or axis is None: + self.report_error("device_key and axis are required for home", PIStatus.ERROR_REFERENCE) + return False + self._require_connection() allowed_methods = {"FRF", "FNL", "FPL"} if method not in allowed_methods: - self.logger.error( - "Invalid reference method: %s. Must be one of %s", method, allowed_methods + self.report_error( + f"Invalid reference method: {method}. Must be one of {allowed_methods}", PIStatus.ERROR_INVALID_METHOD ) return False @@ -329,27 +446,122 @@ def reference_move(self, device_key, axis, method="FRF", blocking=True, timeout= # Check if the device supports the specified method if not getattr(device, "Has%s", method)(): - self.logger.error("Device %s does not support method '%s'", device_key, method) + self.report_error(f"Device {device_key} does not support method '{method}'", PIStatus.ERROR_INVALID_METHOD) return False try: getattr(device, method)(axis) - self.logger.info( - "Started reference move '%s' on axis %s (device %s)", method, axis, device_key + self.report_info( + f"Started reference move '{method}' on axis {axis} (device {device_key})", PIStatus.HOMED ) if blocking: start_time = time.time() while self.is_moving(device_key, axis): if time.time() - start_time > timeout: - self.logger.error( - "Reference move timed out after %s seconds on axis %s", timeout, axis + self.report_error( + f"Reference move timed out after {timeout} seconds on axis {axis}", PIStatus.ERROR_TIMEOUT ) return False time.sleep(0.1) return True except (GCSError, OSError, ValueError) as ex: - self.logger.error( - "Error during reference move '%s' on axis %s: %s", method, axis, ex + self.report_error( + f"Error during reference move '{method}' on axis {axis}: {ex}", PIStatus.ERROR_REFERENCE ) return False + + def connect(self, *args, **kwargs) -> None: + """ + Establish a connection to PI controller(s). + + This method supports multiple connection modes: + 1. TCP single device: connect(ip_address, port) + 2. TCP daisy chain: connect(ip_address, port, daisy_chain=True, blocking=True) + + Examples: + controller.connect("192.168.1.100", 50000) + controller.connect("192.168.1.100", 50000, daisy_chain=True) + """ + if len(args) < 2: + self.report_error("connect requires at least ip_address and port", PIStatus.ERROR_CONNECTION_FAILED) + return + + ip_address = args[0] + port = args[1] + daisy_chain = kwargs.get('daisy_chain', False) + blocking = kwargs.get('blocking', True) + + try: + if daisy_chain: + self.connect_tcpip_daisy_chain(ip_address, port, blocking=blocking) + else: + self.connect_tcp(ip_address, port) + + self._set_connected(True) + self.report_info(f"Successfully connected to PI controller at {ip_address}:{port}", PIStatus.CONNECTED) + except (GCSError, RuntimeError) as ex: + self.report_error(f"Failed to connect: {ex}", PIStatus.ERROR_CONNECTION_FAILED) + self._set_connected(False) + + def disconnect(self) -> None: + """ + Disconnect from all PI controller devices. + """ + try: + self.disconnect_all() + self._set_connected(False) + except Exception as ex: + self.report_error(f"Error during disconnect: {ex}", PIStatus.ERROR_CONNECTION_FAILED) + + def _send_command(self, command, device_key=None) -> bool: + """ + Send a raw GCS command to the specified device. + + Args: + command: GCS command string to send + device_key: Tuple (ip, port, device_id) identifying the device + + Returns: + True if command was sent successfully, False otherwise + """ + if device_key is None: + self.report_error("device_key is required for _send_command", PIStatus.ERROR_COMMAND) + return False + + self._require_connection() + + try: + with self.lock: + device = self.devices[device_key] + device.send(command) + self.logger.debug(f"Sent command to device {device_key}: {command}") + return True + except (GCSError, KeyError) as ex: + self.report_error(f"Error sending command: {ex}", PIStatus.ERROR_COMMAND) + return False + + def _read_reply(self, device_key=None): + """ + Read a reply from the specified device. + + Args: + device_key: Tuple (ip, port, device_id) identifying the device + + Returns: + Reply string or None if error + """ + if device_key is None: + self.report_error("device_key is required for _read_reply", PIStatus.ERROR_COMMAND) + return None + + self._require_connection() + + try: + device = self.devices[device_key] + reply = device.read() + self.logger.debug(f"Read reply from device {device_key}: {reply}") + return reply + except (GCSError, KeyError) as ex: + self.report_error(f"Error reading reply: {ex}", PIStatus.ERROR_COMMAND) + return None diff --git a/pyproject.toml b/pyproject.toml index b28c9e0..9c304f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,11 +3,11 @@ requires = ["setuptools>=61"] build-backend = "setuptools.build_meta" [project] -# Basic information about your project. name = "pi" version = "0.1.0" dependencies = [ - "pipython" + "pipython", + "hardware_device_base@git+https://github.com/COO-Utilities/hardware_device_base" ] requires-python = ">=3.9" authors = [ @@ -19,9 +19,6 @@ maintainers = [ description = "Low-level Python library to control PI 863 and PI 663 motion controllers using the [pipython](https://pypi.org/project/pipython/) library." readme = "README.md" license = "MIT" -classifiers = [ - "Programming Language :: Python" -] [project.urls] # Various URLs related to your project. These links are displayed on PyPI. @@ -42,9 +39,8 @@ dev = [ [tool.pytest.ini_options] testpaths = ["tests"] -[tool.setuptools.packages.find] -where = ["."] - [tool.setuptools] -py-modules = ["pi"] +packages = ["pi"] +[tool.setuptools.package-dir] +pi = "." diff --git a/tests/test_pi_basic.py b/tests/test_pi_basic.py index a172dcf..c1e8cce 100644 --- a/tests/test_pi_basic.py +++ b/tests/test_pi_basic.py @@ -7,8 +7,8 @@ def test_initialization(): """Test that the PIControllerBase initializes correctly.""" - controller = PIControllerBase() - assert not controller.connected + controller = PIControllerBase(log=False) + assert not controller.is_connected() def test_connection_fail(): diff --git a/tests/test_pi_mock.py b/tests/test_pi_mock.py index 5146efb..1051e87 100644 --- a/tests/test_pi_mock.py +++ b/tests/test_pi_mock.py @@ -17,11 +17,11 @@ def test_connect_tcpip_daisy_chain(mock_gcs_device_cls): mock_device.dcid = 1 mock_gcs_device_cls.return_value = mock_device - controller = PIControllerBase() + controller = PIControllerBase(log=False) controller.connect_tcpip_daisy_chain("192.168.29.100", 10003) mock_device.OpenTCPIPDaisyChain.assert_called_once_with("192.168.29.100", 10003) - assert controller.connected + assert controller.is_connected() assert controller.daisy_chains[("192.168.29.100", 10003)] == [ (1, "PI Device 1"), (2, "PI Device 2"), @@ -34,7 +34,7 @@ def test_connect_tcpip_daisy_chain(mock_gcs_device_cls): def test_list_devices_on_chain(): """Test listing devices on a daisy chain.""" - controller = PIControllerBase(quiet=True) + controller = PIControllerBase(log=False) ip_port = ("192.168.29.100", 10003) controller.daisy_chains[ip_port] = [(1, "PI Device 1"), (2, "PI Device 2")] @@ -48,38 +48,38 @@ def test_connect_disconnect_device(mock_gcs_device_cls): mock_device = MagicMock() mock_gcs_device_cls.return_value = mock_device - controller = PIControllerBase(quiet=True) + controller = PIControllerBase(log=False) controller.connect_tcp("127.0.0.1", 50000) device_key = ("127.0.0.1", 50000, 1) mock_device.ConnectTCPIP.assert_called_once_with("127.0.0.1", 50000) - assert controller.connected + assert controller.is_connected() controller.disconnect_device(device_key) mock_device.CloseConnection.assert_called_once() - assert not controller.connected + assert not controller.is_connected() def test_disconnect_all(): """Test disconnecting all devices.""" mock_device1 = MagicMock() mock_device2 = MagicMock() - controller = PIControllerBase(quiet=True) + controller = PIControllerBase(log=False) controller.devices[("ip", 1, 1)] = mock_device1 controller.devices[("ip", 1, 2)] = mock_device2 - controller.connected = True + controller._set_connected(True) controller.disconnect_all() mock_device1.CloseConnection.assert_called_once() mock_device2.CloseConnection.assert_called_once() assert not controller.devices - assert not controller.connected + assert not controller.is_connected() def test_get_serial_number(): """Test getting the serial number of a device.""" - controller = PIControllerBase(quiet=True) - controller.connected = True + controller = PIControllerBase(log=False) + controller._set_connected(True) device_key = ("ip", 1, 1) device = MagicMock() device.qIDN.return_value = "PI,Model,123456,1.0.0" @@ -90,28 +90,28 @@ def test_get_serial_number(): @patch("pi.pi_controller.GCSDevice") -def test_get_position(mock_gcs_device_cls): +def test_get_pos(mock_gcs_device_cls): """Test getting the position of an axis.""" mock_device = MagicMock() mock_gcs_device_cls.return_value = mock_device - controller = PIControllerBase(quiet=True) - controller.connected = True + controller = PIControllerBase(log=False) + controller._set_connected(True) device_key = ("ip", 1, 1) mock_device.axes = ["1", "2"] mock_device.qPOS.return_value = {"1": 42.0} controller.devices[device_key] = mock_device - pos = controller.get_position(device_key, "1") + pos = controller.get_pos(device_key, "1") assert pos == 42.0 mock_device.qPOS.assert_called_once_with("1") def test_set_named_position(tmp_path): """Test setting a named position.""" - controller = PIControllerBase(quiet=True) + controller = PIControllerBase(log=False) controller.named_position_file = tmp_path / "positions.json" - controller.connected = True + controller._set_connected(True) device_key = ("ip", 1, 1) device = MagicMock() device.axes = ["1"] @@ -132,14 +132,14 @@ def test_set_named_position(tmp_path): def test_go_to_named_position(tmp_path): """Test going to a named position.""" - controller = PIControllerBase(quiet=True) + controller = PIControllerBase(log=False) controller.named_position_file = tmp_path / "positions.json" - controller.connected = True + controller._set_connected(True) device_key = ("ip", 1, 1) device = MagicMock() controller.devices[device_key] = device controller.get_serial_number = MagicMock(return_value="123456") - controller.set_position = MagicMock() + controller.set_pos = MagicMock() # Prepare a named position file named_positions = {"123456": {"home": ["1", 42.0]}} @@ -147,25 +147,25 @@ def test_go_to_named_position(tmp_path): json.dump(named_positions, file) controller.go_to_named_position(device_key, "home", blocking=False) - controller.set_position.assert_called_once_with(device_key, "1", 42.0, False) + controller.set_pos.assert_called_once_with(42.0, device_key, "1", False, 20) @patch("pi.pi_controller.GCSDevice") -def test_reference_move_success(mock_gcs_device_cls): - """Test successful reference move.""" +def test_home_success(mock_gcs_device_cls): + """Test successful home (reference move).""" mock_device = MagicMock() mock_device.IsMoving.return_value = {"1": False} mock_gcs_device_cls.return_value = mock_device - controller = PIControllerBase(quiet=True) - controller.connected = True + controller = PIControllerBase(log=False) + controller._set_connected(True) device_key = ("ip", 1, 1) controller.devices[device_key] = mock_device # Test allowed method for method in ["FRF", "FNL", "FPL"]: getattr(mock_device, method).reset_mock() - result = controller.reference_move( + result = controller.home( device_key, "1", method=method, blocking=True, timeout=1 ) assert result is True @@ -173,35 +173,35 @@ def test_reference_move_success(mock_gcs_device_cls): @patch("pi.pi_controller.GCSDevice") -def test_reference_move_invalid_method(mock_gcs_device_cls): - """Test reference move with an invalid method.""" +def test_home_invalid_method(mock_gcs_device_cls): + """Test home with an invalid method.""" mock_device = MagicMock() mock_gcs_device_cls.return_value = mock_device - controller = PIControllerBase(quiet=True) - controller.connected = True + controller = PIControllerBase(log=False) + controller._set_connected(True) device_key = ("ip", 1, 1) controller.devices[device_key] = mock_device # Test disallowed method - result = controller.reference_move(device_key, "1", method="INVALID", blocking=True) + result = controller.home(device_key, "1", method="INVALID", blocking=True) assert result is False @patch("pi.pi_controller.GCSDevice") -def test_reference_move_timeout(mock_gcs_device_cls): - """Test reference move with a timeout.""" +def test_home_timeout(mock_gcs_device_cls): + """Test home with a timeout.""" mock_device = MagicMock() # Simulate IsMoving always True mock_device.IsMoving.return_value = {"1": True} mock_gcs_device_cls.return_value = mock_device - controller = PIControllerBase(quiet=True) - controller.connected = True + controller = PIControllerBase(log=False) + controller._set_connected(True) device_key = ("ip", 1, 1) controller.devices[device_key] = mock_device - result = controller.reference_move( + result = controller.home( device_key, "1", method="FRF", blocking=True, timeout=0.1 ) assert result is False