Feature/turbovac

Dockerfile

@@ -1,7 +1,7 @@
 ARG img_user=ghcr.io/driplineorg
 ARG img_repo=dripline-python
 #ARG img_tag=develop-dev
-ARG img_tag=v5.1.0
+ARG img_tag=v5.1.1
 
 FROM ${img_user}/${img_repo}:${img_tag}
 

MANIFEST.in

@@ -0,0 +1 @@
+recursive-include dragonfly *.txt

docker-compose.yaml

@@ -55,3 +55,21 @@ services:
       - /usr/local/src_dragonfly/dragonfly/watchdog.py
       - --config
       - /root/AlarmSystem.yaml
+
+  Turbo1:
+    image: dragonfly_turbovac:latest
+    depends_on:
+      - rabbit-broker
+    volumes:
+      - ./examples/turbo_1.yaml:/root/devices/turbo_1.yaml
+    environment:
+      - DRIPLINE_USER=dripline
+      - DRIPLINE_PASSWORD=dripline
+    command:
+      - dl-serve
+      - -c
+      - /root/devices/turbo_1.yaml
+      - -b
+      - rabbit-broker
+      - -vvv
+

dripline/extensions/__init__.py

@@ -4,6 +4,7 @@
 
 # Subdirectories
 from . import jitter
+from . import turbovac
 
 # Modules in this directory
 

dripline/extensions/turbovac/README.md

@@ -0,0 +1,7 @@
+The files contained in telegram are a copy from the telegram folder of
+https://github.com/fkivela/TurboCtl/tree/work
+I modified the files to be compatible with python versions < 3.10, by changing some of the newer syntax features to clasical implementations.
+We use the TurboCtl.telegram to generate the content of the telegram but use the dripline service to send it. Thus we can not make use of any of the other implementations.
+
+The TurboVac Ethernet Service implements the USS protocol which consists of a STX-byte (\x02), LENGTH-byte, ADDRESS-byte, payload, XOR-checksum-byte.
+The Payload is configured within the Endpoint class and makes use of the TurboCtl.telegram module. 

dripline/extensions/turbovac/__init__.py

@@ -0,0 +1,4 @@
+__all__ = []
+
+from .ethernet_uss_service import *
+from .turbovac_endpoint import *

dripline/extensions/turbovac/ethernet_uss_service.py

@@ -0,0 +1,143 @@
+import re
+import socket
+import threading
+
+from dripline.core import Service, ThrowReply
+
+import logging
+logger = logging.getLogger(__name__)
+
+__all__ = []
+
+
+__all__.append('EthernetUSSService')
+class EthernetUSSService(Service):
+    '''
+    '''
+    def __init__(self,
+                 socket_timeout=1.0,
+                 socket_info=('localhost', 1234),
+                 **kwargs
+                 ):
+        '''
+        Args:
+            socket_timeout (int): number of seconds to wait for a reply from the device before timeout.
+            socket_info (tuple or string): either socket.socket.connect argument tuple, or string that
+                parses into one.
+        '''
+        Service.__init__(self, **kwargs)
+
+        if isinstance(socket_info, str):
+            print(f"Formatting socket_info: {socket_info}")
+            re_str = "\([\"'](\S+)[\"'], ?(\d+)\)"
+            (ip,port) = re.findall(re_str,socket_info)[0]
+            socket_info = (ip,int(port))
+
+        self.alock = threading.Lock()
+        self.socket = socket.socket()
+        self.socket_timeout = float(socket_timeout)
+        self.socket_info = socket_info
+        self.STX = b'\x02'
+        self._reconnect()
+
+    def _reconnect(self):
+        '''
+        Method establishing socket to ethernet instrument.
+        Called by __init__ or send (on failed communication).
+        '''
+        # close previous connection
+        # open new connection
+        # check if connection was successful
+
+        self.socket.close()
+        self.socket = socket.socket()
+        try:
+            self.socket = socket.create_connection(self.socket_info, self.socket_timeout)
+        except (socket.error, socket.timeout) as err:
+            print(f"connection {self.socket_info} refused: {err}")
+            raise Exception('resource_error_connection', f"Unable to establish ethernet socket {self.socket_info}")
+        print(f"Ethernet socket {self.socket_info} established")
+
+    def send_to_device(self, commands, **kwargs):
+        '''
+        Standard device access method to communicate with instrument.
+        NEVER RENAME THIS METHOD!
+
+        commands (list||None): list of command(s) to send to the instrument following (re)connection to the instrument, still must return a reply!
+                             : if impossible, set as None to skip
+        '''
+        if not isinstance(commands, list):
+            commands = [commands]
+
+        self.alock.acquire()
+
+        try:
+            data = self._send_commands(commands)
+        except socket.error as err:
+            print(f"socket.error <{err}> received, attempting reconnect")
+            self._reconnect()
+            data = self._send_commands(commands)
+            print("Ethernet connection reestablished")
+        # exceptions.DriplineHardwareResponselessError
+        except Exception as err:
+            print(str(err))
+            try:
+                self._reconnect()
+                data = self._send_commands(commands)
+                print("Query successful after ethernet connection recovered")
+            except socket.error: # simply trying to make it possible to catch the error below
+                print("Ethernet reconnect failed, dead socket")
+                raise Exception('resource_error_connection', "Broken ethernet socket")
+            except Exception as err: ##TODO handle all exceptions, that seems questionable
+                print("Query failed after successful ethernet socket reconnect")
+                raise Exception('resource_error_no_response', str(err))
+        finally:
+            self.alock.release()
+        to_return = b''.join(data)
+        print(f"should return:\n{to_return}")
+        return to_return
+
+    def _send_commands(self, commands):
+        '''
+        Take a list of telegrams, send to instrument and receive responses, do any necessary formatting.
+
+        commands (list||None): list of command(s) to send to the instrument following (re)connection to the instrument, still must return a reply!
+                             : if impossible, set as None to skip
+        '''
+        all_data=[]
+
+        for command in commands:
+            if not isinstance(command, bytes):
+                raise ValueError("Command is not of type bytes: {command}, {type(command)}")
+            print(f"sending: {command}")
+            self.socket.send(command)
+            data = self._listen()
+            print(f"sync: {repr(command)} -> {repr(data)}")
+            all_data.append(data)
+        return all_data
+
+    def _listen(self):
+        '''
+        Query socket for response.
+        '''
+        data = b''
+        length = None
+        try:
+            while True:
+                tmp = self.socket.recv(1024)
+                data += tmp
+                if self.STX in data:
+                    start_idx = data.find(self.STX)
+                    data = data[start_idx:] # get rid of everything before the start
+                if len(data)>1:             # if >1 data we have a length info
+                    length = int(data[1])+2
+                    if len(data) >= length:  # if we are >= LENGTH we have all we need
+                        break
+                if tmp == '':
+                    raise Exception('resource_error_no_response', "Empty socket.recv packet")
+        except socket.timeout:
+            print(f"socket.timeout condition met; received:\n{repr(data)}")
+            raise Exception('resource_error_no_response', "Timeout while waiting for a response from the instrument")
+        print(repr(data))
+        data = data[0:length]
+        return data

