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

enhance documentation

Browse Source 
- flatten hierarchy (some links do not work when using folders)
- add a tutorial for programming a simple driver
- clean description using inspect.cleandoc
+ fix a bug with 'unit' pseudo property in a Parameter used as override

Change-Id: I31ddba5d516d1ee5e785e28fbd79fca44ed23f5e
Reviewed-on: https://forge.frm2.tum.de/review/c/sine2020/secop/playground/+/25000
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
zolliker
2021-02-05 11:23:15 +01:00
parent a19425684c
commit ed02131a37
46 changed files with 1124 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
secop_psi/ccu4.py Normal file
View File

@ -0,0 +1,99 @@
# -*- coding: utf-8 -*-
# *****************************************************************************
#
# 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>
#
# *****************************************************************************
"""drivers for CCU4, the cryostat control unit at SINQ"""
# the most common Frappy classes can be imported from secop.core
from secop.core import Readable, Parameter, FloatRange, EnumType, StringIO, HasIodev
class CCU4IO(StringIO):
"""communication with CCU4"""
# for completeness: (not needed, as it is the default)
end_of_line = '\n'
# on connect, we send 'cid' and expect a reply starting with 'CCU4'
identification = [('cid', r'CCU4.*')]
# inheriting the HasIodev mixin creates us a private attribute *_iodev*
# for talking with the hardware
# Readable as a base class defines the value and status parameters
class HeLevel(HasIodev, Readable):
"""He Level channel of CCU4"""
# define the communication class to create the IO module
iodevClass = CCU4IO
# define or alter the parameters
# as Readable.value exists already, we give only the modified property 'unit'
value = Parameter(unit='%')
empty_length = Parameter('warm length when empty', FloatRange(0, 2000, unit='mm'),
readonly=False)
full_length = Parameter('warm length when full', FloatRange(0, 2000, unit='mm'),
readonly=False)
sample_rate = Parameter('sample rate', EnumType(slow=0, fast=1), readonly=False)
Status = Readable.Status
# conversion of the code from the CCU4 parameter 'hsf'
STATUS_MAP = {
0: (Status.IDLE, 'sensor ok'),
1: (Status.ERROR, 'sensor warm'),
2: (Status.ERROR, 'no sensor'),
3: (Status.ERROR, 'timeout'),
4: (Status.ERROR, 'not yet read'),
5: (Status.DISABLED, 'disabled'),
}
def query(self, cmd):
"""send a query and get the response
:param cmd: the name of the parameter to query or '<parameter>=<value'
for changing a parameter
:returns: the (new) value of the parameter
"""
name, txtvalue = self._iodev.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
def read_value(self):
return self.query('h')
def read_status(self):
return self.STATUS_MAP[int(self.query('hsf'))]
def read_empty_length(self):
return self.query('hem')
def write_empty_length(self, value):
return self.query('hem=%g' % value)
def read_full_length(self):
return self.query('hfu')
def write_full_length(self, value):
return self.query('hfu=%g' % value)
def read_sample_rate(self):
return self.query('hf')
def write_sample_rate(self, value):
return self.query('hf=%d' % value)

59
secop_psi/ppms.py
View File

