m-labs · Dec 19, 2016 · Dec 19, 2016 · Dec 19, 2016
diff --git a/artiq/devices/korad_ka3005p/__init__.py b/artiq/devices/korad_ka3005p/__init__.py
diff --git a/artiq/devices/korad_ka3005p/driver.py b/artiq/devices/korad_ka3005p/driver.py
@@ -0,0 +1,147 @@
+# Written by Joe Britton, 2016
+
+import logging
+import asyncio
+import asyncserial
+
+logger = logging.getLogger(__name__)
+
+
+class UnexpectedResponse(Exception):
+    pass
+
+
+class KoradKA3005P:
+    """The Korad KA3005P is a 1-channel programmable power supply
+    (0-30V/0-5A) with both USB/serial and RS232 connectivity.
+
+    All amplitudes are in volts.
+    All currents are in amperes.
+    """
+
+    # Serial interface gleaned from the following.
+    # https://github.com/starforgelabs/py-korad-serial
+    # https://sigrok.org/wiki/Korad_KAxxxxP_series
+
+    def __init__(self, serial_dev):
+        if serial_dev is None:
+            self.simulation = True
+        else:
+            self.simulation = False
+            self.port = asyncserial.AsyncSerial(serial_dev, baudrate=9600)
+
+    def close(self):
+        """Close the serial port."""
+        if not self.simulation:
+            self.port.close()
+
+    async def _ser_read(self, fixed_length=None):
+        """ strings returned by firmware are zero-terminated or fixed length
+        """
+        c = (await self.port.read(1)).decode()
+        r = c
+        while len(c) > 0 and ord(c) != 0 and not len(r) == fixed_length:
+            c = (await self.port.read(1)).decode()
+            r += c
+        logger.debug("_read %s: ", r)
+        return r
+
+    async def _ser_write(self, cmd):
+        logger.debug("_write %s: ", cmd)
+        await asyncio.sleep(0.1)
+        await self.port.write(cmd.encode('ascii'))
+
+    async def setup(self):
+        """Configure in known state."""
+        await self.set_output(False)
+        await self.set_v(0)
+        await self.set_ovp(False)
+        await self.set_i(0)
+        await self.set_ocp(False)
+
+    async def get_id(self):
+        """Request identification from device.
+        """
+        if self.simulation:
+            return "KORADKA3005PV2.0"
+        await self._ser_write("*IDN?")
+        return await self._ser_read()
+
+    async def set_output(self, b):
+        """Enable/disable the power output.
+        """
+        if b:
+            await self._ser_write("OUT1")
+        else:
+            await self._ser_write("OUT0")
+
+    async def set_v(self, v):
+        """Set the maximum output voltage."""
+        await self._ser_write("VSET1:{0:05.2f}".format(v))
+
+    async def get_v(self):
+        """Request the voltage as set by the user."""
+        await self._ser_write("VSET1?")
+        return float(await self._ser_read(fixed_length=5))
+
+    async def measure_v(self):
+        """Request the actual voltage output."""
+        await self._ser_write("VOUT1?")
+        return float(await self._ser_read(fixed_length=5))
+
+    async def set_ovp(self, b):
+        """Enable/disable the "Over Voltage Protection", the PS will switch off the
+        output when the voltage rises above the actual level."""
+        if b:
+            await self._ser_write("OVP1")
+        else:
+            await self._ser_write("OVP0")
+
+    async def set_i(self, v):
+        """Set the maximum output current."""
+        await self._ser_write("ISET1:{0:05.3f}".format(v))
+
+    async def get_i(self):
+        """Request the current as set by the user. """
+
+        # ISET1? replies with a sixth byte on many models (all?)
+        # which is the sixth character from *IDN?
+        # reply if *IDN? was queried before (during same power cycle).
+        # This byte is read and discarded.
+        await self._ser_write("ISET1?")
+        r = await self._ser_read(fixed_length=5)
+        if r[0] == "K":
+            r = r[1:-1]
+        return float(r)
+
+    async def measure_i(self):
+        """Request the actual output current."""
+        await self._ser_write("IOUT1?")
+        r = await self._ser_read(fixed_length=6)
+        if r[0] == "K":
+            r = r[1:-1]
+        return float(r)
+
+    async def set_ocp(self, b):
+        """Enable/disable the "Over Current Protection", the PS will switch off
+        the output when the current rises above the actual level."""
+        if b:
+            await self._ser_write("OCP1")
+        else:
+            await self._ser_write("OCP0")
+
+    async def ping(self):
+        """Check if device is responding."""
+        if self.simulation:
+            return True
+        try:
+            id = await self.get_id()
+        except asyncio.CancelledError:
+            raise
+        except:
+            return False
+        if id == "KORADKA3005PV2.0":
+            logger.debug("ping successful")
+            return True
+        else:
+            return False
diff --git a/artiq/frontend/korad_ka3005p_controller.py b/artiq/frontend/korad_ka3005p_controller.py
@@ -0,0 +1,54 @@
+#!/usr/bin/env python3.5
+
+# Written by Joe Britton, 2016
+
+import argparse
+import logging
+import sys
+import asyncio
+import os
+
+from artiq.devices.korad_ka3005p.driver import KoradKA3005P
+from artiq.protocols.pc_rpc import simple_server_loop
+from artiq.tools import *
+
+
+logger = logging.getLogger(__name__)
+
+
+def get_argparser():
+    parser = argparse.ArgumentParser(
+        description="ARTIQ controller for the Korad KA3005P programmable DC power supply")
+    simple_network_args(parser, 3256)
+    parser.add_argument(
+        "-d", "--device", default=None,
+        help="serial port.")
+    parser.add_argument(
+        "--simulation", action="store_true",
+        help="Put the driver in simulation mode, even if --device is used.")
+    verbosity_args(parser)
+    return parser
+
+
+def main():
+    args = get_argparser().parse_args()
+    init_logger(args)
+
+    if os.name == "nt":
+        asyncio.set_event_loop(asyncio.ProactorEventLoop())
+
+    if not args.simulation and args.device is None:
+        print("You need to specify either --simulation or -d/--device "
+              "argument. Use --help for more information.")
+        sys.exit(1)
+
+    dev = KoradKA3005P(args.device if not args.simulation else None)
+    asyncio.get_event_loop().run_until_complete(dev.setup())
+    try:
+        simple_server_loop(
+            {"korad_ka3005p": dev}, bind_address_from_args(args), args.port)
+    finally:
+        dev.close()
+
+if __name__ == "__main__":
+    main()
diff --git a/artiq/test/test_korad_ka3005p.py b/artiq/test/test_korad_ka3005p.py
@@ -0,0 +1,38 @@
+import sys
+import unittest
+
+from artiq.test.hardware_testbench import GenericControllerCase, ControllerCase
+
+
+class GenericKoradKA3005PTest:
+    def test_parameters_readback(self):
+
+        # check device ID baked into firmware
+        ids = self.driver.get_id()
+        self.assertEqual(ids, "KORADKA3005PV2.0")
+
+
+class TestKoradKA3005P(GenericKoradKA3005PTest, ControllerCase):
+    def setUp(self):
+        ControllerCase.setUp(self)
+        self.start_controller("koradka3005p")
+        self.driver = self.device_mgr.get("koradka3005p")
+
+
+class TestKoradKA3005P(GenericKoradKA3005PTest, GenericControllerCase):
+    def get_device_db(self):
+        return {
+            "korad_ka3005p": {
+                "type": "controller",
+                "host": "::1",
+                "port": 3256,
+                "command": (sys.executable.replace("\\", "\\\\")
+                            + " -m artiq.frontend.korad_ka3005p_controller "
+                            + "-p {port} --simulation")
+            }
+        }
+
+    def setUp(self):
+        GenericControllerCase.setUp(self)
+        self.start_controller("korad_ka3005p")
+        self.driver = self.device_mgr.get("korad_ka3005p")
diff --git a/doc/manual/default_network_ports.rst b/doc/manual/default_network_ports.rst
@@ -30,3 +30,6 @@ Default network ports
 +--------------------------+--------------+
 | Thorlabs T-Cube          | 3255         |
 +--------------------------+--------------+
