149 lines
4.7 KiB
Python
149 lines
4.7 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>
|
|
# *****************************************************************************
|
|
|
|
import os
|
|
from glob import glob
|
|
from frappy.core import Readable, Writable, Parameter, BoolType, StringType,\
|
|
FloatRange, Property, TupleOf, ERROR, IDLE
|
|
from frappy.errors import ConfigError
|
|
from math import log
|
|
|
|
basepaths = '/sys/class/ionopimax', '/sys/class/ionopi'
|
|
|
|
|
|
class Base:
|
|
addr = Property('address', StringType())
|
|
_devpath = None
|
|
devclass = None
|
|
|
|
def initModule(self):
|
|
super().initModule()
|
|
# candidates = glob(f'/sys/class/iono*/*/{self.addr}')
|
|
# if not candidates:
|
|
# raise ConfigError(f'can not find path for {self.addr}')
|
|
for basepath in basepaths:
|
|
for devclass in ([self.devclass] if isinstance(self.devclass, str) else self.devclass):
|
|
devpath = f'{basepath}/{devclass}'
|
|
if os.path.exists(devpath):
|
|
self._devpath = devpath
|
|
return
|
|
else:
|
|
self.log.info('%s does not exist', devpath)
|
|
else:
|
|
raise ConfigError(f'device path for {self.devclass} not found {devpath}')
|
|
|
|
def read(self, addr, scale=None):
|
|
with open(f'{self._devpath}/{addr}') as f:
|
|
result = f.read()
|
|
if scale:
|
|
return float(result) / scale
|
|
return result
|
|
|
|
def write(self, addr, value, scale=None):
|
|
value = str(round(value * scale)) if scale else str(value)
|
|
with open(f'{self._devpath}/{addr}', 'w') as f:
|
|
f.write(value)
|
|
|
|
|
|
class DigitalInput(Base, Readable):
|
|
value = Parameter('input state', BoolType())
|
|
devclass = 'digital_in'
|
|
|
|
def read_value(self):
|
|
return self.read(self.addr, 1)
|
|
|
|
|
|
class DigitalOutput(DigitalInput, Writable):
|
|
target = Parameter('output state', BoolType(), readonly=False)
|
|
devclass = 'digital_out', 'relay'
|
|
|
|
def write_target(self, value):
|
|
self.write(self.addr, value, 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()))
|
|
devclass = 'analog_in'
|
|
|
|
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 broken'
|
|
else:
|
|
self.status = IDLE, ''
|
|
return result
|
|
|
|
|
|
class AnalogOutput(AnalogInput, Writable):
|
|
target = Parameter('outputvalue', FloatRange())
|
|
devclass = 'analog_out'
|
|
|
|
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')
|