bec/tomcat_bec
0
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
gac-x02da
2025-06-16 22:20:42 +02:00
parent f49018754e
commit a9b275c0b9
3 changed files with 99 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
tomcat_bec/devices/pco_edge/pcoedgecamera.py
View File

@@ -434,11 +434,16 @@ class PcoEdge5M(PSIDeviceBase, PcoEdgeBase):
"/gpfs/test/test-beamline" # FIXME: This should be from the scan message
)
if "file_prefix" not in scan_args:
scan_args["file_prefix"] = scan_msg.info["file_components"][0].split("/")[-1] + "_"
file_base = scan_msg.info["file_components"][0].split("/")[-1]
file_suffix = scan_msg.info.get("file_suffix") or ""
comps = [file_base, self.name]
if file_suffix:
comps.append(file_suffix)
scan_args["file_prefix"] = "_".join(comps)
self.configure(scan_args)
if scan_msg.scan_type == "step":
num_points = self.frames_per_trigger.get() * scan_msg.num_points # type: ignore
num_points = self.frames_per_trigger.get() * max(scan_msg.num_points, 1) # type: ignore
else:
num_points = self.frames_per_trigger.get()

16
tomcat_bec/scans/__init__.py
View File

@@ -1,9 +1,9 @@
from .simple_scans import TomoFlyScan, TomoScan
from .simple_scans import TomoFlyScan, TomoScan, AcquireDark
from .tomcat_scans import TomcatSimpleSequence, TomcatSnapNStep
from .tutorial_fly_scan import (
AcquireDark,
AcquireProjections,
AcquireRefs,
AcquireWhite,
TutorialFlyScanContLine,
)
# from .tutorial_fly_scan import (
# # AcquireDark,
# AcquireProjections,
# AcquireRefs,
# AcquireWhite,
# TutorialFlyScanContLine,
# )

113
tomcat_bec/scans/simple_scans.py
View File

