axis.py

# -*- coding: utf-8 -*-
#
# This file is part of the bliss project
#
# Copyright (c) 2015-2020 Beamline Control Unit, ESRF
# Distributed under the GNU LGPLv3. See LICENSE for more info.

"""
Axis related classes (:class:`~bliss.common.axis.Axis`, \
:class:`~bliss.common.axis.AxisState`, :class:`~bliss.common.axis.Motion`
and :class:`~bliss.common.axis.GroupMove`)
"""
from bliss import global_map
from bliss.common.cleanup import capture_exceptions
from bliss.common.motor_config import StaticConfig
from bliss.common.motor_settings import AxisSettings
from bliss.common import event
from bliss.common.greenlet_utils import protect_from_one_kill
from bliss.common.utils import with_custom_members, safe_get
from bliss.config.channels import Channel
from bliss.common.logtools import log_debug, lprint, lprint_disable
from bliss.common.utils import rounder
from bliss.common.utils import autocomplete_property

import enum
import gevent
import re
import sys
import math
import functools
import collections
import numpy
from unittest import mock
import warnings

warnings.simplefilter("once", DeprecationWarning)


#: Default polling time
DEFAULT_POLLING_TIME = 0.02


class GroupMove:
    def __init__(self, parent=None):
        self.parent = parent
        self._move_task = None
        self._motions_dict = dict()
        self._stop_motion = None
        self._user_stopped = False

    # Public API

    @property
    def is_moving(self):
        # A greenlet evaluates to True when not dead
        return bool(self._move_task)

    def move(
        self,
        motions_dict,
        prepare_motion,
        start_motion,
        stop_motion,
        move_func=None,
        wait=True,
        polling_time=None,
    ):
        self._motions_dict = motions_dict
        self._stop_motion = stop_motion
        self._user_stopped = False

        hooks = collections.defaultdict(list)
        executed_hooks = dict()
        axes = set()
        hooked_axes = set()
        for motions in motions_dict.values():
            for motion in motions:
                axis = motion.axis
                axes.add(axis)

                # group motion hooks
                for hook in axis.motion_hooks:
                    hooks[hook].append(motion)

        with capture_exceptions(raise_index=0) as capture:
            for hook, motions in hooks.items():
                hooked_axes.union({m.axis for m in motions})

                with capture():
                    hook._init()
                    hook.pre_move(motions)

                executed_hooks[hook] = motions

                if capture.failed:
                    # something wrong happened with this hook:
                    # let's call post_move for all executed hooks so far
                    # (including this one), in reversed order
                    for hook, motions in reversed(list(executed_hooks.items())):
                        with capture():
                            hook.post_move(motions)
                    return

        # now check if axes are ready ;
        # the check happens after pre_move hooks execution,
        # some axes can **become** ready because of the hook
        with capture_exceptions(raise_index=0) as capture:
            for axis in axes:
                with capture():
                    axis._check_ready()
                if capture.failed:
                    # this axis _check_ready() had a problem:
                    # need to ensure post_move hook is called,
                    # if the pre_move was executed
                    for hook in reversed(axis.motion_hooks):
                        motions = executed_hooks.get(hook)
                        if motions:
                            with capture():
                                hook.post_move(motions)
                    return

        for controller, motions in motions_dict.items():
            if prepare_motion is not None:
                prepare_motion(controller, motions)
            for motion_obj in motions:
                msg = motion_obj.user_msg
                if msg:
                    lprint(msg)

        started = gevent.event.Event()

        self._move_task = gevent.spawn(
            self._move,
            motions_dict,
            start_motion,
            stop_motion,
            move_func,
            started,
            polling_time,
        )

        try:
            # Wait for the move to be started (or finished)
            gevent.wait([started, self._move_task], count=1)
        except BaseException:
            self.stop()
            raise
        # Wait if necessary and raise the move task exception if any
        if wait or self._move_task.ready():
            self.wait()

    def wait(self):
        if self._move_task is not None:
            try:
                self._move_task.get()
            except BaseException:
                self.stop()
                raise

    def stop(self, wait=True):
        with capture_exceptions(raise_index=0) as capture:
            if self._move_task is not None:
                with capture():
                    self._stop_move(self._motions_dict, self._stop_motion)
                if wait:
                    self._move_task.get()

    # Internal methods

    def _monitor_move(self, motions_dict, move_func, polling_time):
        monitor_move = dict()
        for controller, motions in motions_dict.items():
            for motion in motions:
                if move_func is None:
                    move_func = "_handle_move"
                axis_polling_time = (
                    motion.axis._polling_time if polling_time is None else polling_time
                )
                task = gevent.spawn(
                    getattr(motion.axis, move_func), motion, axis_polling_time
                )
                monitor_move[motion] = task
        try:
            gevent.joinall(monitor_move.values(), raise_error=True)
        finally:
            # update the last motor state
            for motion, task in monitor_move.items():
                try:
                    motion.last_state = task.get(block=False)
                except BaseException:
                    pass

    def _stop_move(self, motions_dict, stop_motion):
        self._user_stopped = True
        stop = []
        for controller, motions in motions_dict.items():
            stop.append(gevent.spawn(stop_motion, controller, motions))
        # Raise exception if any, when all the stop tasks are finished
        for task in gevent.joinall(stop):
            task.get()

    def _stop_wait(self, motions_dict, exception_capture):
        stop_wait = []
        for controller, motions in motions_dict.items():
            for motion in motions:
                stop_wait.append(gevent.spawn(motion.axis._move_loop))
        gevent.joinall(stop_wait)
        task_index = 0
        for controller, motions in motions_dict.items():
            for motion in motions:
                with exception_capture():
                    motion.last_state = stop_wait[task_index].get()
                task_index += 1

    @protect_from_one_kill
    def _do_backlash_move(self, motions_dict, polling_time):
        backlash_move = []
        for controller, motions in motions_dict.items():
            for motion in motions:
                if motion.backlash:
                    if self._user_stopped:
                        # have to recalculate target: do backlash from where it stopped
                        motion.target_pos = (
                            motion.axis.dial * motion.axis.steps_per_unit
                        )
                    backlash_motion = Motion(
                        motion.axis,
                        motion.target_pos + motion.backlash,
                        motion.backlash,
                    )
                    axis_polling_time = (
                        motion.axis._polling_time
                        if polling_time is None
                        else polling_time
                    )

                    backlash_move.append(
                        gevent.spawn(
                            motion.axis._backlash_move,
                            backlash_motion,
                            axis_polling_time,
                        )
                    )
        gevent.joinall(backlash_move)
        gevent.joinall(backlash_move, raise_error=True)

    def _move(
        self,
        motions_dict,
        start_motion,
        stop_motion,
        move_func,
        started_event,
        polling_time,
    ):
        # Set axis moving state
        for motions in motions_dict.values():
            for motion in motions:
                motion.last_state = None
                motion.axis._set_moving_state()

                for _, chan in motion.axis._beacon_channels.items():
                    chan.unregister_callback(chan._setting_update_cb)
        with capture_exceptions(raise_index=0) as capture:
            try:
                # Spawn start motion for all controllers
                start = [
                    gevent.spawn(start_motion, controller, motions)
                    for controller, motions in motions_dict.items()
                ]

                # Wait for the controllers to be started
                with capture():
                    gevent.joinall(start, raise_error=True)
                if capture.failed:
                    gevent.joinall(start)
                    # start failed, stop all axes and wait end of motion
                    with capture():
                        self._stop_move(motions_dict, stop_motion)

                    self._stop_wait(motions_dict, capture)
                    return

                # All the controllers are now started
                started_event.set()

                if self.parent:
                    event.send(self.parent, "move_done", False)

                # Spawn the monitoring for all motions
                with capture():
                    self._monitor_move(motions_dict, move_func, polling_time)
                if capture.failed:
                    with capture():
                        self._stop_move(motions_dict, stop_motion)
                    self._stop_wait(motions_dict, capture)

                # Do backlash move, if needed
                with capture():
                    self._do_backlash_move(motions_dict, polling_time)
                if capture.failed:
                    with capture():
                        self._stop_move(motions_dict, stop_motion)
                    self._stop_wait(motions_dict, capture)
            finally:
                reset_setpos = capture.failed or self._user_stopped

                # cleanup
                # -------
                # update final state ; in case of exception
                # state is set to FAULT
                for motions in motions_dict.values():
                    for motion in motions:
                        state = motion.last_state
                        if state is not None:
                            continue

                        with capture():
                            state = motion.axis.hw_state
                        if state is None:
                            state = AxisState("FAULT")
                        # update state and update dial pos.
                        with capture():
                            motion.axis._update_settings(state)

                # update set position if motor has been stopped,
                # or if an exception happened or if motion type is
                # home search or hw limit search ;
                # as state update happened just before, this
                # is equivalent to sync_hard -> emit the signal
                # (useful for real motor positions update in case
                # of pseudo axis)
                # -- jog move is a special case
                if len(motions_dict) == 1:
                    motion = motions_dict[list(motions_dict.keys()).pop()][0]
                    if motion.type == "jog":
                        reset_setpos = False
                        motion.axis._jog_cleanup(
                            motion.saved_velocity, motion.reset_position
                        )
                    elif motion.type == "homing":
                        reset_setpos = True
                    elif motion.type == "limit_search":
                        reset_setpos = True
                if reset_setpos:
                    with capture():
                        for motions in motions_dict.values():
                            for motion in motions:
                                motion.axis._set_position = motion.axis.position
                                event.send(motion.axis, "sync_hard")

                hooks = collections.defaultdict(list)
                for motions in motions_dict.values():
                    for motion in motions:
                        axis = motion.axis

                        # group motion hooks
                        for hook in axis.motion_hooks:
                            hooks[hook].append(motion)

                        # set move done
                        for _, chan in axis._beacon_channels.items():
                            chan.register_callback(chan._setting_update_cb)

                        motion.axis._set_move_done()

                if self._user_stopped:
                    lprint("")
                    for motion in motions:
                        _axis = motion.axis
                        _axis_pos = safe_get(_axis, "position", on_error="!ERR")
                        lprint(f"Axis {_axis.name} stopped at position {_axis_pos}")

                try:
                    if self.parent:
                        event.send(self.parent, "move_done", True)
                finally:
                    for hook, motions in reversed(list(hooks.items())):
                        with capture():
                            hook.post_move(motions)


class Modulo:
    def __init__(self, mod=360):
        self.modulo = mod

    def __call__(self, axis):
        dial_pos = axis.dial
        axis._Axis__do_set_dial(dial_pos % self.modulo)


class Motion:
    """Motion information

    Represents a specific motion. The following members are present:

    * *axis* (:class:`Axis`): the axis to which this motion corresponds to
    * *target_pos* (:obj:`float`): final motion position
    * *delta* (:obj:`float`): motion displacement
    * *backlash* (:obj:`float`): motion backlash

    Note: target_pos and delta can be None, in case of specific motion
    types like homing or limit search
    """

    def __init__(
        self, axis, target_pos, delta, motion_type="move", user_target_pos=None
    ):
        self.__axis = axis
        self.__type = motion_type
        self.user_target_pos = user_target_pos
        self.target_pos = target_pos
        self.delta = delta
        self.backlash = 0

    @property
    def axis(self):
        """Reference to :class:`Axis`"""
        return self.__axis

    @property
    def type(self):
        return self.__type

    @property
    def user_msg(self):
        start_ = rounder(self.axis.tolerance, self.axis.position)
        if self.type == "jog":
            msg = (
                f"Moving {self.axis.name} from {start_} until it is stopped, at constant velocity in {'positive' if self.delta > 0 else 'negative'} direction: {abs(self.target_pos/self.axis.steps_per_unit)}\n"
                f"To stop it: {self.axis.name}.stop()"
            )
            return msg

        else:
            if self.user_target_pos is None:
                return None
            else:
                if isinstance(self.user_target_pos, str):
                    # can be a string in case of special move like limit search, homing...
                    end_ = self.user_target_pos
                else:
                    end_ = rounder(self.axis.tolerance, self.user_target_pos)
                return f"Moving {self.axis.name} from {start_} to {end_}"


