linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

unify name and module on Attached property

Browse Source 
- setting the attribute using the name of an attached module
- getting the attribute results in the module object

+ change names iodev to io, iodevClass to ioClass,
  sendRecv to communicate, HasIodev to HasIO

Change-Id: I200b63a5a7dc1453bf6ac998782b065645201900
Reviewed-on: https://forge.frm2.tum.de/review/c/sine2020/secop/playground/+/27575
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
zolliker
2022-01-27 18:11:42 +01:00
parent b911bc1838
commit c1307cdd03
21 changed files with 301 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
secop_psi/ah2700.py
View File

@@ -20,7 +20,7 @@
# *****************************************************************************
"""Andeen Hagerling capacitance bridge"""
from secop.core import Done, FloatRange, HasIodev, Parameter, Readable, StringIO
from secop.core import Done, FloatRange, HasIO, Parameter, Readable, StringIO
class Ah2700IO(StringIO):
@@ -28,19 +28,19 @@ class Ah2700IO(StringIO):
timeout = 5
class Capacitance(HasIodev, Readable):
class Capacitance(HasIO, Readable):
value = Parameter('capacitance', FloatRange(unit='pF'), poll=True)
freq = Parameter('frequency', FloatRange(unit='Hz'), readonly=False, default=0)
voltage = Parameter('voltage', FloatRange(unit='V'), readonly=False, default=0)
loss = Parameter('loss', FloatRange(unit='deg'), default=0)
iodevClass = Ah2700IO
ioClass = Ah2700IO
def parse_reply(self, reply):
if reply.startswith('SI'): # this is an echo
self.sendRecv('SERIAL ECHO OFF')
reply = self.sendRecv('SI')
self.communicate('SERIAL ECHO OFF')
reply = self.communicate('SI')
if not reply.startswith('F='): # this is probably an error message like "LOSS TOO HIGH"
self.status = [self.Status.ERROR, reply]
return
@@ -59,14 +59,14 @@ class Capacitance(HasIodev, Readable):
if lossunit == 'DS':
self.loss = loss
else: # the unit was wrong, we want DS = tan(delta), not NS = nanoSiemens
reply = self.sendRecv('UN DS').split() # UN DS returns a reply similar to SI
reply = self.communicate('UN DS').split() # UN DS returns a reply similar to SI
try:
self.loss = reply[7]
except IndexError:
pass # don't worry, loss will be updated next time
def read_value(self):
self.parse_reply(self.sendRecv('SI')) # SI = single trigger
self.parse_reply(self.communicate('SI')) # SI = single trigger
return Done
def read_freq(self):
@@ -77,14 +77,14 @@ class Capacitance(HasIodev, Readable):
self.read_value()
return Done
def read_volt(self):
def read_voltage(self):
self.read_value()
return Done
def write_freq(self, value):
self.parse_reply(self.sendRecv('FR %g;SI' % value))
self.parse_reply(self.communicate('FR %g;SI' % value))
return Done
def write_volt(self, value):
self.parse_reply(self.sendRecv('V %g;SI' % value))
def write_voltage(self, value):
self.parse_reply(self.communicate('V %g;SI' % value))
return Done

11
secop_psi/ccu4.py
View File

