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.hook import execute_pre_move_hooks
from bliss.common.protocols import Scannable
from bliss.common.cleanup import capture_exceptions
from bliss.common.motor_config import MotorConfig
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, user_print, log_warning
from bliss.common.utils import rounder
from bliss.common.utils import autocomplete_property
from bliss.comm.exceptions import CommunicationError

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

warnings.simplefilter("once", DeprecationWarning)


#: Default polling time
DEFAULT_POLLING_TIME = 0.02


class AxisOnLimitError(RuntimeError):
    pass


class AxisFaultError(RuntimeError):
    pass


def float_or_inf(value, inf_sign=1):
    if value is None:
        value = float("inf")
        sign = math.copysign(1, inf_sign)
    else:
        sign = 1
    value = float(value)  # accepts float or numpy array of 1 element
    return sign * value


def _prepare_one_controller_motions(controller, motions):
    try:
        controller.prepare_all(*motions)
    except NotImplementedError:
        for motion in motions:
            controller.prepare_move(motion)


def _start_one_controller_motions(controller, motions):
    try:
        controller.start_all(*motions)
    except NotImplementedError:
        for motion in motions:
            controller.start_one(motion)


def _stop_one_controller_motions(controller, motions):
    try:
        controller.stop_all(*motions)
    except NotImplementedError:
        for motion in motions:
            controller.stop(motion.axis)