class Trajectory(object):
    """ Trajectory information

    Represents a specific trajectory motion.

    """

    def __init__(self, axis, pvt):
        """
        Args:
            axis -- axis to which this motion corresponds to
            pvt  -- numpy array with three fields ('position','velocity','time')
        """
        self.__axis = axis
        self.__pvt = pvt
        self._events_positions = numpy.empty(
            0, dtype=[("position", "f8"), ("velocity", "f8"), ("time", "f8")]
        )

    @property
    def axis(self):
        return self.__axis

    @property
    def pvt(self):
        return self.__pvt

    @property
    def events_positions(self):
        return self._events_positions

    @events_positions.setter
    def events_positions(self, events):
        self._events_positions = events

    def has_events(self):
        return self._events_positions.size

    def __len__(self):
        return len(self.pvt)

    def convert_to_dial(self):
        """
        Return a new trajectory with pvt position, velocity converted to dial units and steps per unit
        """
        user_pos = self.__pvt["position"]
        user_velocity = self.__pvt["velocity"]
        pvt = numpy.copy(self.__pvt)
        pvt["position"] = self.axis.user2dial(user_pos) * self.axis.steps_per_unit
        pvt["velocity"] *= self.axis.steps_per_unit
        new_obj = self.__class__(self.axis, pvt)
        pattern_evts = numpy.copy(self._events_positions)
        pattern_evts["position"] *= self.axis.steps_per_unit
        pattern_evts["velocity"] *= self.axis.steps_per_unit
        new_obj._events_positions = pattern_evts
        return new_obj