@@ -23,7 +23,7 @@
"""drivers for CCU4, the cryostat control unit at SINQ"""
# the most common Frappy classes can be imported from secop.core
from secop.core import EnumType, FloatRange, \
HasIodev, Parameter, Readable, StringIO
HasIO, Parameter, Readable, StringIO
class CCU4IO(StringIO):
@@ -34,14 +34,13 @@ class CCU4IO(StringIO):
identification = [('cid', r'CCU4.*')]
# inheriting the HasIodev mixin creates us a private attribute *_iodev*
# for talking with the hardware
# inheriting HasIO allows us to use the communicate method for talking with the hardware
# Readable as a base class defines the value and status parameters
class HeLevel(HasIodev, Readable):
class HeLevel(HasIO, Readable):
"""He Level channel of CCU4"""
# define the communication class to create the IO module
iodevClass = CCU4IO
ioClass = CCU4IO
# define or alter the parameters
# as Readable.value exists already, we give only the modified property 'unit'
@@ -71,7 +70,7 @@ class HeLevel(HasIodev, Readable):
for changing a parameter
:returns: the (new) value of the parameter
"""
name, txtvalue = self._iodev.communicate(cmd).split('=')
name, txtvalue = self.communicate(cmd).split('=')
assert name == cmd.split('=')[0] # check that we got a reply to our command
return txtvalue # Frappy will automatically convert the string to the needed data type

56
secop_psi/k2601b.py
View File

@@ -23,7 +23,7 @@
not tested yet"""
from secop.core import Attached, BoolType, EnumType, FloatRange, \
HasIodev, Module, Parameter, StringIO, Writable
HasIO, Module, Parameter, StringIO, Writable
class K2601bIO(StringIO):
@@ -41,7 +41,7 @@ SOURCECMDS = {
}
class SourceMeter(HasIodev, Module):
class SourceMeter(HasIO, Module):
resistivity = Parameter('readback resistivity', FloatRange(unit='Ohm'), poll=True)
power = Parameter('readback power', FloatRange(unit='W'), poll=True)
@@ -49,19 +49,19 @@ class SourceMeter(HasIodev, Module):
readonly=False, default=0)
active = Parameter('output enable', BoolType(), readonly=False, poll=True)
iodevClass = K2601bIO
ioClass = K2601bIO
def read_resistivity(self):
return self.sendRecv('print(smua.measure.r())')
return self.communicate('print(smua.measure.r())')
def read_power(self):
return self.sendRecv('print(smua.measure.p())')
return self.communicate('print(smua.measure.p())')
def read_active(self):
return self.sendRecv('print(smua.source.output)')
return self.communicate('print(smua.source.output)')
def write_active(self, value):
return self.sendRecv('smua.source.output = %d print(smua.source.output)' % value)
return self.communicate('smua.source.output = %d print(smua.source.output)' % value)
# for now, mode will not be read from hardware
@@ -69,11 +69,11 @@ class SourceMeter(HasIodev, Module):
if value == 0:
self.write_active(0)
else:
self.sendRecv(SOURCECMDS[value] + ' print(0)')
self.communicate(SOURCECMDS[value] + ' print(0)')
return value
class Current(HasIodev, Writable):
class Current(HasIO, Writable):
sourcemeter = Attached()
value = Parameter('measured current', FloatRange(unit='A'), poll=True)
@@ -82,41 +82,41 @@ class Current(HasIodev, Writable):
limit = Parameter('current limit', FloatRange(0, 2.0, unit='A'), default=2, poll=True)
def read_value(self):
return self.sendRecv('print(smua.measure.i())')
return self.communicate('print(smua.measure.i())')
def read_target(self):
return self.sendRecv('print(smua.source.leveli)')
return self.communicate('print(smua.source.leveli)')
def write_target(self, value):
if not self.active:
raise ValueError('current source is disabled')
if value > self.limit:
raise ValueError('current exceeds limit')
return self.sendRecv('smua.source.leveli = %g print(smua.source.leveli)' % value)
return self.communicate('smua.source.leveli = %g print(smua.source.leveli)' % value)
def read_limit(self):
if self.active:
return self.limit
return self.sendRecv('print(smua.source.limiti)')
return self.communicate('print(smua.source.limiti)')
def write_limit(self, value):
if self.active:
return value
return self.sendRecv('smua.source.limiti = %g print(smua.source.limiti)' % value)
return self.communicate('smua.source.limiti = %g print(smua.source.limiti)' % value)
def read_active(self):
return self._sourcemeter.mode == 1 and self._sourcemeter.read_active()
return self.sourcemeter.mode == 1 and self.sourcemeter.read_active()
def write_active(self, value):
if self._sourcemeter.mode != 1:
if self.sourcemeter.mode != 1:
if value:
self._sourcemeter.write_mode(1) # switch to current
self.sourcemeter.write_mode(1) # switch to current
else:
return 0
return self._sourcemeter.write_active(value)
return self.sourcemeter.write_active(value)
class Voltage(HasIodev, Writable):
class Voltage(HasIO, Writable):
sourcemeter = Attached()
value = Parameter('measured voltage', FloatRange(unit='V'), poll=True)
@@ -125,35 +125,35 @@ class Voltage(HasIodev, Writable):
limit = Parameter('current limit', FloatRange(0, 2.0, unit='V'), default=2, poll=True)
def read_value(self):
return self.sendRecv('print(smua.measure.v())')
return self.communicate('print(smua.measure.v())')
def read_target(self):
return self.sendRecv('print(smua.source.levelv)')
return self.communicate('print(smua.source.levelv)')
def write_target(self, value):
if not self.active:
raise ValueError('voltage source is disabled')
if value > self.limit:
raise ValueError('voltage exceeds limit')
return self.sendRecv('smua.source.levelv = %g print(smua.source.levelv)' % value)
return self.communicate('smua.source.levelv = %g print(smua.source.levelv)' % value)
def read_limit(self):
if self.active:
return self.limit
return self.sendRecv('print(smua.source.limitv)')
return self.communicate('print(smua.source.limitv)')
def write_limit(self, value):
if self.active:
return value
return self.sendRecv('smua.source.limitv = %g print(smua.source.limitv)' % value)
return self.communicate('smua.source.limitv = %g print(smua.source.limitv)' % value)
def read_active(self):
return self._sourcemeter.mode == 2 and self._sourcemeter.read_active()
return self.sourcemeter.mode == 2 and self.sourcemeter.read_active()
def write_active(self, value):
if self._sourcemeter.mode != 2:
if self.sourcemeter.mode != 2:
if value:
self._sourcemeter.write_mode(2) # switch to voltage
self.sourcemeter.write_mode(2) # switch to voltage
else:
return 0
return self._sourcemeter.write_active(value)
return self.sourcemeter.write_active(value)

