M-Labs
/
artiq
mirror of https://github.com/m-labs/artiq.git
2
0
Fork 2
Code Issues Projects Releases Wiki Activity

devices/novatech409b: convert to asyncserial

This commit is contained in:
Sebastien Bourdeauducq 2016-03-22 21:55:58 +08:00
parent 7657b67ea6
commit b5441fd107
2 changed files with 57 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
artiq/devices/novatech409b/driver.py
View File

@ -1,10 +1,10 @@
# Written by Joe Britton, 2015 # Written by Joe Britton, 2015
import time
import math import math
import logging import logging
import asyncio
import serial import asyncserial
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -41,22 +41,28 @@ class Novatech409B:
self.simulation = True self.simulation = True
else: else:
self.simulation = False self.simulation = False
self.port = serial.serial_for_url( self.port = asyncserial.AsyncSerial(
serial_dev, serial_dev,
baudrate=19200, baudrate=19200,
bytesize=8, bytesize=8,
parity="N", parity="N",
stopbits=1, stopbits=1,
xonxoff=0, xonxoff=0)
timeout=1.0)
self.setup()
def close(self): def close(self):
"""Close the serial port.""" """Close the serial port."""
if not self.simulation: if not self.simulation:
self.port.close() self.port.close()
def _ser_send(self, cmd, get_response=True): async def _ser_readline(self):
c = await self.port.read(1)
r = c
while c != b"\n":
c = await self.port.read(1)
r += c
return r
async def _ser_send(self, cmd, get_response=True):
"""Send a string to the serial port.""" """Send a string to the serial port."""
# Low-level routine for sending serial commands to device. It sends # Low-level routine for sending serial commands to device. It sends
@ -67,48 +73,42 @@ class Novatech409B:
if self.simulation: if self.simulation:
logger.info("simulation _ser_send(\"%s\")", cmd) logger.info("simulation _ser_send(\"%s\")", cmd)
else: else:
self.port.flushInput() logger.debug("_ser_send(\"%s\")", cmd)
self.port.write((cmd + "\r\n").encode()) self.port.ser.reset_input_buffer()
result = self.port.readline().rstrip().decode() await self.port.write((cmd + "\r\n").encode())
if get_response: if get_response:
result = (await self._ser_readline()).rstrip().decode()
logger.debug("got response from device: %s", result) logger.debug("got response from device: %s", result)
if result == "OK": if result != "OK":
pass errstr = self.error_codes.get(result, "Unrecognized reply")
elif result == "": s = "Erroneous reply from device: {ec}, {ecs}".format(
raise UnexpectedResponse("Response from device timed out")
else:
try:
errstr = self.error_codes[result]
except KeyError:
errstr = "Unrecognized reply: '{}'".format(result)
s = "Error Code = {ec}, {ecs}".format(
ec=result, ecs=errstr) ec=result, ecs=errstr)
raise UnexpectedResponse(s) raise ValueError(s)
else: else:
pass pass
def reset(self): async def reset(self):
"""Hardware reset of 409B.""" """Hardware reset of 409B."""
self._ser_send("R", get_response=False) await self._ser_send("R", get_response=False)
time.sleep(1) await asyncio.sleep(1)
self.setup() await self.setup()
def setup(self): async def setup(self):
"""Initial setup of 409B.""" """Initial setup of 409B."""
# Setup the Novatech 409B with the following defaults: # Setup the Novatech 409B with the following defaults:
# * command echo off ("E d") # * command echo off ("E d")
# * external clock ("") 10 MHz sinusoid -1 to +7 dBm # * external clock ("") 10 MHz sinusoid -1 to +7 dBm
self._ser_send("E d", get_response=False) await self._ser_send("E d", get_response=False)
self.set_phase_continuous(True) await self.set_phase_continuous(True)
self.set_simultaneous_update(False) await self.set_simultaneous_update(False)
def save_state_to_eeprom(self): async def save_state_to_eeprom(self):
"""Save current state to EEPROM.""" """Save current state to EEPROM."""
self._ser_send("S") await self._ser_send("S")
def set_phase_continuous(self, is_continuous): async def set_phase_continuous(self, is_continuous):
"""Toggle phase continuous mode. """Toggle phase continuous mode.
Sends the "M n" command. This turns off the automatic Sends the "M n" command. This turns off the automatic
@ -120,11 +120,11 @@ class Novatech409B:
:param is_continuous: True or False :param is_continuous: True or False
""" """
if is_continuous: if is_continuous:
self._ser_send("M n") await self._ser_send("M n")
else: else:
self._ser_send("M a") await self._ser_send("M a")
def set_simultaneous_update(self, simultaneous): async def set_simultaneous_update(self, simultaneous):
"""Set simultaneous update mode. """Set simultaneous update mode.
Sends the "I m" command. In this mode an update Sends the "I m" command. In this mode an update
@ -134,29 +134,29 @@ class Novatech409B:
simultaneously. simultaneously.
""" """
if simultaneous: if simultaneous:
self._ser_send("I m") await self._ser_send("I m")
else: else:
self._ser_send("I a") await self._ser_send("I a")
def do_simultaneous_update(self): async def do_simultaneous_update(self):
"""Apply update in simultaneous update mode.""" """Apply update in simultaneous update mode."""
self._ser_send("I p") await self._ser_send("I p")
def set_freq(self, ch_no, freq): async def set_freq(self, ch_no, freq):
"""Set frequency of one channel.""" """Set frequency of one channel."""
# Novatech expects MHz # Novatech expects MHz
self._ser_send("F{:d} {:f}".format(ch_no, freq/1e6)) await self._ser_send("F{:d} {:f}".format(ch_no, freq/1e6))
def set_phase(self, ch_no, phase): async def set_phase(self, ch_no, phase):
"""Set phase of one channel.""" """Set phase of one channel."""
# phase word is required by device # phase word is required by device
# N is an integer from 0 to 16383. Phase is set to # N is an integer from 0 to 16383. Phase is set to
# N*360/16384 deg; in ARTIQ represent phase in cycles [0, 1] # N*360/16384 deg; in ARTIQ represent phase in cycles [0, 1]
phase_word = round(phase*16383) phase_word = round(phase*16383)
cmd = "P{:d} {:d}".format(ch_no, phase_word) cmd = "P{:d} {:d}".format(ch_no, phase_word)
self._ser_send(cmd) await self._ser_send(cmd)
def set_gain(self, ch_no, volts): async def set_gain(self, ch_no, volts):
"""Set amplitude of one channel.""" """Set amplitude of one channel."""
# due to error in Novatech it doesn't generate an error for # due to error in Novatech it doesn't generate an error for
@ -167,9 +167,9 @@ class Novatech409B:
raise ValueError(s) raise ValueError(s)
s = "V{:d} {:d}".format(ch_no, dac_value) s = "V{:d} {:d}".format(ch_no, dac_value)
self._ser_send(s) await self._ser_send(s)
def get_status(self): async def get_status(self):
if self.simulation: if self.simulation:
return ["00989680 2000 01F5 0000 00000000 00000000 000301", return ["00989680 2000 01F5 0000 00000000 00000000 000301",
"00989680 2000 01F5 0000 00000000 00000000 000301", "00989680 2000 01F5 0000 00000000 00000000 000301",
@ -177,17 +177,19 @@ class Novatech409B:
"00989680 2000 01F5 0000 00000000 00000000 000301", "00989680 2000 01F5 0000 00000000 00000000 000301",
"80 BC0000 0000 0102 21"] "80 BC0000 0000 0102 21"]
else: else:
# status message is multi-line self.port.reset_input_buffer()
self.port.flushInput() await self.port.write(("QUE" + "\r\n").encode())
self.port.write(("QUE" + "\r\n").encode()) for i in range(5):
result = self.port.readlines() m = (await self._ser_readline()).rstrip().decode()
result = [r.rstrip().decode() for r in result] result.append(m)
logger.debug("got device status: %s", result) logger.debug("got device status: %s", result)
return result return result
def ping(self): async def ping(self):
try: try:
stat = self.get_status() stat = await self.get_status()
except asyncio.CancelledError:
raise
except: except:
return False return False
# check that version number matches is "21" # check that version number matches is "21"

2
artiq/frontend/novatech409b_controller.py
View File

@ -5,6 +5,7 @@
import argparse import argparse
import logging import logging
import sys import sys
import asyncio
from artiq.devices.novatech409b.driver import Novatech409B from artiq.devices.novatech409b.driver import Novatech409B
from artiq.protocols.pc_rpc import simple_server_loop from artiq.protocols.pc_rpc import simple_server_loop
@ -38,6 +39,7 @@ def main():
sys.exit(1) sys.exit(1)
dev = Novatech409B(args.device if not args.simulation else None) dev = Novatech409B(args.device if not args.simulation else None)
asyncio.get_event_loop().run_until_complete(dev.setup())
try: try:
simple_server_loop( simple_server_loop(
{"novatech409b": dev}, bind_address_from_args(args), args.port) {"novatech409b": dev}, bind_address_from_args(args), args.port)