@ -54,7 +54,15 @@ except ImportError:
class IOHandler(secop.iohandler.IOHandler):
CMDARGS = ['no']
"""IO handler for PPMS commands
deals with typical format:
- query command: ``<command>?``
- reply: ``<value1>,<value2>, ..``
- change command: ``<command> <value1>,<value2>,...``
"""
CMDARGS = ['no'] # the channel number is needed in channel commands
CMDSEPARATOR = None # no command chaining
def __init__(self, name, querycmd, replyfmt):
@ -63,7 +71,7 @@ class IOHandler(secop.iohandler.IOHandler):
class Main(Communicator):
"""general ppms dummy module"""
"""ppms communicator module"""
parameters = {
'pollinterval': Parameter('poll interval', readonly=False,
@ -126,6 +134,7 @@ class Main(Communicator):
class PpmsMixin(HasIodev, Module):
"""common methods for ppms modules"""
properties = {
'iodev': Attached(),
}
@ -143,25 +152,19 @@ class PpmsMixin(HasIodev, Module):
started_callback()
def read_value(self):
"""polling is done by the main module
and PPMS does not deliver really more fresh values when polled more often
"""
# polling is done by the main module
# and PPMS does not deliver really more fresh values when polled more often
return Done
def read_status(self):
"""polling is done by the main module
and PPMS does not deliver really fresh status values anyway: the status is not
changed immediately after a target change!
"""
# polling is done by the main module
# and PPMS does not deliver really fresh status values anyway: the status is not
# changed immediately after a target change!
return Done
def update_value_status(self, value, packed_status):
"""update value and status
to be reimplemented for modules looking at packed_status
"""
# update value and status
# to be reimplemented for modules looking at packed_status
if not self.enabled:
self.status = (self.Status.DISABLED, 'disabled')
return
@ -173,6 +176,7 @@ class PpmsMixin(HasIodev, Module):
class Channel(PpmsMixin, Readable):
"""channel base class"""
parameters = {
'value':
Override('main value of channels', poll=True),
@ -201,6 +205,8 @@ class Channel(PpmsMixin, Readable):
class UserChannel(Channel):
"""user channel"""
parameters = {
'pollinterval':
Override(visibility=3),
@ -223,6 +229,8 @@ class UserChannel(Channel):
class DriverChannel(Channel):
"""driver channel"""
drvout = IOHandler('drvout', 'DRVOUT? %(no)d', '%d,%g,%g')
parameters = {
@ -247,6 +255,8 @@ class DriverChannel(Channel):
class BridgeChannel(Channel):
"""bridge channel"""
bridge = IOHandler('bridge', 'BRIDGE? %(no)d', '%d,%g,%g,%d,%d,%g')
# pylint: disable=invalid-name
ReadingMode = Enum('ReadingMode', standard=0, fast=1, highres=2)
@ -306,11 +316,10 @@ class Level(PpmsMixin, Readable):
channel = 'level'
def update_value_status(self, value, packed_status):
"""must be a no-op
when called from Main.read_data, value is always None
value and status is polled via settings
"""
pass
# must be a no-op
# when called from Main.read_data, value is always None
# value and status is polled via settings
def analyze_level(self, level, status):
# ignore 'old reading' state of the flag, as this happens only for a short time
@ -377,7 +386,6 @@ class Chamber(PpmsMixin, Drivable):
channel = 'chamber'
def update_value_status(self, value, packed_status):
"""update value and status"""
status_code = (packed_status >> 8) & 0xf
if status_code in self.STATUS_MAP:
self.value = status_code
@ -390,10 +398,8 @@ class Chamber(PpmsMixin, Drivable):
return dict(target=target)
def change_chamber(self, change):
"""write settings, combining <pname>=<value> and current attributes
and request updated settings
"""
# write settings, combining <pname>=<value> and current attributes
# and request updated settings
if change.target == self.Operation.noop:
return None
return (change.target,)
@ -474,7 +480,6 @@ class Temp(PpmsMixin, Drivable):
_ramp_at_limit = False
def update_value_status(self, value, packed_status):
"""update value and status"""
if value is None:
self.status = (self.Status.ERROR, 'invalid value')
return
@ -649,7 +654,6 @@ class Field(PpmsMixin, Drivable):
_last_change = 0 # means no target change is pending
def update_value_status(self, value, packed_status):
"""update value and status"""
if value is None:
self.status = (self.Status.ERROR, 'invalid value')
return
@ -776,7 +780,6 @@ class Position(PpmsMixin, Drivable):
_within_target = 0 # time since we are within target
def update_value_status(self, value, packed_status):
"""update value and status"""
if not self.enabled:
self.status = (self.Status.DISABLED, 'disabled')
return