24
secop_psi/ls370res.py
View File

@@ -28,7 +28,7 @@ from secop.lib import formatStatusBits
from secop.modules import Attached, Done, \
Drivable, Parameter, Property, Readable
from secop.poller import REGULAR, Poller
from secop.io import HasIodev
from secop.io import HasIO
Status = Drivable.Status
@@ -58,7 +58,7 @@ class StringIO(secop.io.StringIO):
wait_before = 0.05
class Main(HasIodev, Drivable):
class Main(HasIO, Drivable):
value = Parameter('the current channel', poll=REGULAR, datatype=IntRange(0, 17))
target = Parameter('channel to select', datatype=IntRange(0, 17))
@@ -66,7 +66,7 @@ class Main(HasIodev, Drivable):
pollinterval = Parameter(default=1, export=False)
pollerClass = Poller
iodevClass = StringIO
ioClass = StringIO
_channel_changed = 0 # time of last channel change
_channels = None # dict <channel no> of <module object>
@@ -81,7 +81,7 @@ class Main(HasIodev, Drivable):
super().startModule(start_events)
for ch in range(1, 16):
if ch not in self._channels:
self.sendRecv('INSET %d,0,0,0,0,0;INSET?%d' % (ch, ch))
self.communicate('INSET %d,0,0,0,0,0;INSET?%d' % (ch, ch))
def read_value(self):
channel, auto = scan.send_command(self)
@@ -114,7 +114,7 @@ class Main(HasIodev, Drivable):
def write_target(self, channel):
scan.send_change(self, channel, self.autoscan)
# self.sendRecv('SCAN %d,%d;SCAN?' % (channel, self.autoscan))
# self.communicate('SCAN %d,%d;SCAN?' % (channel, self.autoscan))
if channel != self.value:
self.value = 0
self._channel_changed = time.time()
@@ -123,11 +123,11 @@ class Main(HasIodev, Drivable):
def write_autoscan(self, value):
scan.send_change(self, self.value, value)
# self.sendRecv('SCAN %d,%d;SCAN?' % (channel, self.autoscan))
# self.communicate('SCAN %d,%d;SCAN?' % (channel, self.autoscan))
return value
class ResChannel(HasIodev, Readable):
class ResChannel(HasIO, Readable):
"""temperature channel on Lakeshore 336"""
RES_RANGE = {key: i+1 for i, key in list(
@@ -142,7 +142,7 @@ class ResChannel(HasIodev, Readable):
for val in [2, 6.32, 20, 63.2, 200, 632]))}
pollerClass = Poller
iodevClass = StringIO
ioClass = StringIO
_main = None # main module
_last_range_change = 0 # time of last range change
@@ -183,7 +183,7 @@ class ResChannel(HasIodev, Readable):
return Done
# we got here, when we missed the idle state of self._main
self._trigger_read = False
result = self.sendRecv('RDGR?%d' % self.channel)
result = self.communicate('RDGR?%d' % self.channel)
result = float(result)
if self.autorange == 'soft':
now = time.time()
@@ -216,9 +216,9 @@ class ResChannel(HasIodev, Readable):
def read_status(self):
if not self.enabled:
return [self.Status.DISABLED, 'disabled']
if self.channel != self._main.value:
if self.channel != self.main.value:
return Done
result = int(self.sendRecv('RDGST?%d' % self.channel))
result = int(self.communicate('RDGST?%d' % self.channel))
result &= 0x37 # mask T_OVER and T_UNDER (change this when implementing temperatures instead of resistivities)
statustext = ' '.join(formatStatusBits(result, STATUS_BIT_LABELS))
if statustext:
@@ -283,5 +283,5 @@ class ResChannel(HasIodev, Readable):
def write_enabled(self, value):
inset.write(self, 'enabled', value)
if value:
self._main.write_target(self.channel)
self.main.write_target(self.channel)
return Done