dripline/extensions/turbovac/telegram/__init__.py

@@ -0,0 +1,3 @@
+"""This package is used to form telegrams that send commands to the 
+pump.
+"""

dripline/extensions/turbovac/telegram/api.py

@@ -0,0 +1,127 @@
+"""This module defines an API of functions which can be used to send commands
+to the pump.
+
+All functions in this module share the following common arguments:
+
+    *connection*:
+        This is a :class:`serial.Serial` instance, which is used to send the
+        command.
+
+    *pump_on*:
+        If this evaluates to ``True``, control bits telling the pump to turn or
+        stay on are added to the telegram. Otherwise receiving the telegram
+        will cause the pump to turn off if it is on.  
+
+If a command cannot be sent due to an error in the connection, a
+:class:`serial.SerialException` will be raised.
+
+The functions return both the query sent to the pump and the reply received
+back as :class:`~turboctl.telegram.telegram.TelegramReader` instances.
+"""
+
+from .codes import ControlBits
+from .telegram import (Telegram, TelegramBuilder, 
+                                        TelegramReader)
+
+
+_PUMP_ON_BITS = [ControlBits.COMMAND, ControlBits.ON]
+_PUMP_OFF_BITS = [ControlBits.COMMAND]
+
+
+def send(connection, telegram):
+    """Send *telegram* to the pump.
+
+    Args:
+        telegram: A :class:`~turboctl.telegram.telegram.Telegram` instance.
+    """
+    connection.write(bytes(telegram))
+    reply_bytes = connection.read(Telegram.LENGTH)
+    reply = TelegramBuilder().from_bytes(reply_bytes).build()
+    return TelegramReader(telegram, 'query'), TelegramReader(reply, 'reply')
+
+
+def status(connection, pump_on=None):
+    """Request pump status.
+
+    This function sends an empty telegram to the pump, which causes it to send
+    back a reply containing some data about the status of the pump.
+
+    This can also be used for turning the pump on or off by setting *pump_on*
+    to ``True`` or ``False``.
+    """
+    builder = TelegramBuilder()
+    if pump_on:
+        builder.set_flag_bits(_PUMP_ON_BITS)
+    else:
+        builder.set_flag_bits(_PUMP_OFF_BITS)
+
+    query = builder.build()
+    return send(connection, query)
+
+
+def reset_error(connection):
+    """Reset the error status of the pump.
+
+    This function sends a "reset error" command to the pump.
+    """
+    builder = TelegramBuilder()
+    clear_error = [ControlBits.COMMAND, ControlBits.RESET_ERROR]
+    builder.set_flag_bits(clear_error)
+    query = builder.build()
+    return send(connection, query)
+
+
+def _access_parameter(connection, mode, number, value, index, pump_on):
+    """This auxiliary function provides functionality for both reading and
+    writing parameter values, since the processes are very similar.
+    """
+    builder = (TelegramBuilder()
+        .set_parameter_mode(mode)
+        .set_parameter_number(number)
+        .set_parameter_index(index)
+        .set_parameter_value(value)
+    )
+
+    if pump_on:
+        builder.set_flag_bits(_PUMP_ON_BITS)
+
+    query = builder.build()
+    return send(connection, query)
+
+
+def read_parameter(connection, number, index=0, pump_on=True):
+    """Read the value of an index of a parameter.
+
+    Args:
+        number
+            The number of the parameter.
+
+        index
+            The index of the parameter (0 for unindexed parameters).
+
+    Raises:
+        ValueError:
+            If *number* or *index* have invalid values.
+    """
+    return _access_parameter(connection, 'read', number, 0, index, pump_on)
+
+
+def write_parameter(connection, number, value, index=0, pump_on=True):
+    """Write a value to an index of a parameter.
+
+    Args:
+        number:
+            The number of the parameter.
+
+        value:
+            The value to be written.
+
+        index:
+            The index of the parameter (0 for unindexed parameters).
+
+    Raises:
+        ValueError:
+            If *number* or *index* have invalid values.
+    """
+    return _access_parameter(
+        connection, 'write', number, value, index, pump_on)