Commit 00377566b5a by Julien Salort

Run black

parent fc120d65e92
......@@ -24,7 +24,7 @@ def del_and_load_exp(exp, **kargs):
path_save = exp.path_save
# str_path = os.path.split(path_save)[-1]
str_path = path_save
del (exp)
del exp
# load the experiment
exp = fluidlab.load_exp(str_path, **kargs)
# clean by removing the directory
......
......@@ -19,17 +19,19 @@ from time import sleep
class Julabo(Driver):
default_physical_interface = PhysicalInterfaceType.Serial
default_inter_params = {'baudrate': 9600,
'bytesize': 8,
'parity': "N",
'stopbits': 1,
'timeout': 5.0,
'xonxoff': True,
'rtscts': False,
'dsrdtr': False,
'eol': '\r\n',
'autoremove_eol': True}
default_inter_params = {
"baudrate": 9600,
"bytesize": 8,
"parity": "N",
"stopbits": 1,
"timeout": 5.0,
"xonxoff": True,
"rtscts": False,
"dsrdtr": False,
"eol": "\r\n",
"autoremove_eol": True,
}
def __enter__(self):
super(Julabo, self).__enter__()
identification = self.interface.query("version")
......
......@@ -130,14 +130,16 @@ class Lauda(Driver):
# untested model without knowning.
Models = {"RP 845": 845, "RP 855": 855, "E200": 200, "VC": 1200}
default_physical_interface = PhysicalInterfaceType.Serial
default_inter_params = {'baudrate': 9600,
'bytesize': 8,
'parity': "N",
'stopbits': 1,
'timeout': 1,
'xonxoff': False,
'rtscts': False,
'dsrdtr': False}
default_inter_params = {
"baudrate": 9600,
"bytesize": 8,
"parity": "N",
"stopbits": 1,
"timeout": 1,
"xonxoff": False,
"rtscts": False,
"dsrdtr": False,
}
def __enter__(self):
super(Lauda, self).__enter__()
......
......@@ -131,14 +131,16 @@ class ThermocubeValue(Value):
class Thermocube(Driver):
default_physical_interface = PhysicalInterfaceType.Serial
default_inter_params = {'baudrate': 9600,
'bytesize': 8,
'parity': "N",
'stopbits': 1,
'timeout': 2,
'xonxoff': False,
'rtscts': False,
'dsrdtr': False}
default_inter_params = {
"baudrate": 9600,
"bytesize": 8,
"parity": "N",
"stopbits": 1,
"timeout": 2,
"xonxoff": False,
"rtscts": False,
"dsrdtr": False,
}
features = [
......
......@@ -34,8 +34,9 @@ class Driver:
The interface used to communicate with the instrument.
"""
default_physical_interface = None
default_inter_params = {'port': 0}
default_inter_params = {"port": 0}
@classmethod
def _build_class_with_features(cls, features):
......@@ -44,9 +45,11 @@ class Driver:
def __init__(self, interface=None):
if isinstance(interface, str):
interface = interface_from_string(interface,
self.default_physical_interface,
**self.default_inter_params)
interface = interface_from_string(
interface,
self.default_physical_interface,
**self.default_inter_params
)
elif not interface:
interface = FalseInterface()
elif not isinstance(interface, Interface):
......@@ -54,7 +57,7 @@ class Driver:
self._interface = self.interface = interface
self.values = {}
for name in dir(self):
v = getattr(self, name)
if isinstance(v, SuperValue):
......@@ -107,5 +110,6 @@ class Driver:
def __exit__(self, type_, value, cb):
self.interface.__exit__(type_, value, cb)
if __name__ == "__main__":
driver = Driver("ASRL4::INSTR", backend="@sim")
......@@ -192,19 +192,19 @@ class Value(SuperValue):
if self.channel_argument:
# here we don't call _convert_as_str to allow the user to choose
# the desired format in the command_set string
command = self.command_set.format(
channel=channel, value=value
)
command = self.command_set.format(channel=channel, value=value)
else:
command = self.command_set + " " + self._convert_as_str(value)
self._interface.write(command)
if self.check_instrument_value_after_set:
self._check_instrument_value(value)
class ReadOnlyBoolValue(Value):
def get(self):
return self._interface.read_readonlybool(self._adress)
class BoolValue(Value):
def __init__(
self,
......@@ -305,6 +305,7 @@ class StringValue(Value):
)
warnings.warn(msg, UserWarning)
class ReadOnlyInt16Value(Value):
def get(self):
return self._interface.read_readonlyint16(self._adress)
......
......@@ -21,5 +21,5 @@ __all__ = [
"hp_33120a",
"tektronix_afg3022b",
"tti_tsx3510p",
"srs_ds360"
"srs_ds360",
]
......@@ -20,8 +20,13 @@ from fluidlab.instruments.iec60488 import (
)
from fluidlab.instruments.features import FloatScientificValue, FloatValue, BoolValue
from fluidlab.instruments.features import (
FloatScientificValue,
FloatValue,
BoolValue,
)
class StanfordDS360(
IEC60488, PowerOn, Calibration, Trigger, ObjectIdentification, StoredSetting
):
......@@ -32,31 +37,36 @@ class StanfordDS360(
"""
features = [
FloatScientificValue('vrms',
doc='RMS voltage of generated wave (can be zero)',
channel_argument=True,
command_set='AMPL {value:.2e} VR',
command_get='AMPL?VR'),
FloatScientificValue('vdc',
doc='Offset voltage of generated wave',
channel_argument=True,
command_set='OFFS {value:.2e}',
command_get='OFFS?'),
FloatValue('freq',
doc='Frequency of generated wave',
command_set='FREQ',
command_get='FREQ?'),
BoolValue('onoff',
doc='Output ON/OFF',
command_set='OUTE',
command_get='OUTE?'),
BoolValue('balanced',
doc='Output mode (balanced or unbalanced)',
command_set='OUTM',
command_get='OUTM?')
FloatScientificValue(
"vrms",
doc="RMS voltage of generated wave (can be zero)",
channel_argument=True,
command_set="AMPL {value:.2e} VR",
command_get="AMPL?VR",
),
FloatScientificValue(
"vdc",
doc="Offset voltage of generated wave",
channel_argument=True,
command_set="OFFS {value:.2e}",
command_get="OFFS?",
),
FloatValue(
"freq",
doc="Frequency of generated wave",
command_set="FREQ",
command_get="FREQ?",
),
BoolValue(
"onoff", doc="Output ON/OFF", command_set="OUTE", command_get="OUTE?"
),
BoolValue(
"balanced",
doc="Output mode (balanced or unbalanced)",
command_set="OUTM",
command_get="OUTM?",
),
]
StanfordDS360._build_class_with_features(features)
......@@ -132,8 +132,9 @@ class IEC60488(Driver):
Defines the backend used by pyvisa ("@py", "@ni", "@sim"...)
"""
default_physical_interface = PhysicalInterfaceType.GPIB
def __enter__(self):
super(IEC60488, self).__enter__()
identification = self.query_identification()
......@@ -142,9 +143,9 @@ class IEC60488(Driver):
identification = identification.strip()
print(f"Initialization driver for device: {identification}")
return self
def query_event_status_register(self):
number = self.query_esr()
return self.status_enable_register.compute_dict_from_number(number)
......
......@@ -117,10 +117,7 @@ import warnings
import numpy as np
from fluidlab.instruments.features import (
DecimalInt16Value,
Int16StringValue,
)
from fluidlab.instruments.features import DecimalInt16Value, Int16StringValue
from fluidlab.interfaces import PhysicalInterfaceType
from fluiddyn.util.terminal_colors import print_fail, print_warning
......@@ -155,8 +152,9 @@ class BaseUnidriveSP(Driver):
Unidrive SP.
"""
default_physical_interface = PhysicalInterfaceType.Modbus
_constant_nb_pairs_poles = 4
def __init__(
......
......@@ -13,5 +13,10 @@ Provides:
cryocon_24c
"""
__all__ = ["agilent_34970a", "keithley_2700", "lakeshore_224", "keithley_705",
"cryocon_24c"]
__all__ = [
"agilent_34970a",
"keithley_2700",
"lakeshore_224",
"keithley_705",
"cryocon_24c",
]
......@@ -10,23 +10,24 @@ from fluidlab.interfaces import PhysicalInterfaceType
from fluidlab.instruments.iec60488 import IEC60488
from fluidlab.instruments.features import FloatValue
class Cryocon24c(IEC60488):
default_physical_interface = PhysicalInterfaceType.Ethernet
default_inter_params = {'port': 5000}
default_inter_params = {"port": 5000}
features = [
FloatValue("temperature",
FloatValue(
"temperature",
doc="Reads temperature",
command_get="input? {channel:}",
channel_argument=True,
),
)
]
Cryocon24c._build_class_with_features(features)
if __name__ == '__main__':
with Cryocon24c('192.168.0.2') as cc:
Ta = cc.temperature.get('A')
print('Ta =', Ta)
\ No newline at end of file
if __name__ == "__main__":
with Cryocon24c("192.168.0.2") as cc:
Ta = cc.temperature.get("A")
print("Ta =", Ta)
......@@ -68,14 +68,17 @@ class FurnessValue(Value):
class FurnessFCO318(Driver):
default_physical_interface = PhysicalInterfaceType.Serial
default_inter_params = {'baudrate': 2400,
'bytesize': 8,
'parity': "N",
'stopbits': 1,
'timeout': 1.0,
'xonxoff': False,
'rtscts': False,
'dsrdtr': False}
default_inter_params = {
"baudrate": 2400,
"bytesize": 8,
"parity": "N",
"stopbits": 1,
"timeout": 1.0,
"xonxoff": False,
"rtscts": False,
"dsrdtr": False,
}
features = [
FurnessValue(
......
......@@ -106,22 +106,24 @@ class PfeifferMaxiGaugeOnOffValue(PfeifferMaxiGaugeValue):
class PfeifferMaxiGauge(Driver):
default_physical_interface = PhysicalInterfaceType.Serial
default_inter_params = {'baudrate': 9600,
'bytesize': 8,
'parity': "N",
'stopbits': 1,
'timeout': 1.0,
'xonxoff': False,
'rtscts': False,
'dsrdtr': False}
default_inter_params = {
"baudrate": 9600,
"bytesize": 8,
"parity": "N",
"stopbits": 1,
"timeout": 1.0,
"xonxoff": False,
"rtscts": False,
"dsrdtr": False,
}
def __init__(self, serialPort, debug=False):
super().__init__(serialPort)
self.debug = debug
def __enter__(self):
super(PfeifferMaxiGauge, self).__enter__()
self.clear_interface()
print("Pfeiffer MaxiGauge:", self.program_version().decode("ascii"))
long_description = {
......
......@@ -34,24 +34,29 @@ from enum import IntEnum
import sys
import ipaddress
class PhysicalInterfaceType(IntEnum):
GPIB = 0
Ethernet = 1
Serial = 2
Modbus = 3
DefaultInterface = {PhysicalInterfaceType.GPIB: 'VISAInterface',
PhysicalInterfaceType.Ethernet: 'SocketInterface',
PhysicalInterfaceType.Serial: 'SerialInterface',
PhysicalInterfaceType.Modbus: 'MinimalModbusInterface',
}
if sys.platform.startswith('linux'):
DefaultInterface[PhysicalInterfaceType.GPIB] = 'GPIBInterface'
DefaultInterface = {
PhysicalInterfaceType.GPIB: "VISAInterface",
PhysicalInterfaceType.Ethernet: "SocketInterface",
PhysicalInterfaceType.Serial: "SerialInterface",
PhysicalInterfaceType.Modbus: "MinimalModbusInterface",
}
if sys.platform.startswith("linux"):
DefaultInterface[PhysicalInterfaceType.GPIB] = "GPIBInterface"
def set_default_interface(interfaceType, interfaceClassname):
DefaultInterface[interfaceType] = interfaceClassname
def interface_from_string(name, defaultPhysicalInterface=None, **kwargs):
"""
Infer physicalInterface from name if possible, or use the default provided
......@@ -64,79 +69,97 @@ def interface_from_string(name, defaultPhysicalInterface=None, **kwargs):
"""
classname = None
physicalInterface = None
if 'GPIB' in name:
if "GPIB" in name:
physicalInterface = PhysicalInterfaceType.GPIB
elif 'ASRL' in name:
elif "ASRL" in name:
physicalInterface = PhysicalInterfaceType.Serial
classname = 'VISAInterface'
elif isinstance(name, ipaddress.IPv4Address) or isinstance(name, ipaddress.IPv6Address):
classname = "VISAInterface"
elif isinstance(name, ipaddress.IPv4Address) or isinstance(
name, ipaddress.IPv6Address
):
name = str(name)
physicalInterface = PhysicalInterfaceType.Ethernet
else:
try:
ip = ipaddress.ip_address(name)
physicalInterface = PhysicalInterfaceType.Ethernet if defaultPhysicalInterface is not PhysicalInterfaceType.Modbus else PhysicalInterfaceType.Modbus
physicalInterface = (
PhysicalInterfaceType.Ethernet
if defaultPhysicalInterface is not PhysicalInterfaceType.Modbus
else PhysicalInterfaceType.Modbus
)
except ValueError:
pass
if physicalInterface is None:
physicalInterface = defaultPhysicalInterface
if classname is None:
classname = DefaultInterface[physicalInterface]
if classname == 'VISAInterface':
if classname == "VISAInterface":
from fluidlab.interfaces.visa_inter import VISAInterface
return VISAInterface(name)
if classname == 'GPIBInterface':
if classname == "GPIBInterface":
from fluidlab.interfaces.gpib_inter import GPIBInterface
if name.endswith('::INSTR'):
if name.endswith("::INSTR"):
name = name[:-7]
boardname, devnum = name.split('::')
if boardname.startswith('GPIB'):
boardname, devnum = name.split("::")
if boardname.startswith("GPIB"):
boardname = boardname[4:]
board = int(boardname)
dev = int(devnum)
return GPIBInterface(board, dev)
if classname == 'SocketInterface':
if classname == "SocketInterface":
from fluidlab.interfaces.socket_inter import SocketInterface
return SocketInterface(name, **kwargs)
if classname == 'SerialInterface':
if classname == "SerialInterface":
from fluidlab.interfaces.serial_inter import SerialInterface
return SerialInterface(name, **kwargs)
if classname == 'MinimalModbusInterface':
if classname == "MinimalModbusInterface":
from fluidlab.interfaces.modbus_inter import MinimalModbusInterface
return MinimalModbusInterface(name, **kwargs)
if classname == 'PyModbusInterface':
if classname == "PyModbusInterface":
from fluidlab.interfaces.modbus_inter import PyModbusInterface
return PyModbusInterface(name, **kwargs)
raise ValueError('Unknown interface type')
raise ValueError("Unknown interface type")
class InterfaceWarning(Warning):
pass
class Interface:
def _open(self):
# do the actual open (without testing self.opened)
raise NotImplementedError
def _close(self):
# do the actual close (without testing self.opened)
raise NotImplementedError
def __init__(self):
self.opened = False
def open(self):
if not self.opened:
self._open()
self.opened = True
else:
warnings.warn('open() called on already opened interface.', InterfaceWarning)
warnings.warn(
"open() called on already opened interface.", InterfaceWarning
)
def close(self):
if self.opened:
self._close()
self.opened = False
else:
warnings.warn('close() called on already closed interface.', InterfaceWarning)
warnings.warn(
"close() called on already closed interface.", InterfaceWarning
)
def __enter__(self):
self.open()
......@@ -144,9 +167,9 @@ class Interface:
def __exit__(self, type_, value, cb):
self.close()
class QueryInterface(Interface):
class QueryInterface(Interface):
def _write(self, *args, **kwargs):
# do the actual write
raise NotImplementedError
......@@ -154,23 +177,29 @@ class QueryInterface(Interface):
def _read(self, *args, **kwargs):
# do the actual read
raise NotImplementedError
def write(self, *args, **kwargs):
if not self.opened:
warnings.warn('write() called on non-opened interface.', InterfaceWarning)
warnings.warn(
"write() called on non-opened interface.", InterfaceWarning
)
self.open()
self._write(*args, **kwargs)
def read(self, *args, **kwargs):
if not self.opened:
warnings.warn('read() called on non-opened interface.', InterfaceWarning)
warnings.warn(
"read() called on non-opened interface.", InterfaceWarning
)
self.open()
return self._read(*args, **kwargs)
def query(self, command, time_delay=0.1, **kwargs):
if hasattr(self, '_query'):
if hasattr(self, "_query"):
if not self.opened:
warnings.warn('query() called on non-opened interface.', InterfaceWarning)
warnings.warn(
"query() called on non-opened interface.", InterfaceWarning
)
self.open()
return self._query(command, **kwargs)
else:
......@@ -178,20 +207,21 @@ class QueryInterface(Interface):
sleep(time_delay)
return self.read(**kwargs)
class FalseInterface(QueryInterface):
"""
Dummy interface
"""
def _open(self):
pass
def _close(self):
pass
def _write(self, s):
print(s)
def _read(self):
print("just return 0 since it is a false Interface class.")
return 0
\ No newline at end of file
return 0
......@@ -49,20 +49,20 @@ class GPIBInterface(QueryInterface):
self.board_adress = board_adress
self.instrument_adress = instrument_adress
self.default_tmo = closest_timeout(timeout)
def __str__(self):
return f'GPIBInterface({self.board_adress:d}, {self.instrument_adress:d})'
return f"GPIBInterface({self.board_adress:d}, {self.instrument_adress:d})"
def __repr__(self):
return str(self)
def _open(self):
self.handle = gpib.dev(self.board_adress, self.instrument_adress)
gpib.timeout(self.handle, self.default_tmo)
def _close(self):
gpib.close(self.handle)
def _read(self, numbytes=None, verbose=False, tracing=False):
if tracing:
sys.stdout.write("* <- " + str(self.instrument_adress) + " ")
......@@ -80,7 +80,7 @@ class GPIBInterface(QueryInterface):
chunk_size = 150
chunk = gpib.read(self.handle, chunk_size)
data = data + chunk.decode("ascii")
if verbose:
sys.stdout.write(
"{:d} bytes read \r".format(len(data))
......@@ -123,7 +123,7 @@ class GPIBInterface(QueryInterface):
if (sta & gpib.TIMO) != 0:
# Timed out
if time.monotonic() - start > timeout:
print('Timeout occured')
print("Timeout occured")
break
else:
# SRQ asserted
......
......@@ -25,13 +25,12 @@ from fluidlab.interfaces import Interface
class ModbusInterface(Interface):
def __init__(self, port, method="rtu", slave_address=1, timeout=1):
def __init__(self, port, method="rtu", slave_address=1, timeout=1):
self.port = port
self.method = method
self.slave_address = slave_address
self.timeout = timeout
def read_readonlybool(self, addresses):
raise NotImplementedError
......@@ -58,22 +57,25 @@ class ModbusInterface(Interface):
def write_float32(self, addresses, values):
raise NotImplementedError
def __str__(self):
return f'ModbusInterface({self.port:}, {self.method:}, {self.slave_address:}, {self.timeout})'
return f"ModbusInterface({self.port:}, {self.method:}, {self.slave_address:}, {self.timeout})"
def __repr__(self):
return str(self)
class MinimalModbusInterface(ModbusInterface):