Skip to content

v0.17

Choose a tag to compare

View all tags
@coretl coretl released this 04 Jun 15:19
· 8 commits to main since this release
v0.17
fab63a8
This commit was created on GitHub.com and signed with GitHub’s verified signature.
GPG key ID: B5690EEEBB952194
Verified
Learn about vigilant mode.

Breaking changes

AreaDetector initialisation

The data writing init parameters for AreaDetector have been consolidated into a single ADWriterFactory argument. Multiple writer factories can be passed. All AreaDetector subclasses in ophyd-async have been updated:

# old
detector = adaravis.AravisDetector("PREFIX:", static_path_provider, writer_cls=adcore.ADWriterType.HDF)
# new
detector = adaravis.AravisDetector("PREFIX:", adcore.ADWriterFactory.hdf(static_path_provider))

See https://blueskyproject.io/ophyd-async/main/explanations/decisions/0016-ad-writer-factory.html for rationale

StandardDetector rewrite

All detector implementations in ophyd-async are updated, external detector implementations need updating:

Detector Controller → Trigger Logic + Arm Logic

# old
class SimController(DetectorController):
    def __init__(self, driver: ADBaseIO):
        self.driver = driver
        
    async def prepare(self, trigger_info: TriggerInfo):
        assert trigger_info.trigger == TriggerInfo.INTERNAL, "Can only do internal"
        await self.driver.num_images.set(trigger_info.number_of_events)
        
    async def arm(self):
        await self.driver.acquire.set(True)
        
    async def wait_for_idle(self):
        await wait_for_value(self.driver.acquire, False, timeout=DEFAULT_TIMEOUT)
        
    async def disarm(self):
        await self.driver.acquire.set(False)

# new  
class SimTriggerLogic(DetectorTriggerLogic):
    def __init__(self, driver: ADBaseIO):
        self.driver = driver
        
    # Also prepare_edge and prepare_level if hardware triggering supported
    async def prepare_internal(self, num: int, livetime: float, deadtime: float):
        await self.driver.num_images.set(num)

# if ADArmLogic is not suitable
class SimArmLogic(DetectorArmLogic):
    def __init__(self, driver: ADBaseIO):
        self.driver = driver
        
    async def arm(self):
        await self.driver.acquire.set(True)
        
    async def wait_for_idle(self):
        await wait_for_value(self.driver.acquire, False, timeout=DEFAULT_TIMEOUT)
        
    async def disarm(self):
        await self.driver.acquire.set(False)

Detector Writer → Data Logic

# old
class ADHDFWriter(DetectorWriter):
    async def open(self, name: str, exposures_per_event: int = 1):
        # Setup file writing
        return describe_dict
        
# new
class ADHDFDataLogic(DetectorDataLogic):
    async def prepare_unbounded(self, detector_name: str):
        # Setup file writing  
        return StreamResourceDataProvider(...)

TriggerInfo Updates

# old
TriggerInfo(
    number_of_events=10,
    trigger=DetectorTrigger.EDGE_TRIGGER,
    livetime=0.1,
    deadtime=0.01,
)

# new  
TriggerInfo(
    number_of_events=10,
    trigger=DetectorTrigger.EXTERNAL_EDGE,
    livetime=0.1,
    deadtime=0.01,
)

Complete SimDetector Example

# old - controller and writer_cls.with_io

from ophyd_async.epics import adcore

class SimDetector(adcore.AreaDetector[SimController]):
    def __init__(
        self,
        prefix: str,
        path_provider: PathProvider,
        drv_suffix="cam1:",
        writer_cls: type[adcore.ADWriter] = adcore.ADHDFWriter,
        fileio_suffix: str | None = None,
        name="",
        config_sigs: Sequence[SignalR] = (),
        plugins: dict[str, adcore.NDPluginBaseIO] | None = None,
    ):
        driver = adcore.ADBaseIO(prefix + drv_suffix)
        controller = SimController(driver)        
        writer = writer_cls.with_io(
            prefix,
            path_provider,
            dataset_source=driver,
            fileio_suffix=fileio_suffix,
            plugins=plugins,
        )        
        super().__init__(
            controller=controller,
            writer=writer,
            plugins=plugins,
            name=name,
            config_sigs=config_sigs,
        )

# new - handled by the baseclass
from ophyd_async.epics import adcore

class SimDetector(adcore.AreaDetector[adcore.ADBaseIO]):
    """Create an ADSimDetector AreaDetector instance."""

    def __init__(
        self,
        prefix: str,
        *writer_factories: ADWriterFactory,
        driver_suffix="cam1:",
        plugins: dict[str, NDPluginBaseIO] | None = None,
        config_sigs: Sequence[SignalR] = (),
        name: str = "",
    ) -> None:
        driver = ADBaseIO(prefix + driver_suffix)
        super().__init__(
            driver,
            prefix,
            *writer_factories,
            acquire_logic=ADAcquireLogic(driver),
            trigger_logic=SimDetectorTriggerLogic(driver),
            plugins=plugins,
            config_sigs=config_sigs,
            name=name,
        )

Reading Stats Without Files

# old - not easily supported

# new - use PluginSignalDataLogic
detector = SimDetector(prefix, writer_type=None)
detector.add_detector_logics(PluginSignalDataLogic(driver, stats.total))
# Now stats.total appears in read() without file writing

Multiple Data Streams

# old - required complex inheritance

# new - add multiple data logics
detector = AreaDetector(
    driver=driver,
    arm_logic=ADArmLogic(driver),
    writer_type=None,  # Don't create default writer
)
# Add separate HDF writers for different ROIs
detector.add_detector_logics(
    ADHDFDataLogic(..., datakey_suffix="-roi1"),
    ADHDFDataLogic(..., datakey_suffix="-roi2"),
)

What's Changed

New Contributors

Full Changelog: v0.16...v0.17

Contributors

tacaswell, renovate, and 10 other contributors
Assets 5
Loading