Merge pull request #4 from AlexanderBraml/memristor_dev_fixes

Multiple Improvements to Control Lib

py_instrument_control_lib/channels/Channel.py

@@ -2,17 +2,21 @@
 from typing import Protocol, Any
 
 from py_instrument_control_lib.channels.ChannelEnums import ChannelUnit, ChannelIndex
+from py_instrument_control_lib.channels.FutureValue import FutureValue
 
 
 class ParentDevice(Protocol):
 
     def toggle_buffering(self, enable: bool) -> None:
         pass
 
-    def execute_buffered_script(self, blocking: bool, check_errors: bool) -> None:
+    def execute_buffered_script(self, blocking: bool, check_errors: bool = False) -> None:
         pass
 
-    def read_buffer(self, check_errors: bool) -> list:
+    def read_buffer(self, check_errors: bool = False) -> list:
+        pass
+
+    def get_buffer(self, check_errors: bool = False) -> list:
         pass
 
     def next_buffer_element(self, channel_idx: ChannelIndex) -> Any:
@@ -36,25 +40,34 @@ def measure_channel(self, unit: ChannelUnit, channel_number: ChannelIndex, check
 
 class Channel(ABC):
 
-    def __init__(self, device: ParentDevice, channel_idx: ChannelIndex, supported_source_units: list[ChannelUnit],
-                 supported_measure_units: list[ChannelUnit], buffering_available: bool = False):
+    def __init__(self, device: ParentDevice, channel_idx: ChannelIndex, buffering_available: bool = False):
         self._device = device
         self._channel_idx = channel_idx
-        self._supported_source_units = supported_source_units
-        self._supported_measure_units = supported_measure_units
         self._buffering_available = buffering_available
+        self._buffering_enabled = False
 
     def toggle_buffering(self, enable: bool) -> None:
         self.__check_buffering_available()
+        self._buffering_enabled = enable
         self._device.toggle_buffering(enable)
 
     def execute_buffered_script(self, blocking: bool, check_errors: bool = False) -> None:
         self.__check_buffering_available()
-        self._device.execute_buffered_script(blocking, check_errors)
+        self._device.execute_buffered_script(blocking=blocking, check_errors=check_errors)
 
     def read_buffer(self, check_errors: bool = False) -> list:
         self.__check_buffering_available()
-        return self._device.read_buffer(check_errors)
+        return self._device.read_buffer(check_errors=check_errors)
+
+    def read_values(self, blocking: bool, *value_lists: list[Any]) -> None:
+        if self._buffering_enabled:
+            if self._device.get_buffer() is None:
+                self.execute_buffered_script(blocking)
+                self.read_buffer()
+            for value_list in value_lists:
+                for index, value in enumerate(value_list):
+                    if isinstance(value, FutureValue):
+                        value_list[index] = value.get(self._device.get_buffer())
 
     def next_buffer_element(self) -> Any:
         self.__check_buffering_available()
@@ -70,11 +83,12 @@ class SourceChannel(Channel):
 
     def __init__(self, device: ParentDevice, channel_idx: ChannelIndex, supported_source_units: list[ChannelUnit],
                  buffering_available: bool = False):
-        super().__init__(device, channel_idx, supported_source_units, [], buffering_available)
+        Channel.__init__(self, device, channel_idx, buffering_available)
+        self._supported_source_units = supported_source_units
 
     def set_level(self, unit: ChannelUnit, value: float, check_errors: bool = False) -> None:
         if unit not in self._supported_source_units:
-            raise ValueError(f"Unit {unit} is not supported by this channel.")
+            raise ValueError(f"Setting unit {unit} is not supported by this channel.")
         self._device.set_channel_level(unit, self._channel_idx, value, check_errors)
 
     def toggle(self, enabled: bool, check_errors: bool = False) -> None:
@@ -86,20 +100,28 @@ class MeasureChannel(Channel):
 
     def __init__(self, device: ParentDevice, channel_idx: ChannelIndex, supported_measure_units: list[ChannelUnit],
                  buffering_available: bool = False):
-        super().__init__(device, channel_idx, [], supported_measure_units, buffering_available)
+        Channel.__init__(self, device, channel_idx, buffering_available)
+        self._supported_measure_units = supported_measure_units
+        self._buffer_index: dict = {}
 
     def measure(self, unit: ChannelUnit, check_errors: bool = False) -> float:
-        if unit not in self._supported_source_units:
-            raise ValueError(f"Unit {unit} is not supported by this channel.")
-        return self._device.measure_channel(unit, self._channel_idx, check_errors)
+        if unit not in self._supported_measure_units:
+            raise ValueError(f"Measuring unit {unit} is not supported by this channel.")
+
+        value = self._device.measure_channel(unit, self._channel_idx, check_errors)
+
+        if self._buffering_enabled:
+            if self._channel_idx not in self._buffer_index:
+                self._buffer_index[self._channel_idx] = 0
+            value = FutureValue(self._channel_idx, self._buffer_index[self._channel_idx])
+            self._buffer_index[self._channel_idx] += 1
+
+        return value
 
 
 class SourceMeasureChannel(SourceChannel, MeasureChannel):
 
     def __init__(self, device: ParentDevice, channel_idx: ChannelIndex, supported_source_units: list[ChannelUnit],
                  supported_measure_units: list[ChannelUnit], buffering_available: bool = False):
-        self._device = device
-        self._channel_idx = channel_idx
-        self._supported_source_units = supported_source_units
-        self._supported_measure_units = supported_measure_units
-        self._buffering_available = buffering_available
+        SourceChannel.__init__(self, device, channel_idx, supported_source_units, buffering_available)
+        MeasureChannel.__init__(self, device, channel_idx, supported_measure_units, buffering_available)

py_instrument_control_lib/channels/ChannelEnums.py

@@ -21,3 +21,11 @@ def check(self, number_channels: int) -> None:
 
     def get(self) -> int:
         return self._index
+
+    def __eq__(self, other):
+        if isinstance(other, ChannelIndex):
+            return other.get() == self.get()
+        return False
+
+    def __hash__(self):
+        return self.get()

py_instrument_control_lib/channels/FutureValue.py

@@ -0,0 +1,19 @@
+from typing import Any
+
+from py_instrument_control_lib.channels.ChannelEnums import ChannelIndex
+
+
+class FutureValue:
+
+    def __init__(self, channel_idx: ChannelIndex, buffer_index: int):
+        self.channel_index = channel_idx
+        self.buffer_index = buffer_index
+
+    def get(self, buffer: list) -> Any:
+        channel_buffer = buffer[self.channel_index.get() - 1]
+        if not channel_buffer or len(channel_buffer) <= self.buffer_index:
+            raise ValueError('Buffer is not filled appropriately.')
+        return channel_buffer[self.buffer_index]
+
+    def __repr__(self) -> str:
+        return f'FutureValue @ {self.buffer_index} @ C{self.channel_index.get()}'

py_instrument_control_lib/device_generation/DEFAULT_IMPORTS.py

@@ -9,6 +9,4 @@
 from py_instrument_control_lib.manufacturers.FloDevice import FloDevice
 from py_instrument_control_lib.manufacturers.KeithleyDevice import KeithleyDevice
 from py_instrument_control_lib.manufacturers.KeysightDevice import KeysightDevice
-
-_, _, _, _, _, _, _, _, _ = (SMU, FunctionGenerator, PowerSupply, Oscilloscope, TCPDevice,
-                             KeysightDevice, KeithleyDevice, AbstractSwitchMatrix, FloDevice)
+from py_instrument_control_lib.device_base.DeviceException import DeviceException

py_instrument_control_lib/device_generation/generate_module.py

@@ -100,8 +100,10 @@ def generate_code(spec: dict) -> str:
 def generate_module(filepath: str, target_dir: str = '.') -> None:
     spec = load_spec(abspath(filepath))
     module = DISCLAIMER.format(filepath) + '\n\n' + generate_code(spec)
-    with open(os.path.join(target_dir, spec['name'] + '.py'), 'w') as f:
+    file_name = os.path.join(target_dir, spec['name'] + '.py')
+    with open(file_name, 'w') as f:
         f.write(module)
+    os.system(f'python3 -m autopep8 -i --max-line-length 120 {abspath(file_name)}')
 
 
 if __name__ == '__main__':

py_instrument_control_lib/device_types/Oscilloscope.py

@@ -1,16 +1,10 @@
 from abc import ABC
 from enum import Enum
 
+from py_instrument_control_lib.channels.ChannelEnums import ChannelIndex
 from py_instrument_control_lib.device_base.Device import Device
 
 
-class OscChannel(Enum):
-    CHANNEL_1 = "1"
-    CHANNEL_2 = "2"
-    CHANNEL_3 = "3"
-    CHANNEL_4 = "4"
-
-
 class VoltageUnit(Enum):
     VOLT = ""
     MILLI_VOLT = ""
@@ -53,17 +47,17 @@ def auto_scale(self) -> None:
     def set_time_range(self, value: float) -> None:
         pass
 
-    def set_channel_offset(self, channel: OscChannel, offset: float) -> None:
+    def set_channel_offset(self, channel_idx: ChannelIndex, offset: float) -> None:
         pass
 
-    def set_channel_scale(self, channel: OscChannel, value: float) -> None:
+    def set_channel_scale(self, channel_idx: ChannelIndex, value: float) -> None:
         pass
 
-    def set_channel_range(self, channel: OscChannel, value: float, voltage_unit: VoltageUnit) -> None:
+    def set_channel_range(self, channel_idx: ChannelIndex, value: float, voltage_unit: VoltageUnit) -> None:
         pass
 
     def set_trigger_edge(self, edge: TriggerEdge) -> None:
         pass
 
-    def set_trigger_source(self, channel: OscChannel) -> None:
+    def set_trigger_source(self, channel_idx: ChannelIndex) -> None:
         pass