tessl/pypi-ophyd
Bluesky hardware abstraction library with EPICS control system integration for scientific instrument automation.
—
Pending
core-framework.mddocs/
Core Framework
The foundational classes and interfaces that form the backbone of ophyd's device abstraction system. These components enable device composition, signal management, status tracking, and provide the base functionality that all other ophyd components build upon.
Capabilities
Device Class
The base class for all hardware devices in ophyd, providing a standardized interface for reading, configuring, and controlling hardware through a component-based architecture.
class Device:
"""
Base class for all devices in ophyd.
Parameters:
- prefix (str): PV prefix for EPICS devices
- name (str): Device name for identification
- kind (Kind): Component kind classification
- read_attrs (list): Attributes to include in read()
- configuration_attrs (list): Attributes for configuration
- parent (Device): Parent device if nested
- child_name_separator (str): Separator for child component names (default: '_')
- connection_timeout (float): Timeout for signal connections
"""
def __init__(self, prefix='', *, name, kind=None, read_attrs=None, configuration_attrs=None, parent=None, child_name_separator='_', connection_timeout=None, **kwargs): ...
def read(self):
"""
Read current values from all readable components.
Returns:
dict: Mapping of component names to reading dictionaries
with 'value' and 'timestamp' keys
"""
def describe(self):
"""
Describe the data format for all readable components.
Returns:
dict: Mapping of component names to description dictionaries
with 'dtype', 'shape', and other metadata
"""
def configure(self, d):
"""
Configure device settings.
Parameters:
- d (dict): Configuration parameters mapping
Returns:
tuple: (old_config, new_config) dictionaries
"""
def read_configuration(self):
"""
Read current configuration values.
Returns:
dict: Current configuration state
"""
def describe_configuration(self):
"""
Describe configuration data format.
Returns:
dict: Configuration data descriptions
"""
def trigger(self):
"""
Trigger device acquisition or measurement.
Returns:
StatusBase: Status object tracking trigger completion
"""
def stage(self):
"""
Stage device for data acquisition (setup phase).
Returns:
list: List of staged components
"""
def unstage(self):
"""
Unstage device after data acquisition (cleanup phase).
Returns:
list: List of unstaged components
"""
def wait_for_connection(self, timeout=2.0):
"""
Wait for all device components to connect.
Parameters:
- timeout (float): Maximum wait time in seconds
"""
def connected(self):
"""
Check if device is fully connected.
Returns:
bool: True if all components are connected
"""
@property
def hints(self):
"""
Hints for plotting and analysis.
Returns:
dict: Plot hints with 'fields' key
"""Signal Classes
Fundamental classes for representing individual hardware channels and computed values, forming the building blocks of device components.
class Signal:
"""
Base class for readable and writable values.
Parameters:
- name (str): Signal name for identification
- value: Initial value (default: 0.0)
- dtype: Data type specification
- shape: Data shape specification
- timestamp (float): Initial timestamp
- parent (Device): Parent device
- labels (list): Additional signal labels
- kind (Kind): Signal classification (default: Kind.hinted)
- tolerance (float): Absolute tolerance for set operations
- rtolerance (float): Relative tolerance for set operations
- metadata (dict): Additional metadata
- cl: Control layer
- attr_name (str): Parent Device attribute name
"""
def __init__(self, *, name, value=0.0, dtype=None, shape=None, timestamp=None, parent=None, labels=None, kind=Kind.hinted, tolerance=None, rtolerance=None, metadata=None, cl=None, attr_name="", **kwargs): ...
def get(self, **kwargs):
"""
Get current signal value.
Returns:
Current value of the signal
"""
def put(self, value, **kwargs):
"""
Put value to signal (synchronous).
Parameters:
- value: Value to write
"""
def set(self, value, **kwargs):
"""
Set signal to value (asynchronous).
Parameters:
- value: Target value
Returns:
StatusBase: Status object tracking set completion
"""
def read(self):
"""
Read signal value and timestamp.
Returns:
dict: Reading dictionary with 'value' and 'timestamp'
"""
def describe(self):
"""
Describe signal data format.
Returns:
dict: Description with 'dtype', 'shape', 'source', etc.
"""
def subscribe(self, callback, event_type=None, run=True):
"""
Subscribe to signal changes.
Parameters:
- callback (callable): Function to call on changes
- event_type (str): Type of event to monitor
- run (bool): Whether to run callback immediately
Returns:
int: Subscription ID for unsubscribing
"""
def clear_sub(self, cb, event_type=None):
"""
Clear subscription.
Parameters:
- cb: Subscription ID or callback function
- event_type (str): Event type to clear
"""
class SignalRO(Signal):
"""
Read-only signal that cannot be written to.
"""
def __init__(self, *, name, value=0, **kwargs): ...
class DerivedSignal(Signal):
"""
Signal derived from other signals through a function.
Parameters:
- derived_from (dict): Mapping of signals to derive from
- derived_func (callable): Function to compute derived value
"""
def __init__(self, *, derived_from, derived_func=None, **kwargs): ...Component System
The component system enables composition of devices from reusable, declarative building blocks.
class Component:
"""
Device component descriptor for building composite devices.
Parameters:
- cls: Component class to instantiate
- suffix (str): PV suffix to append to device prefix
- lazy (bool): Whether to delay instantiation
- trigger_value: Value to set during trigger()
- add_prefix (tuple): Additional prefix components
- doc (str): Documentation string
- kind (Kind): Component classification
"""
def __init__(self, cls, suffix='', *, lazy=False, trigger_value=None, add_prefix=None, doc=None, kind='normal', **kwargs): ...
class FormattedComponent(Component):
"""
Component with string formatting support for dynamic PV names.
Parameters:
- cls: Component class
- suffix (str): PV suffix with format strings
- **kwargs: Format arguments and component parameters
"""
def __init__(self, cls, suffix, **kwargs): ...
class DynamicDeviceComponent(Component):
"""
Component that creates devices dynamically based on available PVs.
"""
def __init__(self, defn, **kwargs): ...
# Component utility functions
def kind_context(kind):
"""
Context manager for setting default component kind.
Parameters:
- kind (Kind): Default kind for components created in context
"""
ALL_COMPONENTS = object() # Sentinel for selecting all componentsStatus Management
Status objects track the progress and completion of asynchronous operations.
class StatusBase:
"""
Base class for tracking operation status.
Parameters:
- timeout (float): Maximum time to wait for completion (None = wait forever)
- settle_time (float): Time to wait after completion before callbacks (default: 0)
- done (deprecated): Initial completion state
- success (deprecated): Initial success state
"""
def __init__(self, *, timeout=None, settle_time=0, done=None, success=None): ...
@property
def done(self):
"""
Whether operation is complete.
Returns:
bool: True if operation finished (success or failure)
"""
@property
def success(self):
"""
Whether operation completed successfully.
Returns:
bool: True if completed without errors
"""
def wait(self, timeout=None):
"""
Wait for operation completion.
Parameters:
- timeout (float): Maximum wait time
Raises:
StatusTimeoutError: If timeout is exceeded
"""
def add_callback(self, callback):
"""
Add callback for status completion.
Parameters:
- callback (callable): Function to call when complete
"""
class Status(StatusBase):
"""
Basic status implementation with manual completion control.
"""
def __init__(self, obj=None, **kwargs): ...
def set_finished(self):
"""Mark status as successfully completed."""
def set_exception(self, exc):
"""Mark status as failed with exception."""
class DeviceStatus(StatusBase):
"""
Status for device operations, can combine multiple sub-statuses.
"""
def __init__(self, device, **kwargs): ...
def add_status(self, status):
"""Add sub-status to track."""
class MoveStatus(StatusBase):
"""
Status specifically for positioner movements.
"""
def __init__(self, positioner, target, **kwargs): ...
def wait(status, timeout=None, poll_period=0.05):
"""
Wait for one or more status objects to complete.
Parameters:
- status (StatusBase or list): Status object(s) to wait for
- timeout (float): Maximum total wait time
- poll_period (float): Polling interval
Raises:
StatusTimeoutError: If any status times out
WaitTimeoutError: If overall timeout is exceeded
"""Object Registry and Utilities
Utilities for managing ophyd objects and their relationships.
class OphydObject:
"""
Base class for all ophyd objects providing common functionality.
Parameters:
- name (str): Object name
- parent (OphydObject): Parent object
"""
def __init__(self, *, name=None, parent=None, **kwargs): ...
@property
def name(self):
"""Full dotted name including parent hierarchy."""
@property
def dotted_name(self):
"""Dotted name for hierarchical identification."""
def destroy(self):
"""Clean up object resources."""
# Registry functions
def register_instances_keyed_on_name(cls, registry=None):
"""
Register class instances in a dictionary keyed by name.
Parameters:
- cls: Class to register instances for
- registry (dict): Dictionary to store instances
"""
def register_instances_in_weakset(cls, registry=None):
"""
Register class instances in a weak set.
Parameters:
- cls: Class to register instances for
- registry (WeakSet): Set to store instances
"""
def select_version(obj, version):
"""
Select specific version of an object.
Parameters:
- obj: Object with version support
- version: Version identifier
"""Utility Functions
Helper functions for namespace management, string processing, and instance handling.
class OrderedDefaultDict(OrderedDict):
"""
Combination of defaultdict and OrderedDict.
Provides ordered dictionary functionality with automatic
default value creation for missing keys.
Parameters:
- default_factory (callable): Factory function for default values
"""
def __init__(self, default_factory=None, *a, **kw): ...
def __missing__(self, key): ...
def copy(self): ...
def instances_from_namespace(classes, *, ns=None):
"""
Get instances of specified classes from user namespace.
Searches the IPython user namespace (or provided namespace)
for instances of the specified class types.
Parameters:
- classes: Type or sequence of types to match with isinstance()
- ns (dict): Namespace to search (defaults to IPython user_ns)
Returns:
list: Instances matching the specified classes
"""
def ducks_from_namespace(attrs, *, ns=None):
"""
Get instances with specified attributes (duck typing).
Finds objects that have all specified attributes, implementing
"duck typing" - if it looks like a duck and quacks like a duck...
Parameters:
- attrs: Attribute name(s) to check for
- ns (dict): Namespace to search (defaults to IPython user_ns)
Returns:
list: Objects that have all specified attributes
"""
def underscores_to_camel_case(underscores):
"""
Convert underscore-separated strings to camelCase.
Parameters:
- underscores (str): String with underscores (e.g., "abc_def_ghi")
Returns:
str: CamelCase version (e.g., "abcDefGhi")
"""
def getattrs(obj, gen):
"""
Get multiple attributes from an object as generator.
Parameters:
- obj: Object to get attributes from
- gen: Iterable of attribute names
Yields:
tuple: (attribute_name, attribute_value) pairs
"""
def doc_annotation_forwarder(base_klass):
"""
Decorator that forwards documentation and annotations from base class.
Copies __doc__ and __annotations__ from base class method to wrapper.
Useful for maintaining documentation in method override patterns.
Parameters:
- base_klass: Base class to forward documentation from
Returns:
callable: Decorator function
"""
def adapt_old_callback_signature(callback):
"""
Adapt old callback signatures for backward compatibility.
Wraps callbacks with signature callback() to match expected
callback(status) signature for status objects.
Parameters:
- callback: Callable with signature callback() or callback(status)
Returns:
callable: Callback with signature callback(status)
"""Usage Examples
Creating a Custom Device
from ophyd import Device, Component, EpicsSignal
class MyTemperatureController(Device):
"""Temperature controller with setpoint and readback."""
setpoint = Component(EpicsSignal, 'SP', kind='config')
temperature = Component(EpicsSignal, 'RBV', kind='hinted')
status = Component(EpicsSignal, 'STAT', kind='omitted')
def heat_to(self, temperature):
"""Heat to target temperature."""
return self.setpoint.set(temperature)
# Create and use device
temp_controller = MyTemperatureController('TEMP:CTRL:', name='temp_ctrl')
temp_controller.wait_for_connection()
# Read all values
reading = temp_controller.read()
print(f"Current temperature: {reading['temp_ctrl_temperature']['value']}")
# Set temperature and wait
status = temp_controller.heat_to(25.0)
wait(status)Working with Signals
from ophyd import Signal, DerivedSignal
# Create base signals
voltage = Signal(name='voltage', value=5.0)
current = Signal(name='current', value=2.0)
# Create derived signal for power
def calculate_power(voltage=None, current=None, **kwargs):
return voltage * current
power = DerivedSignal(
derived_from={'voltage': voltage, 'current': current},
derived_func=calculate_power,
name='power'
)
# Subscribe to changes
def power_changed(value=None, **kwargs):
print(f"Power changed to: {value} W")
power.subscribe(power_changed)
# Update base signals
voltage.put(6.0) # Triggers power calculation and callbackInstall with Tessl CLI
npx tessl i tessl/pypi-ophyd