class GroupMove:
    def __init__(self, parent=None):
        self.parent = parent
        self._move_task = None
        self._motions_dict = dict()
        self._stop_motion = None
        self._interrupted_move = False
        self._backlash_started_event = gevent.event.Event()

    # Public API

    @property
    def is_moving(self):
        # A greenlet evaluates to True when it is alive
        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._interrupted_move = False

        # motions_dict is { controller: [motion, ...] }
        all_motions = list(itertools.chain(*motions_dict.values()))
        with execute_pre_move_hooks(all_motions):
            for axis in (m.axis for m in all_motions):
                axis._check_ready()

        for controller, motions in motions_dict.items():
            if prepare_motion is not None:
                prepare_motion(controller, motions)

            for motion_obj in motions:
                target_pos = motion_obj.user_target_pos
                if target_pos is not None and not isinstance(target_pos, str):
                    motion_obj.axis._set_position = target_pos

                msg = motion_obj.user_msg
                if msg:
                    user_print(msg)

        started = gevent.event.Event()

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

        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, wait=False)
                if wait:
                    self._move_task.get()

    # Internal methods

    def _monitor_move(self, motions_dict, move_func, stop_func):
        monitor_move_tasks = {}
        for controller, motions in motions_dict.items():
            for motion in motions:
                if move_func is None:
                    move_func = "_handle_move"
                task = gevent.spawn(getattr(motion.axis, move_func), motion)
                monitor_move_tasks[task] = motion

        try:
            gevent.joinall(monitor_move_tasks, raise_error=True)
        except BaseException:
            # in case of error, all moves are stopped
            # _stop_move is called with the same monitoring tasks:
            # the stop command will be sent, then the same monitoring continues
            # in '_stop_move'
            self._stop_move(motions_dict, stop_func, monitor_move_tasks)
            raise
        else:
            # everything went fine: update the last motor state ;
            # we know the tasks have all completed successfully
            for task, motion in monitor_move_tasks.items():
                motion.last_state = task.get()

    def _stop_move(self, motions_dict, stop_motion, stop_wait_tasks=None, wait=True):
        self._interrupted_move = True

        stop_tasks = []
        for controller, motions in motions_dict.items():
            stop_tasks.append(gevent.spawn(stop_motion, controller, motions))

        with capture_exceptions(raise_index=0) as capture:
            # wait for all stop commands to be sent
            with capture():
                gevent.joinall(stop_tasks, raise_error=True)
            if capture.failed:
                with capture():
                    gevent.joinall(stop_tasks)

            if wait:
                if stop_wait_tasks is None:
                    # create tasks to wait for end of motion
                    stop_wait_tasks = {}
                    for controller, motions in motions_dict.items():
                        for motion in motions:
                            stop_wait_tasks[
                                gevent.spawn(
                                    motion.axis._move_loop, motion.polling_time
                                )
                            ] = motion

                # wait for end of motion
                gevent.joinall(stop_wait_tasks)

                for task, motion in stop_wait_tasks.items():
                    motion.last_state = None
                    with capture():
                        motion.last_state = task.get()

    @protect_from_one_kill
    def _do_backlash_move(self, motions_dict):
        backlash_motions = collections.defaultdict(list)
        for controller, motions in motions_dict.items():
            for motion in motions:
                if motion.backlash:
                    if self._interrupted_move:
                        # have to recalculate target: do backlash move from where it stopped
                        motion.target_pos = (
                            motion.axis.dial * motion.axis.steps_per_unit
                        )
                        # Adjust the difference between encoder and motor controller indexer
                        if (
                            motion.axis._read_position_mode
                            == Axis.READ_POSITION_MODE.ENCODER
                        ):
                            controller_position = controller.read_position(motion.axis)
                            enc_position = motion.target_pos
                            delta_pos = controller_position - enc_position
                            motion.target_pos += delta_pos

                    backlash_motion = Motion(
                        motion.axis,
                        motion.target_pos + motion.backlash,
                        motion.backlash,
                    )
                    backlash_motions[controller].append(backlash_motion)

        if backlash_motions:
            backlash_mv_group = GroupMove()
            backlash_mv_group._do_move(
                backlash_motions,
                _start_one_controller_motions,
                _stop_one_controller_motions,
                None,
                self._backlash_started_event,
            )

    def _do_move(
        self, motions_dict, start_motion, stop_motion, move_func, started_event
    ):
        for controller, motions in motions_dict.items():
            for motion in motions:
                motion.last_state = None

        with capture_exceptions(raise_index=0) as capture:
            # Spawn start motion tasks for all controllers
            start = [
                gevent.spawn(start_motion, controller, motions)
                for controller, motions in motions_dict.items()
            ]

            # wait for start tasks to be all done ;
            # in case of error or if wait is interrupted (ctrl-c, kill...),
            # immediately stop and return
            with capture():
                gevent.joinall(start, raise_error=True)
            if capture.failed:
                # either a start task failed, or ctrl-c or kill happened.
                # First, let all start task to finish
                # /!\ it is important to join those, to ensure stop is called
                # after tasks are done otherwise there is a risk 'end' is
                # called before 'start' is all done
                with capture():
                    gevent.joinall(start)
                # then, stop all axes and wait end of motion
                self._stop_move(motions_dict, stop_motion)
                # exit
                return

            # All controllers are now started
            if started_event is not None:
                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, stop_motion)

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

                motion.axis.settings.unregister_channels_callbacks()

        with capture_exceptions(raise_index=0) as capture:
            with capture():
                self._do_move(
                    motions_dict, start_motion, stop_motion, move_func, started_event
                )
            # Do backlash move, if needed
            with capture():
                self._do_backlash_move(motions_dict)

            reset_setpos = bool(capture.failed) or self._interrupted_move

            # 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 None:
                        # update state and update dial pos.
                        with capture():
                            motion.axis._update_settings()

            # 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)

                    axis.settings.register_channels_callbacks()

                    # set move done
                    motion.axis._set_move_done()

            if self._interrupted_move:
                user_print("")
                for motion in motions:
                    _axis = motion.axis
                    _axis_pos = safe_get(_axis, "position", on_error="!ERR")
                    user_print(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
        self.polling_time = DEFAULT_POLLING_TIME

    @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:
    """ 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):
        if self.disabled:
            raise RuntimeError(f"Axis {self.name} is disabled")
        try:
            self.controller._initialize_axis(self)
        except Exception as e:
            if isinstance(e, CommunicationError):
                # also disable the controller
                self.controller._disabled = True
            self._disabled = True
            raise
        else:
            if not self.controller.axis_initialized(self):
                # failed to initialize
                self._disabled = True
        return func(self, *args, **kwargs)

    return func_wrapper


@with_custom_members
class Axis(Scannable):
    """
    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.__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 = MotorConfig(config)
        self.__settings = AxisSettings(self)
        self._init_config_properties()
        self.__no_offset = False
        self._group_move = GroupMove()
        self._lock = gevent.lock.Semaphore()
        self.__positioner = True
        self._disabled = 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 setting_name in disabled_cache:
            self.settings.disable_cache(setting_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])

        # create Beacon channels
        self.settings.init_channels()
        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,
        )

    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

    @property
    def _positioner(self):
        """Axis positioner"""
        return self.__positioner

    @_positioner.setter
    def _positioner(self, new_p):
        self.__positioner = new_p

    @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.MotorConfig`"""
        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 "velocity" in self.settings.config_settings:
                self.__config_velocity = self.config.get("velocity", float)
            if "jog_velocity" in self.settings.config_settings:
                self.__config_jog_velocity = self.config.get(
                    "jog_velocity", float, self.__config_velocity
                )
            self.__config_velocity_low_limit = self.config.get(
                "velocity_low_limit", float, float("inf")
            )
            self.__config_velocity_high_limit = self.config.get(
                "velocity_high_limit", float, float("inf")
            )
        if acceleration:
            if "acceleration" in self.settings.config_settings:
                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
    @lazy_init
    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

    @property
    def disabled(self):
        return self._disabled

    def enable(self):
        self._disabled = False
        self.hw_state  # force update

    @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
        if self._read_position_mode == self.READ_POSITION_MODE.ENCODER:
            # no setting, first time pos is read, init with controller hw pos.
            # issue 2463
            position = self._do_read_hw_position()
        else:
            position = self.position
        self._set_position = position
        return position

    @_set_position.setter
    @lazy_init
    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