Commit 51db6349f42 by Julien Salort

- Fix bug in instruments/drivers.py where default value for…

- Fix bug in instruments/drivers.py where default value for `default_inter_params` should be a dict and not a set.
- Run black
parent 8935d67f424
......@@ -29,7 +29,7 @@ class Driver:
"""
default_physical_interface = None
default_inter_params = {}
default_inter_params = dict()
@classmethod
def _build_class_with_features(cls, features):
......
......@@ -13,7 +13,12 @@ from ipaddress import ip_address
from fluidlab.interfaces import PhysicalInterfaceType
from fluidlab.instruments.drivers import Driver
from fluidlab.instruments.features import QueryCommand, WriteCommand, FloatValue, StringValue
from fluidlab.instruments.features import (
QueryCommand,
WriteCommand,
FloatValue,
StringValue,
)
__all__ = ["NeelLHLM"]
......@@ -23,8 +28,8 @@ def out_port(ip):
Out port is 12000 + last number of ip address
"""
ip = ip_address(ip)
return 12000 + (int(ip) & 0xFF)
return 12000 + (int(ip) & 0xFF)
class NeelLHLM(Driver):
default_physical_interface = PhysicalInterfaceType.Ethernet
......@@ -34,60 +39,60 @@ class NeelLHLM(Driver):
"ethernet_protocol": "udp",
}
class PrefixValue(FloatValue):
class PrefixValue(FloatValue):
def __init__(self, *args, **kwargs):
self.prefix = kwargs.pop('prefix')
self.prefix = kwargs.pop("prefix")
self._fmt = self.prefix + ":{:.4f}"
super().__init__(*args, **kwargs)
def _convert_from_str(self, value):
data = value.split(':')
data = value.split(":")
if data[0] != self.prefix:
raise ValueError(f'Wrong value "{value:}"')
try:
return float(data[1])
except ValueError:
return 0.0
features = [
QueryCommand("query_identification", "Identification query", "*IDN?"),
WriteCommand("start_measurement", "Start a measurement", "MEAS"),
PrefixValue("level",
doc="Helium level in meters",
command_get="LEVEL?",
prefix="LEVEL"),
PrefixValue("volume",
doc="Helium volume in liters",
command_get="VOLUME?",
prefix="VOLUME"),
PrefixValue(
"level",
doc="Helium level in meters",
command_get="LEVEL?",
prefix="LEVEL",
),
PrefixValue(
"volume",
doc="Helium volume in liters",
command_get="VOLUME?",
prefix="VOLUME",
),
WriteCommand("reboot", "Restart device", "REBOOT"),
WriteCommand("automode", "Activate auto mode", "AUTOMODE"),
WriteCommand("manualmode", "Activate manual mode", "MANUALMODE"),
PrefixValue("ohm",
doc="Resistance in ohms",
command_get="RES?",
prefix="RES"),
StringValue("mode",
doc="Measurement mode",
command_get="MODE?"),
PrefixValue("vdc",
doc="Wire voltage",
command_get="VOLTAGE?",
prefix="VOLTAGE"),
StringValue("datasource",
doc="Source of probe parameters",
command_get="DATASOURCE?")
]
PrefixValue(
"ohm", doc="Resistance in ohms", command_get="RES?", prefix="RES"
),
StringValue("mode", doc="Measurement mode", command_get="MODE?"),
PrefixValue(
"vdc", doc="Wire voltage", command_get="VOLTAGE?", prefix="VOLTAGE"
),
StringValue(
"datasource", doc="Source of probe parameters", command_get="DATASOURCE?"
),
]
NeelLHLM._build_class_with_features(features)
if __name__ == '__main__':
with NeelLHLM('192.168.0.4') as lhlm:
if __name__ == "__main__":
with NeelLHLM("192.168.0.4") as lhlm:
identification = lhlm.query_identification()
print('ID:', identification)
print("ID:", identification)
level = lhlm.level.get()
volume = lhlm.volume.get()
print('Level =', level, 'm')
print('Volume =', volume, 'L')
print("Level =", level, "m")
print("Volume =", volume, "L")
......@@ -141,14 +141,18 @@ def interface_from_string(name, default_physical_interface=None, **kwargs):
dev = int(devnum)
return GPIBInterface(board, dev)
if classname == "SocketInterface":
from fluidlab.interfaces.socket_inter import TCPSocketInterface, UDPSocketInterface
protocol = kwargs.pop('ethernet_protocol', 'tcp')
if protocol == 'tcp':
from fluidlab.interfaces.socket_inter import (
TCPSocketInterface,
UDPSocketInterface,
)
protocol = kwargs.pop("ethernet_protocol", "tcp")
if protocol == "tcp":
return TCPSocketInterface(name, **kwargs)
elif protocol == 'udp':
elif protocol == "udp":
return UDPSocketInterface(name, **kwargs)
else:
raise ValueError('Wrong ethernet_protocol. Should be tcp or udp')
raise ValueError("Wrong ethernet_protocol. Should be tcp or udp")
if classname == "SerialInterface":
from fluidlab.interfaces.serial_inter import SerialInterface
......
......@@ -25,20 +25,20 @@ class SocketInterface(QueryInterface):
super().__init__()
self.ip_address = ip_address
self.autoremove_eol = autoremove_eol
def __str__(self):
return f'SocketInterface("{self.ip_address:}")'
def __repr__(self):
return str(self)
def _write(self, data):
if isinstance(data, str):
data = data.encode("ascii")
if self.autoremove_eol and not data.endswith(b"\r\n"):
data = data + b"\r\n"
self._send(data)
def _read(self, timeout=10.0):
data = self._recv(timeout)
if self.autoremove_eol and data.endswith("\r\n"):
......@@ -57,21 +57,22 @@ class UDPSocketInterface(SocketInterface):
self.out_port = out_port(ip_address)
else:
self.out_port = out_port
def _open(self):
self.in_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.out_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.in_sock.bind((socket.gethostbyname(socket.gethostname()),
self.in_port))
self.in_sock.bind(
(socket.gethostbyname(socket.gethostname()), self.in_port)
)
self.out_sock.connect((self.ip_address, self.out_port))
def _close(self):
self.out_sock.close()
self.in_sock.close()
def _send(self, data):
self.out_sock.sendall(data)
def _recv(self, timeout):
self.in_sock.settimeout(timeout)
chunks = []
......@@ -87,8 +88,8 @@ class UDPSocketInterface(SocketInterface):
break
data = b"".join(chunks).decode("ascii")
return data
class TCPSocketInterface(SocketInterface):
def __init__(self, ip_address, port, autoremove_eol=True):
super().__init__(ip_address, autoremove_eol)
......
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