frappy_psi: add electromagnet (IO read implemented)

2026-06-09 14:21:21 +02:00
parent 45b3246a8b
commit 62fe2a86ba
1 changed files with 107 additions and 0 deletions
@@ -0,0 +1,107 @@
+# *****************************************************************************
+# 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>
+#   Anik Stark <anik.stark@psi.ch>
+# *****************************************************************************
+
+import struct
+from frappy.core import BytesIO, HasIO, Parameter, Property, IntRange, FloatRange
+from frappy.errors import CommunicationFailedError
+
+
+class Item:
+
+    def __init__(self, unit, description, reference, value):
+        self.unit = unit
+        self.description = description
+        self.reference = reference
+        self.value = value
+
+
+class IO(BytesIO):
+
+    db = Property('database number', datatype=IntRange(0, 255), default=200)
+    _db_length = 0
+
+    def checkHWIdent(self):
+        reply = self.send_fetch_message(1)    # length in words
+        pattern = struct.pack('>2s14B', b'S5', 16, 1, 3, 6, 15, 3, 0, 255, 7, 0, 0, 0, 0, 0)
+        if reply[0:16] != pattern:
+            raise CommunicationFailedError(f'bad response: {reply}')
+        self._db_length = struct.unpack('>H', reply[16:18])[0] // 2    # length in words (1 word = 2 bytes = 16 bit)
+
+    def send_fetch_message(self, length):
+        msg = struct.pack('>2s10BH2B', b'S5', 16, 1, 3, 5, 3, 8, 1, self.db, 0, 0, length, 255, 2)
+        reply = self.communicate(msg, 16 + length * 2)
+        return reply
+
+    def get_string(self, pos, reply):
+        size, length = reply[pos:pos+2]    # reserved space, actual length (containing info)
+        pos += 2
+        value = reply[pos:pos+length].decode('latin-1')
+        pos += size
+        return pos, value
+
+    def get_data(self):
+        reply = self.send_fetch_message(self._db_length)
+        # self.log.info('get_data reply %s', reply)
+        pos = 16 + 2
+        result = {}
+        while pos < len(reply):
+            typ, alarm = reply[pos:pos+2]
+            pos += 2
+            pos, name = self.get_string(pos, reply)
+            pos, unit = self.get_string(pos, reply)
+            pos, description = self.get_string(pos, reply)
+            pos, reference = self.get_string(pos, reply)
+            self.log.info('pos %d', pos-16+2)
+            if typ == 1:
+                # value = struct.unpack('>H', reply[pos:pos+2])[0]
+                value = None
+                pos += 2
+            elif typ == 4:
+                value = struct.unpack('>H', reply[pos:pos+2])[0]
+                pos += 2
+            elif typ == 2:
+                value = struct.unpack('>I', reply[pos:pos+4])[0]
+                pos += 4
+            elif typ == 3:
+                _, value = struct.unpack('>Hf', reply[pos:pos+6])
+                pos += 6
+            res = {
+                'name': name,
+                'unit': unit,
+                'description': description,
+                'reference': reference,
+                'typ': typ,
+                'alarm': alarm,
+                'value': value
+            }
+            result[name, description] = res
+            self.log.info(res)
+
+
+class Magnet(HasIO):
+
+    ioClass = IO
+
+    value = Parameter('value', datatype=FloatRange())
+
+    def read_value(self):
+        self.io.get_data()
+        return 0
+