Skip to content

Commit

Permalink
fix(#1184): PositionerBase as an ABC to ensure .egu and .move() are o…
Browse files Browse the repository at this point in the history
…verwritten in subclasses
  • Loading branch information
Mathias Guijarro committed Apr 8, 2024
1 parent 5c03c3f commit 50e7658
Showing 1 changed file with 11 additions and 10 deletions.
21 changes: 11 additions & 10 deletions ophyd/positioner.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import functools
import logging
import time
from abc import ABCMeta, abstractmethod
from collections import OrderedDict
from typing import Any, Callable

Expand All @@ -13,18 +14,11 @@
logger = logging.getLogger(__name__)


class PositionerBase(OphydObject):
class PositionerBase(OphydObject, metaclass=ABCMeta):
"""The positioner base class
Subclass from this to implement your own positioners.
.. note ::
Subclasses should add an additional 'wait' keyword argument on
the move method. The `MoveStatus` object returned from
`PositionerBase` can then be waited on after the subclass
finishes the motion configuration.
"""

SUB_START = "start_moving"
Expand Down Expand Up @@ -136,9 +130,10 @@ def report(self):
return rep

@property
@abstractmethod
def egu(self):
"""The engineering units (EGU) for positions"""
raise NotImplementedError("Subclass must implement egu")
pass

@property
def limits(self):
Expand All @@ -152,7 +147,8 @@ def low_limit(self):
def high_limit(self):
return self.limits[1]

def move(self, position, moved_cb=None, timeout=None):
@abstractmethod
def move(self, position, wait=False, moved_cb=None, timeout=None):
"""Move to a specified position, optionally waiting for motion to
complete.
Expand All @@ -175,6 +171,8 @@ def move(self, position, moved_cb=None, timeout=None):
timeout : float, optional
Maximum time to wait for the motion. If None, the default timeout
for this positioner is used.
wait : bool, optional
Wait for the Status object to be complete (False by default)
Returns
-------
Expand Down Expand Up @@ -208,6 +206,9 @@ def move(self, position, moved_cb=None, timeout=None):

self.subscribe(status._finished, event_type=self._SUB_REQ_DONE, run=False)

if wait:
status.wait()

return status

def _done_moving(self, success=True, timestamp=None, value=None, **kwargs):
Expand Down

0 comments on commit 50e7658

Please sign in to comment.