class CyclicTrajectory(Trajectory):
    def __init__(self, axis, pvt, nb_cycles=1, origin=0):
        """
        Args:
            axis -- axis to which this motion corresponds to
            pvt  -- numpy array with three fields ('position','velocity','time')
                    point coordinates are in relative space
        """
        super(CyclicTrajectory, self).__init__(axis, pvt)
        self.nb_cycles = nb_cycles
        self.origin = origin

    @property
    def pvt_pattern(self):
        return super(CyclicTrajectory, self).pvt

    @property
    def events_pattern_positions(self):
        return super(CyclicTrajectory, self).events_positions

    @events_pattern_positions.setter
    def events_pattern_positions(self, values):
        self._events_positions = values

    @property
    def is_closed(self):
        """True if the trajectory is closed (first point == last point)"""
        pvt = self.pvt_pattern
        return (
            pvt["time"][0] == 0
            and pvt["position"][0] == pvt["position"][len(self.pvt_pattern) - 1]
        )

    @property
    def pvt(self):
        """Return the full PVT table. Positions are absolute"""
        pvt_pattern = self.pvt_pattern
        if self.is_closed:
            # take first point out because it is equal to the last
            raw_pvt = pvt_pattern[1:]
            cycle_size = raw_pvt.shape[0]
            size = self.nb_cycles * cycle_size + 1
            offset = 1
        else:
            raw_pvt = pvt_pattern
            cycle_size = raw_pvt.shape[0]
            size = self.nb_cycles * cycle_size
            offset = 0
        pvt = numpy.empty(size, dtype=raw_pvt.dtype)
        last_time, last_position = 0, self.origin
        for cycle in range(self.nb_cycles):
            start = cycle_size * cycle + offset
            end = start + cycle_size
            pvt[start:end] = raw_pvt
            pvt["time"][start:end] += last_time
            last_time = pvt["time"][end - 1]
            pvt["position"][start:end] += last_position
            last_position = pvt["position"][end - 1]

        if self.is_closed:
            pvt["time"][0] = pvt_pattern["time"][0]
            pvt["position"][0] = pvt_pattern["position"][0] + self.origin

        return pvt

    @property
    def events_positions(self):
        pattern_evts = self.events_pattern_positions
        time_offset = 0.0
        last_time = self.pvt_pattern["time"][-1]
        nb_pattern_evts = len(pattern_evts)
        all_events = numpy.empty(
            self.nb_cycles * len(pattern_evts), dtype=pattern_evts.dtype
        )
        for i in range(self.nb_cycles):
            sub_evts = all_events[
                i * nb_pattern_evts : i * nb_pattern_evts + nb_pattern_evts
            ]
            sub_evts[:] = pattern_evts
            sub_evts["time"] += time_offset
            time_offset += last_time
        return all_events

    def convert_to_dial(self):
        """
        Return a new trajectory with pvt position, velocity converted to dial units and steps per unit
        """
        new_obj = super(CyclicTrajectory, self).convert_to_dial()
        new_obj.origin = self.axis.user2dial(self.origin) * self.axis.steps_per_unit
        new_obj.nb_cycles = self.nb_cycles
        return new_obj


