Skip to content

Commit

Permalink
refactor(Virtual): .config files now use address VirtualDevice
Browse files Browse the repository at this point in the history
All other elements of the .config files match real equipment use cases!
Test(Virtual): added framework for making tests of all device functions using new VirtualDevice.
  • Loading branch information
TedKus committed Nov 7, 2024
1 parent 7585e48 commit 2b805c5
Show file tree
Hide file tree
Showing 4 changed files with 256 additions and 63 deletions.
29 changes: 29 additions & 0 deletions examples/using_virtual_devices/new_virtual_config.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
{
"virtual_source": {
"object": "Keithley_2231A",
"definition": "pythonequipmentdrivers.source",
"address": "VirtualDevice",
"init": [
["set_voltage", {"voltage": 0}],
["off", {}],
["set_current", {"current": 0}]
]
},
"virtual_sink": {
"object": "Chroma_63206A",
"definition": "pythonequipmentdrivers.sink",
"address": "VirtualDevice",
"init": [
["set_current", {"current": 0}],
["set_mode", {"mode": "CC"}]
]
},
"virtual_multimeter": {
"object": "Keithley_2700",
"definition": "pythonequipmentdrivers.multimeter",
"address": "VirtualDevice",
"init": [
["set_mode", {"mode": "VDC"}]
]
}
}
98 changes: 98 additions & 0 deletions examples/using_virtual_devices/virtual_device_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
import pythonequipmentdrivers as ped
from pathlib import Path


def test_source_operations(source):
"""Test various source operations"""
print("\n=== Testing Source Operations ===")

# Test voltage operations
print("\nVoltage Operations:")
source.set_voltage(12.0)
voltage = source.measure_voltage()
print(f"Set voltage: 12.0V, Measured: {voltage}V")

# Test current operations
print("\nCurrent Operations:")
source.set_current(2.0)
current = source.measure_current()
print(f"Set current: 2.0A, Measured: {current}A")

# Test power state
print("\nPower State Operations:")
source.on()
source.off()

# Check call history
history = source.get_call_history()
print("\nSource Call History:")
for method, args in history.items():
print(f"Method: {method}, Args: {args}")


def test_sink_operations(sink):
"""Test various sink operations"""
print("\n=== Testing Sink Operations ===")

# Test current operations
print("\nCurrent Operations:")
sink.set_current(1.0)
current = sink.measure_current()
print(f"Set current: 1.0A, Measured: {current}A")

# Test mode operations
print("\nMode Operations:")
sink.set_mode("CC")

# Test custom measurement values
print("\nCustom Measurement Test:")
sink.set_measurement_value("measure_voltage", 24.0)
voltage = sink.measure_voltage()
print(f"Custom voltage measurement: {voltage}V")


def test_multimeter_operations(dmm):
"""Test various multimeter operations"""
print("\n=== Testing Multimeter Operations ===")

# Test voltage measurements
print("\nVoltage Measurements:")
dmm.set_mode("VDC")
voltage = dmm.measure_voltage()
print(f"Measured voltage: {voltage}V")

# Test custom measurements
print("\nCustom Measurement Test:")
dmm.set_measurement_value("measure_voltage", 5.0)
voltage = dmm.measure_voltage()
print(f"Custom voltage measurement: {voltage}V")


def main():
# Load Virtual devices configuration
config_path = Path(__file__).parent / "new_virtual_config.json"
equipment = ped.connect_resources(config=config_path, init=True)

# Test each virtual device
test_source_operations(equipment.virtual_source)
test_sink_operations(equipment.virtual_sink)
test_multimeter_operations(equipment.virtual_multimeter)

# Demonstrate error handling
print("\n=== Testing Error Handling ===")
try:
equipment.virtual_source.invalid_method()
except Exception as e:
print(f"Caught expected error for invalid method: {e}")

# Test device state persistence
print("\n=== Testing State Persistence ===")
equipment.virtual_source.set_voltage(15.0)
print(f"Voltage after setting: {equipment.virtual_source.measure_voltage()}V")
print(f"Voltage using get_voltage: {equipment.virtual_source.get_voltage()}V")

print("\nTest completed successfully!")


if __name__ == "__main__":
main()
161 changes: 109 additions & 52 deletions pythonequipmentdrivers/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -408,64 +408,121 @@ def group_execute_trigger(self, *trigger_devices):


class VirtualDevice:
def __init__(self, address: str, mimic: str = None, **kwargs) -> None:
def __init__(self, address: str = "virtualdevice", definition: str = None,
object: str = None, **kwargs) -> None:
"""Initialize a virtual device that mimics a real instrument.
Args:
address (str): Virtual address for the device
definition (str): Import path for the device class module
object (str): Name of the device class to virtualize
**kwargs: Additional attributes to set on the virtual device
"""
self.address = address
self.mimicked_device_name = None
self.mimicked_device_class = None
self.state = {}
self.state = {} # Stores method call history
self.values = {} # Stores current device state/settings
self.attributes = kwargs
if mimic:
self.mimic(mimic)

