Make the limit mode of Motor input configurable (#87)

* add optional kwarg limit_mode to Motor

* update all calls to _handle_user_meta() to handle both required and optional metadata

* add optional metadata for drive.axes

* add optional keyword motor_settings

* add handling of Axis optional keyword motor_settings to Drive

* add limit mode config to benchtop_run.toml

* add the motor_settings to the actor config dictionaries

* fix typo

* add limit mode selector to the DriveConfigOverlay / AxisCondigOverlay

* add limit_mode settings to the toml example

bapsf_motion/actors/axis_.py

@@ -35,6 +35,10 @@ class Axis(EventActor):
     units_per_rev: float
         The number of ``units`` traversed per motor revolution.
 
+    motor_settings : `dict`, optional
+        A dictionary containing the optionl keyword arguments for
+        |Motor|.  (DEFAULT: `None`)
+
     name: str
         Name the axis.  (DEFAULT: ``'Axis'``)
 
@@ -78,6 +82,7 @@ def __init__(
         ip: str,
         units: str,
         units_per_rev: float,
+        motor_settings: Dict[str, Any] = None,
         name: str = "Axis",
         logger: logging.Logger = None,
         loop: asyncio.AbstractEventLoop = None,
@@ -98,7 +103,7 @@ def __init__(
         )
 
         self._motor = None
-        self._spawn_motor(ip=ip)
+        self._spawn_motor(ip=ip, motor_settings=motor_settings)
 
         if isinstance(self._motor, Motor) and self._motor.terminated:
             # terminate self if Motor is terminated
@@ -129,27 +134,38 @@ def terminate(self, delay_loop_stop=False):
         self.motor.terminate(delay_loop_stop=True)
         super().terminate(delay_loop_stop=delay_loop_stop)
 
-    def _spawn_motor(self, ip):
+    def _spawn_motor(self, ip, motor_settings: Optional[dict]):
         if isinstance(self.motor, Motor) and not self.terminated:
             self.motor.terminate(delay_loop_stop=True)
 
+        if motor_settings is None:
+            motor_settings = {}
+
         self._motor = Motor(
             ip=ip,
             name="motor",
             logger=self.logger,
             loop=self.loop,
             auto_run=False,
             parent=self,
+            **motor_settings,
         )
 
     @property
     def config(self) -> Dict[str, Any]:
         """Dictionary of the axis configuration parameters."""
+        motor_settings = {}
+        for key, val in self.motor.config.items():
+            if key in ("name", "ip"):
+                continue
+            motor_settings[key] = val
+
         return {
             "name": self.name,
             "ip": self.motor.ip,
             "units": str(self.units),
-            "units_per_rev": self.units_per_rev.value.item()
+            "units_per_rev": self.units_per_rev.value.item(),
+            "motor_settings": motor_settings,
         }
     config.__doc__ = EventActor.config.__doc__
 

bapsf_motion/actors/drive_.py

@@ -9,6 +9,7 @@
 import asyncio
 import logging
 
+from collections import UserDict
 from typing import Any, Dict, List, Optional, Tuple
 
 from bapsf_motion.actors.base import EventActor
@@ -157,8 +158,7 @@ def _validate_axes(
 
         return tuple(conditioned_settings)
 
-    @staticmethod
-    def _validate_axis(settings: Dict[str, Any]) -> Dict[str, Any]:
+    def _validate_axis(self, settings: Dict[str, Any]) -> Dict[str, Any]:
         """Validate the |Axis| arguments defined in ``settings``."""
         # TODO: create warnings for logger, loop, and auto_run since
         #       this class overrides in inputs of thos
@@ -187,6 +187,20 @@ def _validate_axis(settings: Dict[str, Any]) -> Dict[str, Any]:
                     f"type {type(settings[key])}."
                 )
 
+        if (
+            "motor_settings" in settings
+            and not isinstance(settings["motor_settings"], (dict, UserDict))
+        ):
+            _motor_settings = settings.pop("motor_settings")
+            if _motor_settings is not None:
+                self.logger.warning(
+                    "Removing motor settings from the input configuration.",
+                    exc_info=TypeError(
+                        "Expected None or dictionary for motor settings "
+                        f"input, got type {type(_motor_settings)}."
+                    ),
+                )
+
         return settings
 
     def _spawn_axis(self, settings: Dict[str, Any]) -> Axis:

bapsf_motion/actors/motion_group_.py

@@ -242,6 +242,7 @@ class MotionGroupConfig(UserDict):
     #: optional keys for the motion group configuration dictionary
     _optional_metadata = {
         "motion_builder": {"exclusion", "layer"},
+        "drive.axes": {"motor_settings"},
     }
 
     #: allowable motion group header names
@@ -352,7 +353,9 @@ def _validate_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
             config.get("motion_builder", {})
         )
 
-        config = self._handle_user_meta(config, self._required_metadata["motion_group"])
+        req_meta = self._required_metadata.get("motion_group", set())
+        opt_meta = self._optional_metadata.get("motion_group", set())
+        config = self._handle_user_meta(config, set.union(req_meta, opt_meta))
 
         # TODO: the below commented out code block is not do-able since
         #       motion_builder.space can be defined as a string for builtin spaces
@@ -375,7 +378,8 @@ def _validate_drive(self, config: Dict[str, Any]) -> Dict[str, Any]:
         """
         Validate the drive component of the motion group configuration.
         """
-        req_meta = self._required_metadata["drive"]
+        req_meta = self._required_metadata.get("drive", set())
+        opt_meta = self._optional_metadata.get("drive", set())
 
         missing_meta = req_meta - set(config.keys())
         if missing_meta:
@@ -389,7 +393,7 @@ def _validate_drive(self, config: Dict[str, Any]) -> Dict[str, Any]:
             # )
             return {}
 
-        config = self._handle_user_meta(config, req_meta)
+        config = self._handle_user_meta(config, set.union(req_meta, opt_meta))
 
         ax_meta = set(config["axes"].keys())
         if len(self._required_metadata["drive.axes"] - ax_meta) == 0:
@@ -435,7 +439,8 @@ def _validate_axis(self, config: Dict[str, Any]) -> Dict[str, Any]:
         Validate the axis (e.g. axes.0) component of the drive
         component of the motion group configuration.
         """
-        req_meta = self._required_metadata["drive.axes"]
+        req_meta = self._required_metadata.get("drive.axes", set())
+        opt_meta = self._optional_metadata.get("drive.axes", set())
 
         missing_meta = req_meta - set(config.keys())
         if missing_meta:
@@ -444,7 +449,7 @@ def _validate_axis(self, config: Dict[str, Any]) -> Dict[str, Any]:
                 f"keys {missing_meta}."
             )
 
-        config = self._handle_user_meta(config, req_meta)
+        config = self._handle_user_meta(config, set.union(req_meta, opt_meta))
 
         # TODO: Is it better to do the type checks here or allow class
         #       instantiation to handle it.

bapsf_motion/actors/motor_.py

@@ -191,6 +191,12 @@ class Motor(EventActor):
     ip: `str`
         IPv4 address for the motor
 
+    limit_mode : `int`, optional
+        Define the operational mode of the motor limit switches. Value
+        should be an integer of value 1, 2, or 3.  1 indicates limit
+        is activated when energized, 2 indicates limit is activated
+        when de-energized, and 3 indicates no limits. (DEFAULT: ``1``)
+
     name: `str`, optional
         Name the motor.  If `None`, then the name will be automatically
         generated. (DEFAULT: `None`)
@@ -500,6 +506,7 @@ def __init__(
         self,
         *,
         ip: str,
+        limit_mode: int = None,
         name: str = None,
         logger: logging.Logger = None,
         loop: asyncio.AbstractEventLoop = None,
@@ -512,6 +519,7 @@ def __init__(
         self._setup = self._setup_defaults.copy()
         self._motor = self._motor_defaults.copy()
         self._status = self._status_defaults.copy()
+        self._limit_mode = limit_mode
 
         # simple signal to tell handlers that _status changed
         self.status_changed = SimpleSignal()
@@ -540,6 +548,30 @@ def __init__(
     def _configure_before_run(self):
         # actions to be done during object instantiation, but before
         # the asyncio event loop starts running.
+        if self._limit_mode is None:
+            self._limit_mode = self.motor["define_limits"]
+        elif not isinstance(self._limit_mode, int):
+            self.logger.warning(
+                "Assuming limit mode 1 for input argument 'limit_mode'.",
+                exc_info=TypeError(
+                    "Was expecting an int of value 1, 2, or 3 for input "
+                    f"argument 'limit_mode', got type "
+                    f"{type(self._limit_mode)} instead."
+                ),
+            )
+            self._limit_mode = self.motor["define_limits"]
+        elif self._limit_mode not in (1, 2, 3):
+            self.logger.warning(
+                "Assuming limit mode 1 for input argument 'limit_mode'.",
+                exc_info=ValueError(
+                    "Was expecting an int of value 1, 2, or 3 for input "
+                    f"argument 'limit_mode', got value "
+                    f"{self._limit_mode} instead."
+                ),
+            )
+            self._limit_mode = self.motor["define_limits"]
+        else:
+            self.motor["define_limits"] = self._limit_mode
 
         self.connect()
 
@@ -626,6 +658,7 @@ def _motor_defaults(self) -> Dict[str, Any]:
             "accel": None,
             "decel": None,
             "protocol_settings": None,
+            "define_limits": 1,  # 1 = energized, 2 = de-energized, 3 = None
         }
 
     @property
@@ -692,7 +725,7 @@ def _configure_motor(self):
         # input is closed (energized)
         # TODO: Replace with normal send_command when "define_limits" command
         #       is added to _commands dict
-        self.send_command("define_limits", 1)
+        self.send_command("define_limits", self.motor["define_limits"])
 
         # set format of immediate commands to decimal
         self._send_raw_command("IFD")
@@ -802,6 +835,7 @@ def config(self) -> Dict[str, Any]:
         return {
             "name": self.name,
             "ip": self.ip,
+            "limit_mode": self.motor["define_limits"],
         }
     config.__doc__ = EventActor.config.__doc__
 
@@ -1675,7 +1709,7 @@ def move_off_limit(self):
             self.move_to(move_to_pos)
 
             self.logger.warning("Moving off limits - enable limits")
-            self.send_command("define_limits", 1)
+            self.send_command("define_limits", self.motor["define_limits"])
             self.sleep(4 * self.heartrate.ACTIVE)
 
             alarm_msg = self.retrieve_motor_alarm(defer_status_update=True)

bapsf_motion/examples/bapsf_motion.toml

@@ -7,10 +7,12 @@ axes.0.name = "X"
 axes.0.ip = "192.168.6.104"
 axes.0.units = "cm"
 axes.0.units_per_rev = 0.254
+axes.0.motor_settings.limit_mode = 1
 axes.1.name = "Y"
 axes.1.ip = "192.168.6.103"
 axes.1.units = "cm"
 axes.1.units_per_rev = 0.254
+axes.1.motor_settings.limit_mode = 3
 
 [bapsf_motion.defaults.drive.1]
 name = "Plastic Room XY"

bapsf_motion/examples/benchtop_run.toml

@@ -10,10 +10,12 @@ axes.0.name = "X"
 axes.0.ip = "192.168.6.104"
 axes.0.units = "cm"
 axes.0.units_per_rev = 0.254
+axes.0.motor_settings.limit_mode = 1
 axes.1.name = "Y"
 axes.1.ip = "192.168.6.103"
 axes.1.units = "cm"
 axes.1.units_per_rev = 0.254
+axes.1.motor_settings.limit_mode = 3
 
 [run.mg.motion_builder]
 space.0.label = "X"