frappy/frappy_psi/ionopimax.py

# *****************************************************************************
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
# Module authors:
#   Markus Zolliker <markus.zolliker@psi.ch>
#   Jael Celia Lorenzana <jael-celia.lorenzana@psi.ch>
# *****************************************************************************

"""support for iono pi max from Sfera Labs

supports also the smaller model iono pi
"""

from pathlib import Path
from frappy.core import Readable, Writable, Parameter, BoolType, StringType,\
    EnumType, FloatRange, Property, TupleOf, ERROR, IDLE
from frappy.errors import ConfigError, OutOfRangeError, ProgrammingError
from math import log


class Base:
    addr = Property('address', StringType())
    _devpath = None
    _devclass = None
    _status = None

    def initModule(self):
        super().initModule()
        candidates = list(Path('/sys/class').glob(f'ionopi*/*/{self.addr}'))
        if not candidates:
            raise ConfigError(f'can not find path for {self.addr}')
        if len(candidates) > 1:
            raise ProgrammingError(f"ambiguous paths {','.join(candidates)}")
        self._devpath = candidates[0].parent
        self._devclass = candidates[0].parent.name

    def read(self, addr, scale=None):
        with open(self._devpath / addr) as f:
            result = f.read()
        if scale:
            return float(result) / scale
        return result.strip()

    def write(self, addr, value, scale=None):
        value = str(round(value * scale)) if scale else str(value)
        with open(self._devpath / addr, 'w') as f:
            f.write(value)

    def read_status(self):
        if self._status is None:
            return IDLE, ''
        return self._status

class DigitalInput(Base, Readable):
    value = Parameter('input state', BoolType())
    true_level = Property('level representing True', EnumType(low=0, high=1), default=1)

    def initModule(self):
        super().initModule()
        if self._devclass == 'digital_io':
            self.write(f'{self.addr}_mode', 'inp')

    def read_value(self):
        return self.read(self.addr, 1) == self.true_level


class DigitalOutput(DigitalInput, Writable):
    target = Parameter('output state', BoolType(), readonly=False)

    def read_value(self):
        reply = self.read(self.addr)
        try:
            self._status = None
            self.read_status()
            value = int(reply)
        except ValueError:
            if reply == 'S':
                if self.addr.startswith('oc'):
                    self._status = ERROR, 'short circuit'
                else:
                    self._status = ERROR, 'fault while closed'
                value = 0
            else:
                self._status = ERROR, 'fault while open'
                value = 1
        return value == self.true_level

    def write_target(self, value):
        self.write(self.addr, value == self.true_level, 1)


class AnalogInput(Base, Readable):
    value = Parameter('analog value', FloatRange())
    rawrange = Property('raw range (electronic)', TupleOf(FloatRange(),FloatRange()))
    valuerange = Property('value range (physical)', TupleOf(FloatRange(),FloatRange()))

    def read_value(self):
        x0, x1 = self.rawrange
        y0, y1 = self.valuerange
        self.x = self.read(self.addr, self.scale)
        return y0 + (y1 - y0) * (self.x - x0) / (x1 - x0)


class VoltageInput(AnalogInput):
    scale = 1e5

    def initModule(self):
        super().initModule()
        self.write(f'{self.addr}_mode','U')


class LogVoltageInput(VoltageInput):

    def read_value(self):
        x0, x1 = self.rawrange
        y0, y1 = self.valuerange
        self.x = self.read(self.addr, self.scale)
        a = (x1-x0)/log(y1/y0,10)
        return 10**((self.x-x1)/a)*y1


class CurrentInput(AnalogInput):
    scale = 1e6
    rawrange = (0.004,0.02)

    def initModule(self):
        super().initModule()
        self.write(f'{self.addr}_mode','U')

    def read_value(self):
        result = super().read_value()
        if self.x > 0.021:
            self.status = ERROR, 'sensor fault'
            raise OutOfRangeError('sensor fault')
        self.status = IDLE, ''
        return result


class AnalogOutput(AnalogInput, Writable):
    target = Parameter('outputvalue', FloatRange())

    def write_target(self, value):
        x0, x1 = self.rawrange
        y0, y1 = self.valuerange
        self.write(self.addr, x0 + (x1 - x0) * (value - y0) / (y1 - y0),self.scale)


class VoltageOutput(AnalogOutput):
    rawrange = (0,10)
    scale = 1e3

    def initModule(self):
        super().initModule()
        self.write(f'{self.addr}_enabled', '0')
        self.write(f'{self.addr}_mode', 'V')
        self.write(f'{self.addr}', '0')
        self.write(f'{self.addr}_enabled', '1')


class VoltagePower(Base, Writable):
    target = Parameter(datatype=FloatRange(0, 24.5, unit='V'), default=12)
    addr = 'vso'

    def write_target(self, value):
        if value:
            self.write(self.addr, value, 1000)
            self.write(f'{self.addr}_enabled', 1)
        else:
            self.write(f'{self.addr}_enabled', 0)