def mimic(self, object: str, definition: str = None):
self.mimicked_device_name = object
self.mimicked_device_definition = definition
if definition:
module = __import__(definition, fromlist=[object])
self.mimicked_device_class = getattr(module, object)
# Initialize attributes from the mimicked class's __init__ method
sig = inspect.signature(self.mimicked_device_class.__init__)
for param in sig.parameters.values():
if param.name not in ['self', 'address'] and \
param.name not in self.attributes:
if param.default is not param.empty:
self.attributes[param.name] = param.default
else:
self.attributes[param.name] = None
self.device_class = None
self.methods = {} # Stores method signatures and return types

if definition and object:
try:
# Import and analyze the device class
module = __import__(definition, fromlist=[object])
self.device_class = getattr(module, object)
self._analyze_device_class()
except (ImportError, AttributeError) as e:
warnings.warn(f"Failed to load device class {object} "
f"from {definition}: {e}")

def _analyze_device_class(self):
"""Analyze device class methods and create virtual equivalents"""
for name, method in inspect.getmembers(self.device_class, inspect.isfunction):
if name.startswith('_'):
continue

# Get method signature and return type
sig = inspect.signature(method)
return_type = get_type_hints(method).get('return')

# Store method info
self.methods[name] = {
'signature': sig,
'return_type': return_type,
'default_value': self._get_default_value(return_type)
}

# Initialize state values for measurement/getter methods
if name.startswith(('measure_', 'get_')):
self.values[name] = self._get_default_value(return_type)

def _get_default_value(self, return_type):
"""Get appropriate default value based on return type"""
if return_type is None:
return None
elif return_type is bool:
return False
elif return_type is int:
return 0
elif return_type is float:
return 0.0
elif return_type is str:
return ""
elif (hasattr(return_type, '__origin__') and
return_type.__origin__ is list):
return []
return None

def _update_state(self, name: str, *args, **kwargs):
"""Update device state based on method calls"""
# Record method call
self.state[name] = (args, kwargs)

# Update values based on method type
if name.startswith('set_'):
# Extract value from set_* methods
get_name = f"get_{name[4:]}" # set_voltage -> get_voltage
measure_name = f"measure_{name[4:]}" # set_voltage -> measure_voltage

# Get the value being set
value = next(iter(kwargs.values())) if kwargs else args[0]

# Update corresponding get/measure methods
if get_name in self.methods:
self.values[get_name] = value
if measure_name in self.methods:
self.values[measure_name] = value

def __getattr__(self, name: str) -> Any:
"""Handle method calls and attribute access"""
if name in self.attributes:
return self.attributes[name]

def method(*args, **kwargs):
if name in self.methods:
def method(*args, **kwargs):
self._update_state(name, *args, **kwargs)

# For measurement/getter methods, return stored value
if name in self.values:
return self.values[name]

# Otherwise return default for return type
return self.methods[name]['default_value']
return method

# Default method for unknown calls
def default_method(*args, **kwargs):
self.state[name] = (args, kwargs)
if self.mimicked_device_class and hasattr(
self.mimicked_device_class,
name):
original_method = getattr(self.mimicked_device_class, name)
return_type = get_type_hints(original_method).get('return')
if return_type:
if return_type == bool:
return False
elif return_type == int:
return 0
elif return_type == float:
return 0.0
elif return_type == str:
return ""
elif (hasattr(return_type, '__origin__') and
return_type.__origin__ == list):
return []
# Add more type checks as needed
return None
return method

def __setattr__(self, name: str, value: Any) -> None:
if name in ['address',
'mimicked_device_name',
'mimicked_device_class',
'state',
'attributes']:
super().__setattr__(name, value)
else:
self.attributes[name] = value
return default_method

def get_call_history(self, method_name: str = None) -> dict:
"""Get history of method calls"""
if method_name:
return self.state.get(method_name)
return self.state

def set_measurement_value(self, method_name: str, value: Any):
"""Set a value to be returned by a measurement method"""
if method_name in self.methods:
self.values[method_name] = value
31 changes: 20 additions & 11 deletions pythonequipmentdrivers/resource_collections.py
Original file line number Diff line number Diff line change
Expand Up @@ -293,19 +293,28 @@ def connect_resources(config: Union[str, Path, dict],
resources = ResourceCollection()
dmms = {}
for name, meta_info in collection_config.items():

try:
# get object to instantate from it's source module
module_name = meta_info.pop("definition")
Module = import_module(module_name)
Resource = getattr(Module, meta_info.pop("object"))

# special keyword for resource initialization, not passed as kwarg
init_sequence = meta_info.pop("init", [])
# Get module_name before potential pop operations
module_name = meta_info.get("definition", "")

# Handle virtual devices based on address
if meta_info.get("address", "").lower() == "virtualdevice":
resource = VirtualDevice(**meta_info)
init_sequence = meta_info.pop("init", [])
else:
# get object to instantiate from it's source module
module_name = meta_info.pop("definition")
Module = import_module(module_name)
Resource = getattr(Module, meta_info.pop("object"))

# special keyword for resource initialization, not passed
# as kwarg
init_sequence = meta_info.pop("init", [])

# create instance of Resource called 'name', any remaining
# items in meta_info will be passed as kwargs
resource = Resource(**meta_info)

# create instance of Resource called 'name', any remaining items in
# meta_info will be passed as kwargs
resource = Resource(**meta_info)
setattr(resources, name, resource)

if "multimeter" in module_name:
Expand Down

0 comments on commit 2b805c5

Please sign in to comment.