@@ -62,6 +62,21 @@ class TomoComponents:
frames_per_trigger=frames_per_trigger,
)
def scan_report_instructions(self):
"""
Generate scan report instructions for the dark image acquisition.
This method provides the necessary instructions to listen to the camera progress during the scan.
"""
if not self.components.cameras:
yield from super().scan_report_instructions()
return
# Use the first camera or "gfcam" if available for reporting
report_camera = (
"gfcam" if "gfcam" in self.components.cameras else self.components.cameras[0]
)
yield from self.stubs.scan_report_instruction({"device_progress": [report_camera]})
def complete(self):
for cam in self.cameras:
yield from self.stubs.send_rpc_and_wait(device=cam, func_name="on_complete")
@@ -88,7 +103,7 @@ class TomoComponents:
device=cam, func_name="update_live_processing_reference", reference_type=ref_type
)
def acquire_dark(self, num_images: int, exposure_time: float, name="dark"):
def acquire_dark(self, num_images: int, exposure_time: float, name="dark", restart=True):
"""
Acquire dark images.
@@ -101,20 +116,22 @@ class TomoComponents:
logger.info(f"Acquiring {num_images} dark images with exposure time {exposure_time}s.")
self.connector.send_client_info(f"Acquiring {num_images} dark images.")
yield from self.restart_cameras(
name=name, prefix=name, num_images=num_images, frames_per_trigger=num_images
)
if restart:
yield from self.restart_cameras(
name=name, prefix=name, num_images=num_images, frames_per_trigger=num_images
)
# yield from self.close_shutter()
yield from self.stubs.trigger(min_wait=exposure_time)
yield from self.complete()
yield from self.update_live_processing_references(ref_type="dark")
yield from self.restore_configs(name=name)
if restart:
yield from self.restore_configs(name=name)
# yield from self.open_shutter()
self.connector.send_client_info("")
logger.info("Dark image acquisition complete.")
def acquire_flat(self, num_images: int, exposure_time: float, name="flat"):
def acquire_flat(self, num_images: int, exposure_time: float, name="flat", restart=True):
"""
Acquire flat images.
@@ -127,20 +144,23 @@ class TomoComponents:
logger.info(f"Acquiring {num_images} flat images with exposure time {exposure_time}s.")
self.connector.send_client_info(f"Acquiring {num_images} flat images.")
yield from self.restart_cameras(
name=name, prefix=name, num_images=num_images, frames_per_trigger=num_images
)
if restart:
yield from self.restart_cameras(
name=name, prefix=name, num_images=num_images, frames_per_trigger=num_images
)
# yield from self.open_shutter()
yield from self.stubs.trigger(min_wait=exposure_time)
yield from self.complete()
yield from self.update_live_processing_references(ref_type="flat")
yield from self.restore_configs(name=name)
if restart:
yield from self.restore_configs(name=name)
logger.info("Flat image acquisition complete.")
self.connector.send_client_info("")
def acquire_references(self, num_darks: int, num_flats: int, exp_time: float, name: str):
yield from self.acquire_dark(num_darks, exposure_time=exp_time, name=name)
yield from self.acquire_flat(num_flats, exposure_time=exp_time, name=name)
def acquire_references(self, num_darks: int, num_flats: int, exp_time: float, restart=True):
yield from self.acquire_dark(num_darks, exposure_time=exp_time, restart=restart)
yield from self.acquire_flat(num_flats, exposure_time=exp_time, restart=restart)
class AcquireDark(ScanBase):
@@ -152,8 +172,7 @@ class AcquireDark(ScanBase):
Args:
num_images (int): Number of dark images to acquire.
exposure_time (float): Exposure time for each dark image in seconds.
name (str): Name for the dark image acquisition. Default: "dark"
exp_time (float): Exposure time for each dark image in seconds.
Returns:
ScanReport
@@ -162,23 +181,59 @@ class AcquireDark(ScanBase):
self.components = TomoComponents(self)
def scan_report_instructions(self):
"""
Generate scan report instructions for the dark image acquisition.
This method provides the necessary instructions to listen to the camera progress during the scan.
"""
if not self.components.cameras:
yield from super().scan_report_instructions()
return
# Use the first camera or "gfcam" if available for reporting
report_camera = (
"gfcam" if "gfcam" in self.components.cameras else self.components.cameras[0]
)
yield from self.stubs.scan_report_instruction({"device_progress": [report_camera]})
yield from self.components.scan_report_instructions()
def scan_core(self):
yield from self.components.acquire_dark(self.frames_per_trigger, self.exp_time, name="dark")
yield from self.components.acquire_dark(self.frames_per_trigger, self.exp_time, restart=False)
class AcquireFlat(ScanBase):
scan_name = "acquire_flat"
def __init__(self, exp_time: float, frames_per_trigger: int = 1, **kwargs):
"""
Acquire flat images.
Args:
num_images (int): Number of flat images to acquire.
exp_time (float): Exposure time for each flat image in seconds.
Returns:
ScanReport
"""
super().__init__(frames_per_trigger=frames_per_trigger, exp_time=exp_time, **kwargs)
self.components = TomoComponents(self)
def scan_report_instructions(self):
yield from self.components.scan_report_instructions()
def scan_core(self):
yield from self.components.acquire_flat(self.frames_per_trigger, self.exp_time, restart=False)
class AcquireReferences(ScanBase):
scan_name = "acquire_refs"
def __init__(self, num_darks:int, num_flats:int, exp_time: float, frames_per_trigger: int = 1, **kwargs):
"""
Acquire flats and darks.
Args:
num_darks (int): Number of dark images to acquire.
num_flats (int): Number of flat images to acquire.
exp_time (float): Exposure time for each flat image in seconds.
Returns:
ScanReport
"""
super().__init__(frames_per_trigger=frames_per_trigger, exp_time=exp_time, **kwargs)
self.num_darks = num_darks
self.num_flats = num_flats
self.components = TomoComponents(self)
def scan_report_instructions(self):
yield from self.components.scan_report_instructions()
def scan_core(self):
yield from self.components.acquire_references(self.num_darks, self.num_flats, self.frames_per_trigger, self.exp_time, restart=False)
class TomoScan(LineScan):
scan_name = "tomo_line_scan"