12
secop_psi/ppms.py
View File

@@ -40,10 +40,10 @@ from secop.datatypes import BoolType, EnumType, \
from secop.errors import HardwareError
from secop.lib import clamp
from secop.lib.enum import Enum
from secop.modules import Attached, Communicator, Done, \
from secop.modules import Communicator, Done, \
Drivable, Parameter, Property, Readable
from secop.poller import Poller
from secop.io import HasIodev
from secop.io import HasIO
try:
import secop_psi.ppmswindows as ppmshw
@@ -130,10 +130,8 @@ class Main(Communicator):
return data # return data as string
class PpmsBase(HasIodev, Readable):
class PpmsBase(HasIO, Readable):
"""common base for all ppms modules"""
iodev = Attached()
# polling is done by the main module
# and PPMS does not deliver really more fresh values when polled more often
value = Parameter(poll=False, needscfg=False)
@@ -150,7 +148,7 @@ class PpmsBase(HasIodev, Readable):
def initModule(self):
super().initModule()
self._iodev.register(self)
self.io.register(self)
def update_value_status(self, value, packed_status):
# update value and status
@@ -197,7 +195,7 @@ class UserChannel(Channel):
datatype=StringType(), export=False, default='')
def write_enabled(self, enabled):
other = self._iodev.modules.get(self.linkenable, None)
other = self.io.modules.get(self.linkenable, None)
if other:
other.enabled = enabled
return enabled

14
secop_psi/trinamic.py
View File

@@ -26,7 +26,7 @@ import time
import struct
from secop.core import BoolType, Command, EnumType, FloatRange, IntRange, \
HasIodev, Parameter, Property, Drivable, PersistentMixin, PersistentParam, Done
HasIO, Parameter, Property, Drivable, PersistentMixin, PersistentParam, Done
from secop.io import BytesIO
from secop.errors import CommunicationFailedError, HardwareError, BadValueError, IsBusyError
from secop.rwhandler import ReadHandler, WriteHandler
@@ -77,7 +77,7 @@ def writable(*args, **kwds):
return PersistentParam(*args, readonly=False, poll=True, initwrite=True, **kwds)
class Motor(PersistentMixin, HasIodev, Drivable):
class Motor(PersistentMixin, HasIO, Drivable):
address = Property('module address', IntRange(0, 255), default=1)
value = Parameter('motor position', FloatRange(unit='deg', fmtstr='%.3f'))
@@ -116,7 +116,7 @@ class Motor(PersistentMixin, HasIodev, Drivable):
readonly=False, default=0, poll=True, visibility=3, group='more')
pollinterval = Parameter(group='more')
iodevClass = BytesIO
ioClass = BytesIO
fast_pollfactor = 0.001 # poll as fast as possible when busy
_started = 0
_calcTimeout = True
@@ -131,20 +131,20 @@ class Motor(PersistentMixin, HasIodev, Drivable):
:param value: if given, the parameter is written, else it is returned
:return: the returned value
"""
if self._calcTimeout and self._iodev._conn:
if self._calcTimeout and self.io._conn:
self._calcTimeout = False
baudrate = getattr(self._iodev._conn.connection, 'baudrate', None)
baudrate = getattr(self.io._conn.connection, 'baudrate', None)
if baudrate:
if baudrate not in BAUDRATES:
raise CommunicationFailedError('unsupported baud rate: %d' % baudrate)
self._iodev.timeout = 0.03 + 200 / baudrate
self.io.timeout = 0.03 + 200 / baudrate
exc = None
byt = struct.pack('>BBBBi', self.address, cmd, adr, bank, round(value))
byt += bytes([sum(byt) & 0xff])
for itry in range(3,0,-1):
try:
reply = self._iodev.communicate(byt, 9)
reply = self.communicate(byt, 9)
if sum(reply[:-1]) & 0xff != reply[-1]:
raise CommunicationFailedError('checksum error')
# will try again