+| Korad KA3005P            | 3256         |
+-------------------------------------------
+
diff --git a/doc/manual/ndsp_reference.rst b/doc/manual/ndsp_reference.rst
@@ -69,6 +69,22 @@ You can choose the LDA model with the ``-P`` parameter. The default is LDA-102.
    :ref: artiq.frontend.lda_controller.get_argparser
    :prog: lda_controller
 
+Korad KA3005P
+-------------
+
+Driver
+++++++
+
+.. automodule:: artiq.devices.korad_ka3005p.driver
+    :members:
+
+Controller
+++++++++++
+
+.. argparse::
+   :ref: artiq.frontend.korad_ka3005p_controller.get_argparser
+   :prog: korad_ka3005p_controller
+
 Novatech 409B
 -------------
 

diff --git a/setup.py b/setup.py
@@ -36,6 +36,7 @@
     "artiq_flash=artiq.frontend.artiq_flash:main",
     "lda_controller=artiq.frontend.lda_controller:main",
     "novatech409b_controller=artiq.frontend.novatech409b_controller:main",
+    "korad_ka3005p_controller=artiq.frontend.korad_ka3005p_controller:main",
     "pdq2_client=artiq.frontend.pdq2_client:main",
     "pdq2_controller=artiq.frontend.pdq2_controller:main",
     "thorlabs_tcube_controller=artiq.frontend.thorlabs_tcube_controller:main",
-Original file line number
+Diff line change
@@ @@ -30,3 +30,6 @@ Default network ports @@
     +--------------------------+--------------+
     | Thorlabs T-Cube          | 3255         |
     +--------------------------+--------------+
+    | Korad KA3005P            | 3256         |
+    -------------------------------------------