Tutorial: Writing a pyhard2 driver¶

Declaring commands¶

We start by declaring the list of commands as it generally does not required any understanding of the protocol itself but simply to translate the instructions given in the manual to pyhard2. In the present case, the manual describes the protocol in form of large tables. We provide an extract of such a table here, limiting ourselves to the most relevant elements.

There are actually over 60 commands given so that, even if we limit ourselves to the 7 commands given above now, we should organize the driver in a meaningful manner. That will allow us add other useful commands later as the needs emerge. We simply organize the commands in the same way as the menus on the hardware controller.

The DE1, IT1 and PB1 commands that set PID values in the controller are in the operation pid menu. C1, C2, PWR, and SP1 are not in any menu. In pyhard2, commands are organized in a Subsystem and Subsystems can be nested. The hardware controller menus can thus be represented by pyhard2‘s Subsystems very straightforwardly. After the necessary imports:

import unittest
import pyhard2.driver as drv
# We define shortcuts.
Cmd, Access = drv.Command, drv.Access
# Access.RW for read-write commands (default),
# Access.RO for read-only, and Access.WO for write-only.

here is the skeleton of the driver with the most important menus:

class Series982(drv.Subsystem):

   """Driver to Watlow Series 982 controller."""

   def __init__(self, socket):
      self.operation = drv.Subsystem(self)
      self.operation.pid = drv.Subsystem(self.operation)
      # For the sake of the example:
      self.operation.system = drv.Subsystem(self.operation)
      self.setup = drv.Subsystem(self)
      self.setup.output = drv.Subsystem(self.setup)
      self.setup.global_ = drv.Subsystem(self.setup)
      self.factory = drv.Subsystem(self)
      self.factory.lockout = drv.Subsystem(self.factory)
      self.factory.diagnostic = drv.Subsystem(self.factory)
      self.factory.calibration = drv.Subsystem(self.factory)

The commands themselves are declared as Command and take the command name given in the manual (C1, C2, etc.) and a series of optional arguments like access (read-only, read-write, or write-only), minimum and maximum values, conversion functions, etc., such that we can now complete the driver, now limiting ourselves to the most relevant commands:

class Series982(drv.Subsystem):

    """Driver to Watlow Series 982 controller."""

    def __init__(self, socket, parent=None):
        super(Series982, self).__init__(parent)
        self.setProtocol(XonXoffProtocol(socket))
        self.temperature1 = Cmd("C1", access=Access.RO)
        self.temperature2 = Cmd("C2", access=Access.RO)
        self.setpoint = Cmd("SP1", minimum=-250, maximum=9999)  # assume deg C
        self.power = Cmd("PWR", access=Access.RO, minimum=0, maximum=100)
        self.operation = drv.Subsystem(self)
        self.operation.pid = drv.Subsystem(self.operation)
        self.operation.pid.proportional = Cmd("PB1")
        self.operation.pid.integral = Cmd("IT1", minimum=0.00, maximum=99.99)
        self.operation.pid.derivative = Cmd("DE1", minimum=0.00, maximum=9.99)

Protocol¶

We should now implement the communication protocol used to read and write commands to the hardware. In the present case, we implement the XON/XOFF protocol described in the manual. Write commands start with ? and end with a carriage return \r and read commands start with = and end with \r. Adapted from the manual:

“=” Command Example

• Master: = SP1 50\r (Set the setpoint prompt value SP1 to 50.)
• Remote: XOFF (byte \x13) (This will be returned once the device starts processing. The master must stay offline.)
• Remote: XON (byte \x11) (Processing is done. Do not send another message until this character is received.)

”?” Command Example

• Master: ? SP1\r (Request the SP1 prompt value.)
• Remote: XOFF (The remote is preparing the response. The master must stay offline.)
• Remote: XON 50\r (The value is returned and the master may send another message once the \r is received.)

that can be translated to the short UML sequence diagram:

class TestSeries982(unittest.TestCase):

    def setUp(self):
        socket = drv.TesterSocket()
        # socket.msg = {} such as: dict(send message: receive message)
        socket.msg = {"? PB1\r": "\x13\x1111\r",  # send lhs, receive rhs
                      "= DE1 3\r": "\x13\x11",
                      "? C1\r": "\x13\x1150\r",
                      "= SP1 25\r": "\x13\x11"}
        self.i = Series982(socket)

    def test_read(self):
        self.assertEqual(self.i.temperature1.read(), 50)

    def test_write(self):
        self.i.setpoint.write(25)

    def test_read_operation_pid(self):
        self.assertEqual(self.i.operation.pid.proportional.read(), 11)

    def test_write_operation_pid(self):
        self.i.operation.pid.derivative.write(3)

class XonXoffProtocol(drv.CommunicationProtocol):

    """Communication using the XON/XOFF protocol."""

    def _xonxoff(self):
        xonxoff = self._socket.read(2)
        if not xonxoff == "\x13\x11":
            raise drv.DriverError("Expected XON/XOFF (%r) got %r instead."
                                  % ("\x13\x11", xonxoff))

    def read(self, context):
        line = "? {mnemonic}\r".format(mnemonic=context.reader)
        self._socket.write(line)
        self._xonxoff()
        return float(self._socket.readline())  # convert unicode to float

    def write(self, context):
        line = "= {mnemonic} {value}\r".format(mnemonic=context.writer,
                                               value=context.value)
        self._socket.write(line)
        self._xonxoff()