197 lines
6.3 KiB
Python
197 lines
6.3 KiB
Python
# *****************************************************************************
|
|
# 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 math import log
|
|
from pathlib import Path
|
|
from frappy.core import Readable, Writable, Parameter, Property, ERROR, IDLE, WARN
|
|
from frappy.errors import ConfigError, OutOfRangeError, ProgrammingError
|
|
from frappy.datatypes import BoolType, EnumType, FloatRange, NoneOr, StringType, TupleOf
|
|
|
|
|
|
class Base:
|
|
addr = Property('address', StringType())
|
|
_devpath = None
|
|
_devclass = None
|
|
_status = IDLE, ''
|
|
|
|
def initModule(self):
|
|
super().initModule()
|
|
self.log.info('initModule %r', self.name)
|
|
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):
|
|
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 = IDLE, ''
|
|
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
|
|
self.read_status()
|
|
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()))
|
|
extendedrange = Property('range outside calibrated range, but not sensor fault',
|
|
NoneOr(TupleOf(FloatRange(), FloatRange())), default=None)
|
|
|
|
def initModule(self):
|
|
super().initModule()
|
|
dt = self.parameters['value'].datatype
|
|
dt.min, dt.max = self.valuerange
|
|
|
|
def read_value(self):
|
|
x0, x1 = self.rawrange
|
|
y0, y1 = self.valuerange
|
|
self.x = self.read(self.addr, self.scale)
|
|
self.read_status()
|
|
if self.status[0] == ERROR:
|
|
raise OutOfRangeError('sensor fault')
|
|
return y0 + (y1 - y0) * (self.x - x0) / (x1 - x0)
|
|
|
|
def read_status(self):
|
|
if self.rawrange[0] <= self.x <= self.rawrange[1]:
|
|
return IDLE, ''
|
|
if self.extendedrange is None or self.extendedrange[0] <= self.x <= self.extendedrange[1]:
|
|
return WARN, 'out of range'
|
|
return ERROR, 'sensor fault'
|
|
|
|
|
|
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)
|
|
self.read_status()
|
|
if self.status[0] == ERROR:
|
|
raise OutOfRangeError('sensor fault')
|
|
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')
|
|
|
|
|
|
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.log.info('write vso %r', value)
|
|
self.write(self.addr, value, 1000)
|
|
self.write(f'{self.addr}_enabled', 1)
|
|
else:
|
|
self.write(f'{self.addr}_enabled', 0)
|
|
|
|
|