tessl/pypi-ophyd
Bluesky hardware abstraction library with EPICS control system integration for scientific instrument automation.
—
Pending
specialized-devices.mddocs/
Specialized Devices
Pre-built device classes for common laboratory instruments including scalers, multi-channel analyzers, and electrometers. These devices provide standardized interfaces to specific hardware types commonly found in scientific facilities.
Capabilities
Scalers
EPICS scaler devices for counting applications and beam monitoring.
class EpicsScaler(Device):
"""
EPICS scaler record interface for multi-channel counting.
Provides access to EPICS scaler records with multiple counting
channels, preset values, and timing control.
Parameters:
- prefix (str): EPICS scaler record prefix
- name (str): Scaler device name
"""
def __init__(self, prefix, *, name, **kwargs): ...
def stage(self):
"""
Stage scaler for data acquisition.
Configures channels and preset values.
"""
def unstage(self):
"""Unstage scaler after acquisition."""
def trigger(self):
"""
Trigger scaler counting sequence.
Returns:
StatusBase: Status tracking count completion
"""
def stop(self):
"""Stop scaler counting."""
def read(self):
"""
Read all enabled scaler channels.
Returns:
dict: Channel readings with counts and count rates
"""
@property
def preset_time(self):
"""
Preset counting time.
Returns:
EpicsSignal: Preset time signal
"""
@property
def elapsed_real_time(self):
"""
Elapsed real counting time.
Returns:
EpicsSignalRO: Real time readback
"""
@property
def elapsed_live_time(self):
"""
Elapsed live counting time.
Returns:
EpicsSignalRO: Live time readback
"""
class ScalerChannel(Device):
"""
Individual scaler channel with counting and configuration.
Represents a single channel of a multi-channel scaler.
"""
def __init__(self, prefix, *, ch_num, **kwargs): ...
@property
def s(self):
"""
Channel counts.
Returns:
EpicsSignalRO: Count value
"""
@property
def preset(self):
"""
Channel preset value.
Returns:
EpicsSignal: Preset count or rate
"""
@property
def gate(self):
"""
Channel gate control.
Returns:
EpicsSignal: Gate enable/disable
"""
# Alias for ScalerChannel
ScalerCH = ScalerChannelMulti-Channel Analyzers (MCA)
Devices for energy-dispersive spectroscopy and pulse height analysis.
class EpicsMCA(Device):
"""
EPICS Multi-Channel Analyzer for energy spectroscopy.
Provides interface to MCA records for pulse height analysis
and energy-dispersive measurements.
Parameters:
- prefix (str): EPICS MCA record prefix
- name (str): MCA device name
"""
def __init__(self, prefix, *, name, **kwargs): ...
def erase(self):
"""
Erase (clear) MCA spectrum data.
Returns:
StatusBase: Erase completion status
"""
def start(self):
"""
Start MCA data acquisition.
Returns:
StatusBase: Start completion status
"""
def stop(self):
"""Stop MCA data acquisition."""
def read(self):
"""
Read MCA spectrum data.
Returns:
dict: Spectrum data and metadata
"""
@property
def spectrum(self):
"""
MCA spectrum data array.
Returns:
EpicsSignalRO: Spectrum counts per channel
"""
@property
def preset_real_time(self):
"""
Preset real counting time.
Returns:
EpicsSignal: Preset time setting
"""
@property
def preset_live_time(self):
"""
Preset live counting time.
Returns:
EpicsSignal: Preset live time setting
"""
@property
def elapsed_real_time(self):
"""
Elapsed real counting time.
Returns:
EpicsSignalRO: Actual real time
"""
@property
def elapsed_live_time(self):
"""
Elapsed live counting time.
Returns:
EpicsSignalRO: Actual live time
"""
class ROI(Device):
"""
Region of Interest for MCA spectrum analysis.
Defines energy windows for peak integration and analysis.
"""
def __init__(self, prefix, *, roi_num, **kwargs): ...
@property
def count(self):
"""
ROI integrated counts.
Returns:
EpicsSignalRO: Total counts in ROI
"""
@property
def net_count(self):
"""
ROI net counts (background subtracted).
Returns:
EpicsSignalRO: Net counts in ROI
"""
@property
def preset_count(self):
"""
ROI preset count value.
Returns:
EpicsSignal: Preset counts for ROI
"""
class EpicsDXP(Device):
"""
Digital X-ray Processor interface for advanced MCA systems.
Provides control of DXP-based multi-element detector systems
with advanced signal processing capabilities.
"""
def __init__(self, prefix, *, name, **kwargs): ...
def erase(self):
"""Erase all DXP spectrum data."""
def start(self):
"""Start DXP acquisition."""
def stop(self):
"""Stop DXP acquisition."""Quad Electrometers
Four-channel electrometers for current measurement and beam monitoring.
class QuadEM(Device):
"""
Base class for quad electrometer devices.
Provides four-channel current measurement with position
calculation capabilities for beam monitoring.
Parameters:
- prefix (str): Device PV prefix
- name (str): Device name
"""
def __init__(self, prefix, *, name, **kwargs): ...
def stage(self):
"""Stage electrometer for acquisition."""
def unstage(self):
"""Unstage electrometer."""
def trigger(self):
"""
Trigger electrometer reading.
Returns:
StatusBase: Trigger completion status
"""
def read(self):
"""
Read all electrometer channels and computed values.
Returns:
dict: Current readings and position calculations
"""
@property
def current1(self):
"""
Channel 1 current reading.
Returns:
EpicsSignalRO: Current in amperes
"""
@property
def current2(self):
"""
Channel 2 current reading.
Returns:
EpicsSignalRO: Current in amperes
"""
@property
def current3(self):
"""
Channel 3 current reading.
Returns:
EpicsSignalRO: Current in amperes
"""
@property
def current4(self):
"""
Channel 4 current reading.
Returns:
EpicsSignalRO: Current in amperes
"""
@property
def sum_all(self):
"""
Sum of all four channels.
Returns:
EpicsSignalRO: Total current
"""
@property
def pos_x(self):
"""
Calculated X position from quad currents.
Returns:
EpicsSignalRO: X position
"""
@property
def pos_y(self):
"""
Calculated Y position from quad currents.
Returns:
EpicsSignalRO: Y position
"""
class NSLS_EM(QuadEM):
"""
NSLS-style electrometer implementation.
Specific implementation for NSLS (National Synchrotron Light Source)
electrometer hardware and control systems.
"""
def __init__(self, prefix, *, name, **kwargs): ...
class TetrAMM(QuadEM):
"""
TetrAMM electrometer device.
Interface to TetrAMM (Tetrode Ammeter) hardware for
four-channel current measurement.
"""
def __init__(self, prefix, *, name, **kwargs): ...
class APS_EM(QuadEM):
"""
APS-style electrometer implementation.
Specific implementation for APS (Advanced Photon Source)
electrometer hardware and control systems.
"""
def __init__(self, prefix, *, name, **kwargs): ...
class QuadEMPort(Device):
"""
Individual port/channel of a quad electrometer.
Represents a single current input channel with its
associated configuration and readback values.
"""
def __init__(self, prefix, *, port_name, **kwargs): ...Ion Chambers and Current Monitors
Additional specialized current measurement devices.
class IonChamber(Device):
"""
Ion chamber device for X-ray intensity monitoring.
Provides current measurement and voltage control for
gas-filled ion chambers used in X-ray beam monitoring.
"""
def __init__(self, prefix, *, name, **kwargs): ...
@property
def current(self):
"""
Ion chamber current reading.
Returns:
EpicsSignalRO: Current in amperes
"""
@property
def voltage(self):
"""
Ion chamber bias voltage.
Returns:
EpicsSignal: Bias voltage setting
"""
class CurrentAmplifier(Device):
"""
Current amplifier device for low-level current measurement.
Provides amplification and measurement of small currents
from photodiodes, ion chambers, and other current sources.
"""
def __init__(self, prefix, *, name, **kwargs): ...
@property
def current(self):
"""
Amplified current reading.
Returns:
EpicsSignalRO: Current measurement
"""
@property
def gain(self):
"""
Amplifier gain setting.
Returns:
EpicsSignal: Gain value
"""Usage Examples
Scaler Usage
from ophyd import EpicsScaler
from ophyd.status import wait
# Create scaler device
scaler = EpicsScaler('XF:28IDC:SCALER:', name='scaler')
scaler.wait_for_connection()
# Configure counting time
scaler.preset_time.put(1.0) # 1 second counting
# Stage and trigger counting
scaler.stage()
status = scaler.trigger()
wait(status) # Wait for counting to complete
# Read results
reading = scaler.read()
for channel_name, data in reading.items():
print(f"{channel_name}: {data['value']} counts")
# Get individual channel data
ch1_counts = scaler.channels.chan1.s.get()
print(f"Channel 1: {ch1_counts} counts")
scaler.unstage()MCA Spectroscopy
from ophyd import EpicsMCA
from ophyd.status import wait
# Create MCA device
mca = EpicsMCA('XF:28IDC:MCA:', name='mca')
mca.wait_for_connection()
# Configure acquisition
mca.preset_real_time.put(10.0) # 10 second acquisition
mca.erase() # Clear previous data
# Start acquisition
status = mca.start()
wait(status)
# Read spectrum
spectrum_data = mca.spectrum.get()
print(f"Spectrum shape: {spectrum_data.shape}")
print(f"Total counts: {spectrum_data.sum()}")
# Work with ROIs
roi1 = mca.rois.roi1
roi1.left.put(100) # Set ROI start channel
roi1.right.put(200) # Set ROI end channel
roi_counts = roi1.count.get()
net_counts = roi1.net_count.get()
print(f"ROI counts: {roi_counts}, Net: {net_counts}")Quad Electrometer Monitoring
from ophyd import NSLS_EM
from ophyd.status import wait
# Create quad electrometer
qem = NSLS_EM('XF:28IDC:QEM:', name='quadem')
qem.wait_for_connection()
# Configure acquisition parameters
qem.acquire_mode.put('Single')
qem.averaging_time.put(0.1) # 100ms averaging
# Stage and trigger reading
qem.stage()
status = qem.trigger()
wait(status)
# Read all channels
reading = qem.read()
# Get individual currents
i1 = qem.current1.get()
i2 = qem.current2.get()
i3 = qem.current3.get()
i4 = qem.current4.get()
print(f"Currents: I1={i1:.3e}, I2={i2:.3e}, I3={i3:.3e}, I4={i4:.3e}")
# Get computed position
x_pos = qem.pos_x.get()
y_pos = qem.pos_y.get()
total_current = qem.sum_all.get()
print(f"Position: X={x_pos:.3f}, Y={y_pos:.3f}")
print(f"Total current: {total_current:.3e} A")
qem.unstage()Multi-Element DXP System
from ophyd import EpicsDXP
# Create DXP system
dxp = EpicsDXP('XF:28IDC:DXP:', name='dxp_detector')
dxp.wait_for_connection()
# Configure acquisition for all elements
dxp.preset_real_time.put(60.0) # 60 second count
dxp.erase() # Clear all spectra
# Start acquisition
status = dxp.start()
wait(status)
# Read data from all elements
reading = dxp.read()
# Access individual detector elements
for i in range(4): # Assuming 4-element detector
element = getattr(dxp, f'mca{i+1}')
spectrum = element.spectrum.get()
dead_time = element.elapsed_dead_time.get()
print(f"Element {i+1}:")
print(f" Total counts: {spectrum.sum()}")
print(f" Dead time: {dead_time:.1f}%")
# Get element-specific ROIs
roi = element.rois.roi1
roi_counts = roi.count.get()
print(f" ROI 1 counts: {roi_counts}")Beam Position Monitoring
from ophyd import TetrAMM
import time
# Create beam position monitor
bpm = TetrAMM('XF:28IDC:BPM:', name='beam_position_monitor')
bpm.wait_for_connection()
# Monitor beam position continuously
print("Monitoring beam position (Ctrl+C to stop):")
try:
while True:
# Trigger reading
bpm.stage()
status = bpm.trigger()
wait(status)
# Get position and intensity
x_pos = bpm.pos_x.get()
y_pos = bpm.pos_y.get()
intensity = bpm.sum_all.get()
print(f"Position: X={x_pos:+6.3f} mm, Y={y_pos:+6.3f} mm, "
f"Intensity: {intensity:.2e} A")
bpm.unstage()
time.sleep(1.0) # Update every second
except KeyboardInterrupt:
print("Monitoring stopped")Custom Multi-Device Setup
from ophyd import Device, Component
from ophyd import EpicsScaler, NSLS_EM, EpicsMCA
class BeamlineInstruments(Device):
"""Combined beamline instrumentation."""
# Intensity monitoring
scaler = Component(EpicsScaler, 'SCALER:')
ion_chamber = Component(NSLS_EM, 'IC1:')
# Spectroscopy
mca = Component(EpicsMCA, 'MCA:')
def monitor_beam(self, count_time=1.0):
"""Monitor beam intensity and position."""
# Configure devices
self.scaler.preset_time.put(count_time)
self.ion_chamber.averaging_time.put(count_time)
# Stage all devices
self.scaler.stage()
self.ion_chamber.stage()
# Trigger simultaneously
scaler_status = self.scaler.trigger()
ic_status = self.ion_chamber.trigger()
# Wait for completion
wait([scaler_status, ic_status])
# Read results
scaler_reading = self.scaler.read()
ic_reading = self.ion_chamber.read()
# Unstage
self.scaler.unstage()
self.ion_chamber.unstage()
return {
'scaler': scaler_reading,
'ion_chamber': ic_reading
}
# Use combined instrument
instruments = BeamlineInstruments('XF:28IDC:', name='instruments')
instruments.wait_for_connection()
# Monitor beam conditions
results = instruments.monitor_beam(count_time=0.5)
print("Beam monitoring results:")
print(results)Install with Tessl CLI
npx tessl i tessl/pypi-ophyd