def lazy_init(func):
    @functools.wraps(func)
    def func_wrapper(self, *args, **kwargs):
        self.controller._initialize_axis(self)
        return func(self, *args, **kwargs)

    return func_wrapper


@with_custom_members
class Axis:
    """
    Bliss motor axis

    Typical usage goes through the bliss configuration (see this module
    documentation above for an example)
    """

    READ_POSITION_MODE = enum.Enum("Axis.READ_POSITION_MODE", "CONTROLLER ENCODER")

    def __init__(self, name, controller, config):
        self.__name = name
        self.__controller = controller
        self.__settings = AxisSettings(self)
        self.__move_done = gevent.event.Event()
        self.__move_done_callback = gevent.event.Event()
        self.__move_done.set()
        self.__move_done_callback.set()
        self.__motion_hooks = []
        for hook in config.get("motion_hooks", []):
            hook._add_axis(self)
            self.__motion_hooks.append(hook)
        self.__encoder = config.get("encoder")
        if self.__encoder is not None:
            self.__encoder.axis = self
        self.__config = StaticConfig(config)
        self.__init_config_properties()
        self._group_move = GroupMove()
        self._beacon_channels = dict()
        self._move_stop_channel = Channel(
            f"axis.{self.name}.move_stop",
            default_value=False,
            callback=self._external_stop,
        )
        self._jog_velocity_channel = Channel(
            f"axis.{self.name}.change_jog_velocity",
            default_value=None,
            callback=self._set_jog_velocity,
        )
        self._lock = gevent.lock.Semaphore()
        self.__no_offset = False

        try:
            config.parent
        # some Axis don't have a controller
        # like SoftAxis
        except AttributeError:
            disabled_cache = list()
        else:
            disabled_cache = config.parent.get(
                "disabled_cache", []
            )  # get it from controller (parent)
        disabled_cache.extend(config.get("disabled_cache", []))  # get it for this axis
        for settings_name in disabled_cache:
            self.settings.disable_cache(settings_name)
        self._unit = self.config.get("unit", str, None)
        self._polling_time = config.get("polling_time", DEFAULT_POLLING_TIME)
        global_map.register(self, parents_list=["axes", controller])

    def __close__(self):
        try:
            controller_close = self.__controller.close
        except AttributeError:
            pass
        else:
            controller_close()

    @property
    def no_offset(self):
        return self.__no_offset

    @no_offset.setter
    def no_offset(self, value):
        self.__no_offset = value

    @property
    def unit(self):
        """Axis name"""
        return self._unit

    @property
    def name(self):
        """Axis name"""
        return self.__name

    @autocomplete_property
    def controller(self):
        """Reference to :class:`~bliss.controllers.motor.Controller`"""
        return self.__controller

    @property
    def config(self):
        """Reference to the :class:`~bliss.common.motor_config.StaticConfig`"""
        return self.__config

    @property
    def settings(self):
        """
        Reference to the
        :class:`~bliss.controllers.motor_settings.AxisSettings`
        """
        return self.__settings

    @property
    def is_moving(self):
        """
        Tells if the axis is moving (:obj:`bool`)
        """
        return not self.__move_done.is_set()

    def __init_config_properties(
        self, velocity=True, acceleration=True, limits=True, sign=True, backlash=True
    ):
        self.__steps_per_unit = self.config.get("steps_per_unit", float, 1)
        self.__tolerance = self.config.get("tolerance", float, 1e-4)
        if velocity:
            if self.controller.axis_settings.config_setting["velocity"]:
                self.__config_velocity = self.config.get("velocity", float)
            if self.controller.axis_settings.config_setting["jog_velocity"]:
                self.__config_jog_velocity = self.config.get(
                    "jog_velocity", float, self.__config_velocity
                )
        if acceleration:
            if self.controller.axis_settings.config_setting["acceleration"]:
                self.__config_acceleration = self.config.get("acceleration", float)
        if limits:
            self.__config_low_limit = self.config.get("low_limit", float, float("-inf"))
            self.__config_high_limit = self.config.get(
                "high_limit", float, float("+inf")
            )
        if backlash:
            self.__config_backlash = self.config.get("backlash", float, 0)

    @property
    def steps_per_unit(self):
        """Current steps per unit (:obj:`float`)"""
        return self.__steps_per_unit

    @property
    def config_backlash(self):
        """Current backlash in user units (:obj:`float`)"""
        return self.__config_backlash

    @property
    @lazy_init
    def backlash(self):
        """Current backlash in user units (:obj:`float`)"""
        backlash = self.settings.get("backlash")
        if backlash is None:
            return 0
        return backlash

    @backlash.setter
    def backlash(self, backlash):
        self.settings.set("backlash", backlash)

    @property
    def tolerance(self):
        """Current Axis tolerance in dial units (:obj:`float`)"""
        return self.__tolerance

    @property
    def encoder(self):
        """
        Reference to :class:`~bliss.common.encoder.Encoder` or None if no
        encoder is defined
        """
        return self.__encoder

    @property
    def motion_hooks(self):
        """Registered motion hooks (:obj:`MotionHook`)"""
        return self.__motion_hooks

    @property
    @lazy_init
    def offset(self):
        """Current offset in user units (:obj:`float`)"""
        offset = self.settings.get("offset")
        if offset is None:
            return 0
        return offset

    @offset.setter
    def offset(self, new_offset):
        if self.no_offset:
            raise RuntimeError(
                f"{self.name}: cannot change offset, axis has 'no offset' flag"
            )
        self.__do_set_position(offset=new_offset)

    @property
    @lazy_init
    def sign(self):
        """Current motor sign (:obj:`int`) [-1, 1]"""
        sign = self.settings.get("sign")
        if sign is None:
            return 1
        return sign

    @sign.setter
    def sign(self, new_sign):
        new_sign = float(
            new_sign
        )  # works both with single float or numpy array of 1 element
        new_sign = math.copysign(1, new_sign)
        if new_sign != self.sign:
            if self.no_offset:
                raise RuntimeError(
                    f"{self.name}: cannot change sign, axis has 'no offset' flag"
                )
            self.settings.set("sign", new_sign)
            # update pos with new sign, offset stays the same
            # user pos is **not preserved** (like spec)
            self.position = self.dial2user(self.dial)

    def set_setting(self, *args):
        """Sets the given settings"""
        self.settings.set(*args)

    def get_setting(self, *args):
        """Return the values for the given settings"""
        return self.settings.get(*args)

    def has_tag(self, tag):
        """
        Tells if the axis has the given tag

        Args:
            tag (str): tag name

        Return:
            bool: True if the axis has the tag or False otherwise
        """
        for t, axis_list in self.__controller._tagged.items():
            if t != tag:
                continue
            if self.name in [axis.name for axis in axis_list]:
                return True
        return False

    @lazy_init
    def on(self):
        """Turns the axis on"""
        if self.is_moving:
            return

        self.__controller.set_on(self)
        state = self.__controller.state(self)
        self.settings.set("state", state)

    @lazy_init
    def off(self):
        """Turns the axis off"""
        if self.is_moving:
            raise RuntimeError("Can't set power off while axis is moving")

        self.__controller.set_off(self)
        state = self.__controller.state(self)
        self.settings.set("state", state)

    @property
    @lazy_init
    def _set_position(self):
        sp = self.settings.get("_set_position")
        if sp is not None:
            return sp
        position = self.position
        self._set_position = position
        return position

    @_set_position.setter
    def _set_position(self, new_set_pos):
        new_set_pos = float(
            new_set_pos
        )  # accepts both float or numpy array of 1 element
        self.settings.set("_set_position", new_set_pos)

    @property
    @lazy_init
    def measured_position(self):
        """
        Return the encoder value in user units.

        Return:
            float: encoder value in user units
        """
        return self.dial2user(self.dial_measured_position)

    @property
    @lazy_init
    def dial_measured_position(self):
        """
        Return the dial encoder position.

        Return:
            float: dial encoder position
        """
        if self.encoder is not None:
            return self.encoder.read()
        else:
            raise RuntimeError("Axis '%s` has no encoder." % self.name)

    def __do_set_dial(self, new_dial):
        user_pos = self.position
        old_dial = self.dial

        # Set the new dial on the encoder
        if self._read_position_mode == self.READ_POSITION_MODE.ENCODER:
            dial_pos = self.encoder.set(new_dial)
        else:
            # Send the new value in motor units to the controller
            # and read back the (atomically) reported position
            new_hw = new_dial * self.steps_per_unit
            hw_pos = self.__controller.set_position(self, new_hw)
            dial_pos = hw_pos / self.steps_per_unit
        self.settings.set("dial_position", dial_pos)

        if self.no_offset:
            self.__do_set_position(dial_pos, offset=0)
        else:
            # set user pos, will recalculate offset
            # according to new dial
            self.__do_set_position(user_pos)

        return dial_pos

    @property
    @lazy_init
    def dial(self):
        """
        Return current dial position, or set dial

        Return:
            float: current dial position (dimensionless)
        """
        dial_pos = self.settings.get("dial_position")
        if dial_pos is None:
            dial_pos = self._update_dial()
        return dial_pos

    @dial.setter
    def dial(self, new_dial):
        if self.is_moving:
            raise RuntimeError(
                "%s: can't set axis dial position " "while moving" % self.name
            )
        new_dial = float(new_dial)  # accepts both float or numpy array of 1 element
        old_dial = self.dial
        new_dial = self.__do_set_dial(new_dial)
        lprint(f"'{self.name}` dial position reset from {old_dial} to {new_dial}")

    def __do_set_position(self, new_pos=None, offset=None):
        dial = self.dial
        curr_offset = self.offset
        if offset is None:
            # calc offset
            offset = new_pos - self.sign * dial
        if math.isnan(offset):
            # this can happen if dial is nan;
            # cannot continue
            return False
        if math.isclose(offset, 0):
            offset = 0
        if not math.isclose(curr_offset, offset):
            self.settings.set("offset", offset)
        if new_pos is None:
            # calc pos from offset
            new_pos = self.sign * dial + offset
        if math.isnan(new_pos):
            # do not allow to assign nan as a user position
            return False
        self.settings.set("position", new_pos)
        self._set_position = new_pos
        return True

    @property
    @lazy_init
    def position(self):