motorDriverTests/common.py

import time
import os

import pytest
import yaml
from caproto.sync.client import read, write
from caproto import CaprotoTimeoutError, ChannelType

FTYPE_TO_TYPE = {
    ChannelType.STRING: str,
    ChannelType.INT: int,
    ChannelType.FLOAT: float,
    ChannelType.ENUM: int,
    ChannelType.CHAR: bytes,
    ChannelType.LONG: int,
    ChannelType.DOUBLE: float,

    ChannelType.TIME_STRING: str,
    ChannelType.TIME_INT: int,
    ChannelType.TIME_FLOAT: float,
    ChannelType.TIME_ENUM: int,
    ChannelType.TIME_CHAR: bytes,
    ChannelType.TIME_LONG: int,
    ChannelType.TIME_DOUBLE: float,

    ChannelType.CTRL_STRING: str,
    ChannelType.CTRL_INT: int,
    ChannelType.CTRL_FLOAT: float,
    ChannelType.CTRL_ENUM: int,
    ChannelType.CTRL_CHAR: bytes,
    ChannelType.CTRL_LONG: int,
    ChannelType.CTRL_DOUBLE: float,
}


def read_config():
    root_dir = os.path.dirname(__file__)
    with open(root_dir + "/config.yaml", "r") as f:
        return yaml.safe_load(f)


class MajorState(Exception):
    pass


class Motor:

    # Motor record fields
    fields = {
        'readpv': 'RBV',
        'writepv': 'VAL',
        'stop': 'STOP',
        'donemoving': 'DMOV',
        'moving': 'MOVN',
        'miss': 'MISS',
        'homeforward': 'HOMF',
        'homereverse': 'HOMR',
        'direction': 'DIR',
        'speed': 'VELO',
        'basespeed': 'VBAS',
        'maxspeed': 'VMAX',
        'offset': 'OFF',
        'dialhighlimit': 'DHLM',
        'diallowlimit': 'DLLM',
        'highlimit': 'HLM',
        'lowlimit': 'LLM',
        'softlimit': 'LVIO',
        'lowlimitswitch': 'LLS',
        'highlimitswitch': 'HLS',
        'resolution': 'MRES',
        'enable': 'CNEN',
        'set': 'SET',
        'foff': 'FOFF',
        'status': 'MSTA',
        'alarm_status': 'STAT',
        'alarm_severity': 'SEVR',
    }

    def __init__(self, ip, port, pv):
        self.ip = ip
        self.port = port
        self.pv = pv

    def write(self, suffix, value):
        write(self.pv + suffix, value)

    def write_field(self, fieldname, value):
        self.write('.' + self.fields[fieldname], value)

    def read(self, suffix, as_string=False):
        response = read(self.pv + suffix).data
        return self._convert_value(response, as_string)

    def _convert_value(self, response, as_string=False):

        # By default, the response data is always a list to cover all possible
        # readback values from EPICS (lists, strings, chars, numbers). The last
        # two cases need to be treated in a special way. They can be identified
        # by the list length being 1.
        if len(response.data) == 1:
            value = response.data[0]
            if isinstance(value, bytes):
                return value.decode()

            # If an empty string is returned, the data has a single entry
            # (the NULL terminator)
            if as_string:
                return bytes(response.data).rstrip(b'\x00').decode(
                    encoding='utf-8', errors='ignore')

            return value

        # Strings are read with their NULL terminator, hence it needs to be
        # stripped before decoding
        if as_string:
            return bytes(response.data).rstrip(b'\x00').decode(
                encoding='utf-8', errors='ignore')
        return response.data

    def read_field(self, fieldname, as_string=False):
        return self.read('.' + self.fields[fieldname], as_string)

    def move_and_wait(self, target):
        """
        Move the motor to the given target and return, once the motor has 
        finished moving. The motor status is polled regulary to see if an error
        occurred during movement. In case this happens, an error is raised.
        """
        self.write(self.fields['writepv'], target)

        # Give the record some time to start
        time.sleep(1)
        self.wait_for_done()

    def limits(self):
        return (self.read_field('lowlimit'), self.read_field('highlimit'))

    def at_target(self, target=None):
        """
        Check if the motor arrived at its target by checking the MISS PV and
        comparing the writepv value with the readpv value (taking precision)
        into account
        """
        if target is None:
            target = self.read_field('writepv')
        return (self.read_field('miss') == 0 and
                abs(self.read_field('readpv') - target) < self.read_field('precision'))

    def wait_for_done(self):
        while not self.read_field('donemoving'):
            if self.has_error():
                raise MajorState('Record is in MAJOR state!')
            time.sleep(1)

    def has_error(self):
        self.read_field('alarm_severity') == 2

    def is_homed(self):
        # See https://epics.anl.gov/bcda/synApps/motor/motorRecord.html#Fields_status
        str = format(self.read_field('status'), '016b')
        return bool(str[15])


class SinqMotor(Motor):
    # PV suffixes used in SinqMotor drivers
    suffixes = {
        'enable': ':Enable',
        'enable_rbv': ':EnableRBV',
        'can_disable': ':CanDisable',
        'connected_rbv': ':Connected',
        'encoder_type': ':EncoderType',
        'reseterrorpv': ':Reset',
        'errormsgpv': '-MsgTxt',
    }

    def write_field(self, fieldname, value):
        if fieldname in self.suffixes:
            self.write(self.suffixes[fieldname], value)
        self.write('.' + self.fields[fieldname], value)

    def read_field(self, fieldname):
        if fieldname in self.suffixes:
            return self.read(self.suffixes[fieldname])
        return self.read('.' + self.fields[fieldname])


class TurboPMAC(SinqMotor):
    pass