Commit f687613b authored by Tammo Jan Dijkema's avatar Tammo Jan Dijkema
Browse files

Make thread safe, remove port searching logic for now

parent edb25e27
import hpib
from hpib import HpibDevice
import astropy.units as u
from astropy.units import Quantity
import threading
class CamrasHpibDevice(HpibDevice):
"""Wrapper around HPIB commands"""
import hpib
import serial.threaded
_CamrasHpibDevice__port = None
_CamrasHpibDevice__port_lock = threading.Lock()
serial_port = serial.Serial("/dev/ttyUSB0")
command_thread = serial.threaded.ReaderThread(serial_port, hpib.GPIBProtocol)
class CamrasHpibDevice(object):
"""Wrapper around HPIB commands"""
def __init__(self, address):
if not CamrasHpibDevice._CamrasHpibDevice__port:
with CamrasHpibDevice._CamrasHpibDevice__port_lock:
_port = hpib.PrologixGpibUsb()
super(CamrasHpibDevice, self).__init__(address, _port)
if not command_thread.is_alive():
self.hpib_address = address
def frequency(self):
"""Returns the frequency of the device as an astropy Quantity.
Throws a RuntimeError if no response"""
freq_str = self.query("freq?")
freq_str = command_thread.protocol.command("freq?", address=self.hpib_address)
if len(freq_str)==0:
raise RuntimeError("Camras device at address {} is not responding".format(self._hpib_address))
return int(float(freq_str)) * u.Hz
......@@ -30,7 +29,7 @@ class CamrasHpibDevice(HpibDevice):
"""Set the device to the specified frequency. Throws a RuntimeError if failed"""
if isinstance(freq, Quantity):
freq =
self.command("freq {:d} Hz".format(int(freq)))
command_thread.protocol.command("freq {:d} Hz".format(int(freq)), address=self.hpib_address)
new_freq = self.frequency
if != int(freq):
raise RuntimeError("Setting frequency failed: tried to set to {}, it is now {}".format(
......@@ -39,17 +38,17 @@ class CamrasHpibDevice(HpibDevice):
class Receiver(CamrasHpibDevice):
"""Wrapper around HPIB commands for the Rohde & Schwartz receiver"""
def __init__(self):
super(Receiver, self).__init__(1)
def __init__(self, **kwargs):
super(Receiver, self).__init__(1, **kwargs)
class LocalOscillator(CamrasHpibDevice):
"""Wrapper around HPIB commands for the local oscillator"""
def __init__(self):
super(LocalOscillator, self).__init__(28)
def __init__(self, **kwargs):
super(LocalOscillator, self).__init__(28, **kwargs)
class ClockGenerator(CamrasHpibDevice):
"""Wrapper around HPIB commands for the clock generator (should be at 140MHz)"""
def __init__(self):
super(SignalGenerator, self).__init__(14)
def __init__(self, **kwargs):
super(SignalGenerator, self).__init__(14, **kwargs)
import serial
import serial.threaded
import threading
import time
import re
import platform
import sys
from threading import Lock
class PortDevice(object):
def command(self, hpib_address, command):
raise NotImplementedError
def query(self, hpib_address, query, count=80):
raise NotImplementedError
def set_local(self, hpib_address):
raise NotImplementedError
class PrologixGpibUsb(PortDevice):
_serialPort = None
_devices = "Not scanned."
def __init__(self, port="", timeout=1, port_format="%d", max_device_num=9, verbose=False,
if port == "":
max_device_num, port, port_format = self._set_default_platform_values()
self._lock = Lock()
if verbose:
print("Looking for ports such as %s in range %d on %s" % \
(port, max_device_num, platform.system()))
self._find_available_port(max_device_num, port, port_format, timeout, verbose)
if find_devices:
def _set_default_platform_values(self):
if platform.system() == "Linux":
port = "/dev/ttyUSB0"
port_format = "%d"
max_device_num = 99
elif platform.system() == "Darwin":
port = "/dev/ttys000"
port_format = "%03d"
max_device_num = 999
raise NotImplementedError("Windows is not supported")
return max_device_num, port, port_format
def _validate_port(self, port):
self._rematch ="(/dev/tty.*[^0-9])([0-9]+$)", port)
if not self._rematch:
raise ValueError(port)
def _find_available_port(self, max_device_num, port, port_format, timeout, verbose):
while int( < max_device_num:
if verbose:
print("trying ", port)
self._serialPort = serial.Serial(port, timeout=timeout)
except serial.SerialException:
port = + (port_format % (int( + 1))
self._rematch ="(/dev/tty.*[^0-9])([0-9]+$)", port)
if self._serialPort is None:
raise serial.SerialException("Cannot find port")
def _write_serial(self, commandString):
if sys.version_info >= (3,0):
self._serialPort.write(bytes(commandString, 'utf-8'))
def _initialize_prologix_gpi(self):
with self._lock:
self._write_serial("++savecfg 0" + chr(10))
self._write_serial("++auto 0" + chr(10))
self._write_serial("++eoi 1" + chr(10))
self._write_serial("++eos 2" + chr(10))
self._write_serial("++eot_enable 0" + chr(10))
self._write_serial("++eot_char 0" + chr(10))
self._write_serial("++read_tmo_ms 500" + chr(10))
def _find_devices(self):
self._devices = []
for address in range(0, 31):
commandStr = "++addr " + str(address) + chr(10) + \
"*idn?" + chr(10) + \
"++read" + chr(10)
with self._lock:
returnStr = self._serialPort.readline()
if returnStr != "":
self._devices += [address, returnStr]
def command(self, hpib_address, command):
command = "++addr " + str(hpib_address) + chr(10) + command
with self._lock:
self._write_serial(command + chr(10))
def query(self, hpib_address, query, count=80):
query = "++addr " + str(hpib_address) + chr(10) + query
query += chr(10) + "++read eoi" + chr(10)
with self._lock:
line = self._serialPort.readline()
return line[:-1]
def set_local(self, hpib_address):
# self._commandStr = "++addr "+str(hpib_address)+chr(10)+"loc "+str(hpib_address)+chr(10)
# self._commandCount = self._write_serial(self._commandStr)
def get_devices(self):
return self._devices
class GPIB_232_485CT_A(PortDevice):
def __init__(self, port="/dev/ttyS0", baudrate=9600, bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, interCharTimeout=None,
pause=0, timeout=10):
self._serialPort = serial.Serial(port, baudrate, bytesize, parity, stopbits,
interCharTimeout, timeout)
self._pause = pause
def command(self, hpib_address, command):
self._commandStr = "wr " + str(hpib_address) + chr(10) + command + chr(13)
self._commandCount = self._write_serial(self._commandStr)
def query(self, hpib_address, query, count=80):
# flushInput should not be necessary
# self._serialPort.flushInput()
self._commandStr = "wr " + str(hpib_address) + chr(10) + query + chr(13)
self._commandCount = self._write_serial(self._commandStr)
self._queryStr = "rd #"+str(count)+" "+str(hpib_address)+chr(13)
self._queryCount = self._write_serial(self._queryStr)
self._returnStr =
self._returnCnt = int(self._serialPort.readline())
return self._returnStr[:self._returnCnt-1]
def set_local(self, hpib_address):
self._commandStr = "loc "+str(hpib_address)+chr(13)
self._commandCount = self._write_serial(self._commandStr)
class HpibDevice(object):
def __init__(self, hpib_address, port, keepLocal=False,pause=0):
self._hpib_address = hpib_address
self._port = port
self._keepLocal = keepLocal
self._pause = pause
def keep_remote(self, keepRemote=False):
self._keepLocal = not keepRemote
def set_local(self, set2Local=True):
if set2Local:
def command(self, commandStr):
self._port.command(self._hpib_address, commandStr)
def query(self, queryStr):
answerStr = self._port.query(self._hpib_address, queryStr)
return answerStr
def get_devices(self):
return ["Not implemented"]
import datetime
import queue
except ImportError:
import Queue as queue
class GPIBProtocol(serial.threaded.LineReader):
def __init__(self):
super(GPIBProtocol, self).__init__()
self.responses = queue.Queue()
self.lock = threading.Lock()
def init_hpib(self):
with self.lock:
self.write_line("++savecfg 0")
self.write_line("++auto 0")
self.write_line("++eoi 1")
self.write_line("++eos 2")
self.write_line("++eot_enable 0")
self.write_line("++eot_char 0")
self.write_line("++read_tmo_ms 500")
def command(self, command, timeout=0.2, address=None):
with self.lock:
if address:
self.write_line("++addr {:d}".format(address))
if "?" in command:
self.write_line("++read eoi")
# The command was a query, wait for response
line = self.responses.get(timeout=timeout)
return line
def handle_line(self, line):
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment