Coverage for src/pytribeam/workflow.py: 68%

188 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2026-09-03 19:02 +0000

1#!/usr/bin/python3 

2"""High-level experiment workflow orchestration. 

3 

4This module coordinates complete `pytribeam` serial-sectioning experiments from 

5a YAML configuration file. It performs pre-flight validation, initializes the 

6microscope and experiment settings, prepares the log file, retracts insertable 

7devices, moves the stage to each step's slice-dependent position, dispatches the 

8configured operation, records position metadata, and runs the main slice/step 

9experiment loop. 

10 

11Most users should enter this module through `run_experiment_cli`, which is used 

12by the command-line interface. Programmatic workflows may call 

13`setup_experiment` and `perform_step` directly when more control over execution 

14is needed. 

15 

16## Typical usage 

17 

18Run an experiment from a YAML configuration file: 

19 

20```python 

21from pathlib import Path 

22from pytribeam import workflow 

23 

24workflow.run_experiment_cli( 

25 start_slice=1, 

26 start_step=1, 

27 yml_path=Path("experiment.yml"), 

28) 

29``` 

30 

31Set up an experiment and manually perform a step: 

32 

33```python 

34from pathlib import Path 

35from pytribeam import workflow 

36 

37experiment = workflow.setup_experiment(Path("experiment.yml")) 

38 

39workflow.perform_step( 

40 slice_number=1, 

41 step_number=1, 

42 experiment_settings=experiment, 

43) 

44``` 

45 

46## Main entry points 

47 

48- `run_experiment_cli`: run the full command-line experiment loop. 

49- `setup_experiment`: validate configuration, create the log file, link the 

50 stage, and retract insertable devices. 

51- `pre_flight_check`: parse and validate the YAML file, connect to hardware, and 

52 construct `tbt.ExperimentSettings`. 

53- `perform_step`: execute one configured step for one slice. 

54- `perform_operation`: dispatch operation execution based on the step settings 

55 type. 

56- `ebsd_eds_conflict_free`: validate EBSD/EDS step compatibility constraints. 

57 

58## Workflow sequence 

59 

60The command-line workflow proceeds as follows: 

61 

621. Read the YAML configuration file. 

632. Determine the supported YAML schema version. 

643. Parse and validate general experiment settings. 

654. Enable or validate EBSD/EDS control when configured. 

665. Connect to the microscope. 

676. Parse and validate the configured step sequence. 

687. Check for unsupported EBSD/EDS step conflicts. 

698. Create the experiment log file. 

709. Link the stage to free working distance. 

7110. Retract all available and enabled insertable devices. 

7211. Log the experiment configuration. 

7312. Iterate over slices and steps. 

7413. For each executed step: 

75 - log the pre-operation stage position, 

76 - retract insertable devices, 

77 - move to the step start position, 

78 - perform the step operation, 

79 - log the post-operation stage position, 

80 - retract insertable devices again. 

8114. Disconnect from the microscope when the experiment completes. 

82 

83## Operation dispatch 

84 

85`perform_operation` is implemented with `functools.singledispatch`. Dispatch is 

86based on the type of `step.operation_settings`. 

87 

88| Settings type | Operation | 

89| --- | --- | 

90| `tbt.ImageSettings` | Acquire an image using `pytribeam.image.image_operation`. | 

91| `tbt.FIBSettings` | Acquire an associated ion image, then run FIB milling. | 

92| `tbt.LaserSettings` | Run a laser milling operation. | 

93| `tbt.EBSDSettings` | Insert EBSD, optionally insert EDS, measure current, image, and map EBSD. | 

94| `tbt.EDSSettings` | Insert EDS, measure current, image, and map EDS. | 

95| `tbt.CustomSettings` | Execute a user-specified external script. | 

96 

97Unsupported operation settings types raise `NotImplementedError`. 

98 

99## Slice and step numbering 

100 

101Slice and step numbers are user-facing and one-indexed. Step sequences are stored 

102internally as Python lists, so `perform_step` converts the requested 

103`step_number` to a zero-indexed list index. 

104 

105Step frequency is evaluated relative to slice 1. A step with frequency `n` runs 

106on slice 1 and then every `n` slices thereafter. 

107 

108## Logging 

109 

110The workflow logs: 

111 

112- experiment settings at the beginning of the run, 

113- stage position before each executed step, 

114- stage position after each executed step, 

115- specimen current for EBSD and EDS operations, 

116- laser power before and after laser operations. 

117 

118Dataset names and HDF5 dtypes are defined in `pytribeam.constants.Constants`. 

119 

120## Safety behavior 

121 

122Before and after each step, all available and enabled insertable devices are 

123retracted. Stage movement is delegated to `pytribeam.stage`, which checks stage 

124limits and verifies final position. Detector insertion, EBSD/EDS control, laser 

125patterning, FIB milling, and imaging are delegated to their subsystem modules. 

126 

127> **Warning** 

128> 

129> Functions in this module orchestrate microscope motion, detector insertion, 

130> FIB milling, laser firing, EBSD/EDS mapping, and external script execution. 

131> Confirm that the configuration file, microscope state, sample geometry, 

132> detector positions, laser state, and stage limits are safe before running an 

133> experiment. 

134 

135<hr style="height: 12px; background-color: #333; border: none;"> 

136""" 

137 

138__all__ = [ 

139 "perform_operation", 

140 "ebsd_eds_conflict_free", 

141 "pre_flight_check", 

142 "setup_experiment", 

143 "perform_step", 

144 "run_experiment_cli", 

145] 

146 

147# Default python modules 

148# from functools import singledispatch 

149from pathlib import Path 

150import sys 

151from typing import List 

152from functools import singledispatch 

153import subprocess 

154 

155# 3rd party module 

156 

157# Local scripts 

158import pytribeam.constants as cs 

159import pytribeam.insertable_devices as devices 

160import pytribeam.factory as factory 

161import pytribeam.types as tbt 

162import pytribeam.utilities as ut 

163import pytribeam.stage as stage 

164import pytribeam.log as log 

165import pytribeam.laser as laser 

166import pytribeam.image as img 

167import pytribeam.fib as fib 

168 

169 

170@singledispatch 

171def perform_operation( 

172 step_settings, 

173 step: tbt.Step, 

174 general_settings: tbt.GeneralSettings, 

175 slice_number: int, 

176) -> bool: 

177 """ 

178 Perform the operation for the specified step settings. 

179 

180 This function performs the operation for the specified step settings, including validation. 

181 

182 ## Parameters 

183 

184 - `step_settings` (`Any`): The step settings for the operation. 

185 - `step` (`tbt.Step`): The step object containing the operation settings. 

186 - `general_settings` (`tbt.GeneralSettings`): The general settings object. 

187 - `slice_number` (`int`): The slice number for the operation. 

188 

189 ## Returns 

190 

191 - `bool`: True if the operation is performed successfully. 

192 

193 ## Raises 

194 

195 - `NotImplementedError`: If no handler is available for the provided step settings type. 

196 """ 

197 _ = step_settings 

198 __ = step 

199 ___ = general_settings 

200 ____ = slice_number 

201 raise NotImplementedError(f"No handler for type {type(step_settings)}") 

202 

203 

204@perform_operation.register 

205def _perform_image_operation( 

206 step_settings: tbt.ImageSettings, 

207 step: tbt.Step, 

208 general_settings: tbt.GeneralSettings, 

209 slice_number: int, 

210) -> bool: 

211 """ 

212 Perform the image operation for the specified step settings. 

213 

214 ## Parameters 

215 

216 - `step_settings` (`tbt.ImageSettings`): The image settings for the operation. 

217 - `step` (`tbt.Step`): The step object containing the operation settings. 

218 - `general_settings` (`tbt.GeneralSettings`): The general settings object. 

219 - `slice_number` (`int`): The slice number for the operation. 

220 

221 ## Returns 

222 

223 - `bool`: True if the image operation is performed successfully. 

224 """ 

225 return img.image_operation( 

226 step=step, 

227 image_settings=step.operation_settings, 

228 general_settings=general_settings, 

229 slice_number=slice_number, 

230 ) 

231 

232 

233@perform_operation.register 

234def _perform_fib_operation( 

235 step_settings: tbt.FIBSettings, 

236 step: tbt.Step, 

237 general_settings: tbt.GeneralSettings, 

238 slice_number: int, 

239) -> bool: 

240 """ 

241 Perform the FIB operation for the specified step settings. 

242 

243 ## Parameters 

244 

245 - `step_settings` (`tbt.FIBSettings`): The FIB settings for the operation. 

246 - `step` (`tbt.Step`): The step object containing the operation settings. 

247 - `general_settings` (`tbt.GeneralSettings`): The general settings object. 

248 - `slice_number` (`int`): The slice number for the operation. 

249 

250 ## Returns 

251 

252 - `bool`: True if the FIB operation is performed successfully. 

253 """ 

254 # collect image 

255 image_step = tbt.Step( 

256 type=tbt.StepType.IMAGE, 

257 name=step.name, 

258 number=step.number, 

259 frequency=step.frequency, 

260 stage=step.stage, 

261 operation_settings=step_settings.image, 

262 ) 

263 type(image_step.operation_settings) 

264 perform_operation( 

265 image_step.operation_settings, 

266 step=image_step, 

267 general_settings=general_settings, 

268 slice_number=slice_number, 

269 ) 

270 # mill pattern 

271 fib.mill_operation( 

272 step=step, 

273 fib_settings=step_settings, 

274 general_settings=general_settings, 

275 slice_number=slice_number, 

276 ) 

277 

278 return True 

279 

280 

281@perform_operation.register 

282def _perform_custom_operation( 

283 step_settings: tbt.CustomSettings, 

284 step: tbt.Step, 

285 general_settings: tbt.GeneralSettings, 

286 slice_number: int, 

287) -> bool: 

288 """ 

289 Perform the custom operation for the specified step settings. 

290 

291 ## Parameters 

292 

293 - `step_settings` (`tbt.CustomSettings`): The custom settings for the operation. 

294 - `step` (`tbt.Step`): The step object containing the operation settings. 

295 - `general_settings` (`tbt.GeneralSettings`): The general settings object. 

296 - `slice_number` (`int`): The slice number for the operation. 

297 

298 ## Returns 

299 

300 - `bool`: True if the custom operation is performed successfully. 

301 """ 

302 # dump out .yml with experiment info 

303 slice_info_path = Path.joinpath(general_settings.exp_dir, "slice_info.yml") 

304 db = {"exp_dir": str(general_settings.exp_dir), "slice_number": slice_number} 

305 ut.dict_to_yml(db=db, file_path=slice_info_path) 

306 

307 output = subprocess.run( 

308 [step_settings.executable_path, step_settings.script_path], 

309 capture_output=True, 

310 ) 

311 stdout, stderr = output.stdout.decode("utf-8"), output.stderr.decode("utf-8") 

312 if stdout: 312 ↛ 315line 312 didn't jump to line 315 because the condition on line 312 was always true

313 print(f"\nCustom script output: {stdout}\n") 

314 

315 if output.returncode != 0: 315 ↛ 316line 315 didn't jump to line 316 because the condition on line 315 was never true

316 if stderr: 

317 print(f"\nCustom script errors: {stderr}\n") 

318 raise ValueError( 

319 f"Subprocess call for script {step_settings.script_path} using executable {step_settings.executable_path} did not execute correctly." 

320 ) 

321 

322 slice_info_path.unlink() 

323 return True 

324 

325 

326@perform_operation.register 

327def _perform_ebsd_operation( 

328 step_settings: tbt.EBSDSettings, 

329 step: tbt.Step, 

330 general_settings: tbt.GeneralSettings, 

331 slice_number: int, 

332) -> bool: 

333 """ 

334 Perform the EBSD operation for the specified step settings. 

335 

336 ## Parameters 

337 

338 - `step_settings` (`tbt.EBSDSettings`): The EBSD settings for the operation. 

339 - `step` (`tbt.Step`): The step object containing the operation settings. 

340 - `general_settings` (`tbt.GeneralSettings`): The general settings object. 

341 - `slice_number` (`int`): The slice number for the operation. 

342 

343 ## Returns 

344 

345 - `bool`: True if the EBSD operation is performed successfully. 

346 """ 

347 image_settings = step_settings.image 

348 microscope = image_settings.microscope 

349 

350 # insert detector 

351 devices.insert_EBSD(microscope=microscope) 

352 if step_settings.enable_eds: 

353 devices.insert_EDS(microscope=microscope) 

354 

355 # measure and log specimen current 

356 found_current_na = devices.specimen_current(microscope=microscope) 

357 log.specimen_current( 

358 step_number=step.number, 

359 step_name=step.name, 

360 slice_number=slice_number, 

361 log_filepath=general_settings.log_filepath, 

362 dataset_name=cs.Constants.specimen_current_dataset_name, 

363 specimen_current_na=found_current_na, 

364 ) 

365 

366 # take image 

367 img.image_operation( 

368 step=step, 

369 image_settings=image_settings, 

370 general_settings=general_settings, 

371 slice_number=slice_number, 

372 ) 

373 

374 # set dynamic focus/tilt correction 

375 dynamic_focus = image_settings.beam.settings.dynamic_focus 

376 tilt_correction = image_settings.beam.settings.tilt_correction 

377 img.beam_angular_correction( 

378 microscope=microscope, 

379 dynamic_focus=dynamic_focus, 

380 tilt_correction=tilt_correction, 

381 ) 

382 

383 # take map 

384 laser.map_ebsd() 

385 

386 # retract detector(s) 

387 devices.retract_EBSD(microscope=microscope) 

388 if step_settings.enable_eds: 388 ↛ 389,   388 ↛ 3912 missed branches: 1) line 388 didn't jump to line 389 because the condition on line 388 was never true, 2) line 388 didn't jump to line 391 because the condition on line 388 was always true

389 devices.retract_EDS(microscope=microscope) 

390 

391 return True 

392 

393 

394@perform_operation.register 

395def _perform_eds_operation( 

396 step_settings: tbt.EDSSettings, 

397 step: tbt.Step, 

398 general_settings: tbt.GeneralSettings, 

399 slice_number: int, 

400) -> bool: 

401 """ 

402 Perform the EDS operation for the specified step settings. 

403 

404 ## Parameters 

405 

406 - `step_settings` (`tbt.EDSSettings`): The EDS settings for the operation. 

407 - `step` (`tbt.Step`): The step object containing the operation settings. 

408 - `general_settings` (`tbt.GeneralSettings`): The general settings object. 

409 - `slice_number` (`int`): The slice number for the operation. 

410 

411 ## Returns 

412 

413 - `bool`: True if the EDS operation is performed successfully. 

414 """ 

415 image_settings = step_settings.image 

416 microscope = image_settings.microscope 

417 

418 # insert detector 

419 devices.insert_EDS(microscope=microscope) 

420 

421 # measure and log specimen current 

422 found_current_na = devices.specimen_current(microscope=microscope) 

423 log.specimen_current( 

424 step_number=step.number, 

425 step_name=step.name, 

426 slice_number=slice_number, 

427 log_filepath=general_settings.log_filepath, 

428 dataset_name=cs.Constants.specimen_current_dataset_name, 

429 specimen_current_na=found_current_na, 

430 ) 

431 

432 # take image 

433 img.image_operation( 

434 step=step, 

435 image_settings=image_settings, 

436 general_settings=general_settings, 

437 slice_number=slice_number, 

438 ) 

439 

440 # set dynamic focus/tilt correction 

441 dynamic_focus = image_settings.beam.settings.dynamic_focus 

442 tilt_correction = image_settings.beam.settings.tilt_correction 

443 img.beam_angular_correction( 

444 microscope=microscope, 

445 dynamic_focus=dynamic_focus, 

446 tilt_correction=tilt_correction, 

447 ) 

448 

449 # take map 

450 laser.map_eds() 

451 

452 # retract detector 

453 devices.retract_EDS(microscope=microscope) 

454 

455 return True 

456 

457 

458@perform_operation.register 

459def _perform_laser_operation( 

460 step_settings: tbt.LaserSettings, 

461 step: tbt.Step, 

462 general_settings: tbt.GeneralSettings, 

463 slice_number: int, 

464) -> bool: 

465 """ 

466 Perform the laser operation for the specified step settings. 

467 

468 ## Parameters 

469 

470 - `step_settings` (`tbt.LaserSettings`): The laser settings for the operation. 

471 - `step` (`tbt.Step`): The step object containing the operation settings. 

472 - `general_settings` (`tbt.GeneralSettings`): The general settings object. 

473 - `slice_number` (`int`): The slice number for the operation. 

474 

475 ## Returns 

476 

477 - `bool`: True if the laser operation is performed successfully. 

478 """ 

479 return laser.laser_operation( 

480 step=step, 

481 general_settings=general_settings, 

482 slice_number=slice_number, 

483 ) 

484 

485 

486def ebsd_eds_conflict_free(step_sequence: List[tbt.Step]) -> bool: 

487 """ 

488 Check if the step sequence is free of EBSD and EDS conflicts. 

489 

490 This function checks if the step sequence is free of EBSD and EDS conflicts. 

491 

492 ## Parameters 

493 

494 - `step_sequence` (`List[tbt.Step]`): The step sequence to check. 

495 

496 ## Returns 

497 

498 - `bool`: True if the step sequence is free of EBSD and EDS conflicts. 

499 

500 ## Raises 

501 

502 - `ValueError`: If an EBSD or EDS conflict is found in the step sequence. 

503 """ 

504 EBSD_EDS_conflict_msg = "Due to current limitations in 3rd party EBSD/EDS integration with the TriBeam, only one of these step types is allowed as only one map can be configured for an experiment, but EDS can be configured to be included with an EBSD type step. See User Guide for more details." 

505 

506 found_EBSD = False 

507 found_EDS = False 

508 

509 for step in step_sequence: 

510 if step.type == tbt.StepType.EBSD: 510 ↛ 511line 510 didn't jump to line 511 because the condition on line 510 was never true

511 if found_EDS == True: 

512 raise ValueError( 

513 f"EBSD step found in sequence after EDS step was already defined. {EBSD_EDS_conflict_msg}" 

514 ) 

515 found_EBSD = True 

516 

517 if step.type == tbt.StepType.EDS: 517 ↛ 518line 517 didn't jump to line 518 because the condition on line 517 was never true

518 if found_EBSD == True: 

519 raise ValueError( 

520 f"EDS step found in sequence after EBSD step was already defined. {EBSD_EDS_conflict_msg}" 

521 ) 

522 found_EDS = True 

523 

524 return True 

525 

526 

527def pre_flight_check(yml_path: Path) -> tbt.ExperimentSettings: 

528 """ 

529 Perform a pre-flight check for the experiment. 

530 

531 This function performs a pre-flight check for the experiment by validating the YAML configuration, connecting to the microscope, and validating the step sequence. 

532 

533 ## Parameters 

534 

535 - `yml_path` (`Path`): The path to the YAML configuration file. 

536 

537 ## Returns 

538 

539 - `tbt.ExperimentSettings`: The validated experiment settings. 

540 

541 ## Raises 

542 

543 - `SystemError`: If there are issues with the EBSD or EDS camera, or if the laser control is not enabled. 

544 - `ValueError`: If the step sequence is not parsed correctly or if there are EBSD/EDS conflicts. 

545 """ 

546 # get configuration from yml 

547 yml_version = ut.yml_version(yml_path) 

548 experiment_settings = ut.yml_to_dict( 

549 yml_path_file=yml_path, 

550 version=yml_version, 

551 required_keys=( 

552 "general", 

553 "config_file_version", 

554 ), 

555 ) 

556 yml_format = ut.yml_format(version=yml_version) 

557 

558 # get general settings and validate them 

559 general_db = ut.general_settings( 

560 exp_settings=experiment_settings, yml_format=yml_format 

561 ) 

562 general_settings = factory.general( 

563 general_db=general_db, 

564 yml_format=yml_format, 

565 ) 

566 

567 # whether to enable EBSD and EDS control 

568 enable_EBSD = ut.enable_external_device(general_settings.EBSD_OEM) 

569 enable_EDS = ut.enable_external_device(general_settings.EDS_OEM) 

570 if enable_EBSD: 570 ↛ 571line 570 didn't jump to line 571 because the condition on line 570 was never true

571 status = devices.connect_EBSD() 

572 if status == tbt.RetractableDeviceState.ERROR: 572 ↛ 573,   572 ↛ 5742 missed branches: 1) line 572 didn't jump to line 573 because the condition on line 572 was never true, 2) line 572 didn't jump to line 574 because the condition on line 572 was always true

573 raise SystemError("EBSD camera is connected but in error state.") 

574 if enable_EDS: 574 ↛ 575line 574 didn't jump to line 575 because the condition on line 574 was never true

575 status = devices.connect_EDS() 

576 if status == tbt.RetractableDeviceState.ERROR: 

577 raise SystemError("EDS camera is connected but in error state.") 

578 

579 # connect to microscope: 

580 connection = general_settings.connection 

581 microscope = tbt.Microscope() 

582 ut.connect_microscope( 

583 microscope=microscope, 

584 quiet_output=True, 

585 connection_host=connection.host, 

586 connection_port=connection.port, 

587 ) 

588 

589 # get step_count and validate settings 

590 num_steps = ut.step_count(exp_settings=experiment_settings, yml_format=yml_format) 

591 step_sequence = [] # empty list of tbt.Step type objects 

592 for step in range(1, num_steps + 1): 

593 step_name, step_settings = ut.step_settings( 

594 exp_settings=experiment_settings, 

595 step_number_key=yml_format.step_number_key, 

596 step_number_val=step, 

597 yml_format=yml_format, 

598 ) 

599 if not step_name: 599 ↛ 600line 599 didn't jump to line 600 because the condition on line 599 was never true

600 raise KeyError( 

601 f"Step name for step {step} of {num_steps} is empty. Please provide a unique name for each step in your configuration." 

602 ) 

603 step_type = ut.step_type( 

604 settings=step_settings, 

605 yml_format=yml_format, 

606 ) 

607 

608 # validate connections for specific step types 

609 if step_type == tbt.StepType.LASER: 609 ↛ 610line 609 didn't jump to line 610 because the condition on line 609 was never true

610 laser_enabled = laser.laser_connected() 

611 if not laser_enabled: 611 ↛ 615line 611 didn't jump to line 615 because the condition on line 611 was always true

612 raise SystemError( 

613 f"Step name '{step_name}' is a Laser step type but Laser control is not currently enabled. Ensure TFS laser API is installed, Laser Control application is open." 

614 ) 

615 if (step_type == tbt.StepType.EDS) and (not enable_EDS): 615 ↛ 616line 615 didn't jump to line 616 because the condition on line 615 was never true

616 raise SystemError( 

617 f"Step name '{step_name}' is an EDS step type but EDS control is not currently enabled." 

618 ) 

619 if (step_type == tbt.StepType.EBSD) and (not enable_EBSD): 619 ↛ 620line 619 didn't jump to line 620 because the condition on line 619 was never true

620 raise SystemError( 

621 f"Step name '{step_name}' is an EBSD step type but EDS control is not currently enabled." 

622 ) 

623 # if (step_type == tbt.StepType.EBSD_EDS) and ( 

624 # (not enable_EBSD) or (not enable_EDS) 

625 # ): 

626 # raise SystemError( 

627 # f"Step name '{step_name}' is an EBSD_EDS step type but EBSD and EDS control are not both currently enabled." 

628 # ) 

629 # create the step settings 

630 step = factory.step( 

631 microscope=microscope, 

632 step_name=step_name, 

633 step_settings=step_settings, 

634 general_settings=general_settings, 

635 yml_format=yml_format, 

636 ) 

637 

638 step_sequence.append(step) 

639 

640 if len(step_sequence) != num_steps: 640 ↛ 641line 640 didn't jump to line 641 because the condition on line 640 was never true

641 raise ValueError( 

642 f"Settings not parsed correctly, expected {num_steps} but only {len(step_sequence)} have been parsed." 

643 ) 

644 

645 # ensure only EBSD or EDS step type exists 

646 ebsd_eds_conflict_free(step_sequence=step_sequence) 

647 

648 experiment_settings = tbt.ExperimentSettings( 

649 microscope=microscope, 

650 general_settings=general_settings, 

651 step_sequence=step_sequence, 

652 enable_EBSD=enable_EBSD, 

653 enable_EDS=enable_EDS, 

654 ) 

655 # print("Pre-flight check complete.") 

656 return experiment_settings 

657 

658 

659def setup_experiment( 

660 yml_path: Path, 

661) -> tbt.ExperimentSettings: 

662 """ 

663 Set up the experiment based on the YAML configuration. 

664 

665 This function sets up the experiment by validating the YAML configuration, creating the log file, linking the stage, and retracting all devices. 

666 

667 ## Parameters 

668 

669 - `yml_path` (`Path`): The path to the YAML configuration file. 

670 

671 ## Returns 

672 

673 - `tbt.ExperimentSettings`: The experiment settings. 

674 """ 

675 # validate yml 

676 experiment_settings = pre_flight_check(yml_path=yml_path) 

677 

678 log_filepath = experiment_settings.general_settings.log_filepath 

679 log.create_file(log_filepath) 

680 

681 # link stage to free working distance 

682 experiment_settings.microscope.specimen.stage.link() 

683 

684 # retract all devices 

685 print("\tRetracting all devices...") 

686 devices.retract_all_devices( 

687 microscope=experiment_settings.microscope, 

688 enable_EBSD=experiment_settings.enable_EBSD, 

689 enable_EDS=experiment_settings.enable_EDS, 

690 ) 

691 

692 return experiment_settings 

693 

694 

695def perform_step( 

696 slice_number: int, 

697 step_number: int, 

698 experiment_settings: tbt.ExperimentSettings, 

699) -> None: 

700 """ 

701 Perform a step in the experiment. 

702 

703 This function performs a step in the experiment based on the slice number, step number, and experiment settings. 

704 

705 ## Parameters 

706 

707 - `slice_number` (`int`): The slice number for the step. 

708 - `step_number` (`int`): The step number for the experiment. 

709 - `experiment_settings` (`tbt.ExperimentSettings`): The experiment settings. 

710 

711 ## Returns 

712 

713 - `bool`: True if the step is performed successfully. 

714 """ 

715 # # breakout experiment settings elements 

716 microscope = experiment_settings.microscope 

717 general_settings = experiment_settings.general_settings 

718 step_sequence = experiment_settings.step_sequence 

719 enable_EBSD = experiment_settings.enable_EBSD 

720 enable_EDS = experiment_settings.enable_EDS 

721 

722 # get operation settings, execute operation. 

723 operation = step_sequence[step_number - 1] # list is 0-indexed 

724 print( 

725 f"Slice {slice_number}, Step {step_number} of {general_settings.step_count}, '{operation.name}', a {operation.type.value} type step." 

726 ) 

727 # slices start at 1, perform all steps on slice 1. 

728 if (slice_number - 1) % operation.frequency != 0: 728 ↛ 729line 728 didn't jump to line 729 because the condition on line 728 was never true

729 print( 

730 f"\tStep frequency is every {operation.frequency} slices, starting on slice 1. Skipping step on this slice.\n" 

731 ) 

732 return 

733 

734 # log step_start position 

735 log.position( 

736 step_number=step_number, 

737 step_name=operation.name, 

738 slice_number=slice_number, 

739 log_filepath=general_settings.log_filepath, 

740 dataset_name=cs.Constants.pre_position_dataset_name, 

741 current_position=factory.active_stage_position_settings( 

742 microscope=microscope, 

743 ), 

744 ) 

745 

746 # retract all devices 

747 print("\tRetracting all devices...") 

748 # with ut.nostdout(): 

749 devices.retract_all_devices( 

750 microscope=microscope, 

751 enable_EBSD=enable_EBSD, 

752 enable_EDS=enable_EDS, 

753 ) 

754 print("\tDevices retracted.") 

755 

756 # move stage to starting position for slice 

757 stage.step_start_position( 

758 microscope=microscope, 

759 slice_number=slice_number, 

760 operation=operation, 

761 general_settings=general_settings, 

762 ) 

763 

764 # perform specific operation 

765 perform_operation( 

766 operation.operation_settings, 

767 step=operation, 

768 general_settings=general_settings, 

769 slice_number=slice_number, 

770 ) 

771 

772 # log step end position 

773 log.position( 

774 step_number=step_number, 

775 step_name=operation.name, 

776 slice_number=slice_number, 

777 log_filepath=general_settings.log_filepath, 

778 dataset_name=cs.Constants.post_position_dataset_name, 

779 current_position=factory.active_stage_position_settings( 

780 microscope=microscope, 

781 ), 

782 ) 

783 

784 # retract all devices 

785 print("\tRetracting all devices...") 

786 # with ut.nostdout(): 

787 devices.retract_all_devices( 

788 microscope=microscope, 

789 enable_EBSD=enable_EBSD, 

790 enable_EDS=enable_EDS, 

791 ) 

792 print("\tDevices retracted. Step Complete.\n") 

793 

794 

795def run_experiment_cli( 

796 start_slice: int, 

797 start_step: int, 

798 yml_path: Path, 

799) -> None: 

800 """ 

801 Main loop for the experiment, accessed through the command line. 

802 

803 This function runs the main loop for the experiment based on the specified start slice, start step, and YAML configuration file. 

804 

805 ## Parameters 

806 

807 - `start_slice` (`int`): The starting slice number for the experiment. 

808 - `start_step` (`int`): The starting step number for the experiment. 

809 - `yml_path` (`Path`): The path to the YAML configuration file. 

810 

811 ## Returns 

812 

813 - `None` 

814 """ 

815 

816 experiment_settings = setup_experiment(yml_path=yml_path) 

817 

818 # warn user of any EBSD/EDS lack of control 

819 warning_text = """is not enabled, you will not have access to safety 

820 checking and these modalities during data collection. Please ensure  

821 this detector is retracted before proceeding.""" 

822 if not experiment_settings.enable_EBSD: 822 ↛ 827line 822 didn't jump to line 827 because the condition on line 822 was always true

823 print(f"\nWARNING: EBSD {warning_text}") 

824 if not ut.yes_no("Continue?"): 824 ↛ 825line 824 didn't jump to line 825 because the condition on line 824 was never true

825 print("\nExiting now...") 

826 sys.exit() 

827 if not experiment_settings.enable_EDS: 827 ↛ 834line 827 didn't jump to line 834 because the condition on line 827 was always true

828 print(f"\nWARNING: EDS {warning_text}") 

829 if not ut.yes_no("Continue?"): 829 ↛ 830line 829 didn't jump to line 830 because the condition on line 829 was never true

830 print("\nExiting now...") 

831 sys.exit() 

832 

833 # main loop 

834 log.experiment_settings( 

835 slice_number=start_slice, 

836 step_number=start_step, 

837 log_filepath=experiment_settings.general_settings.log_filepath, 

838 yml_path=yml_path, 

839 ) 

840 num_steps = len(experiment_settings.step_sequence) 

841 print( 

842 f"\n\nBeginning serial sectioning experiment on slice {start_slice}, step {start_step} of {num_steps}.\n" 

843 ) 

844 

845 for slice_number in range( 

846 start_slice, experiment_settings.general_settings.max_slice_number + 1 

847 ): # inclusive of max slice number 

848 for step_number in range(start_step, num_steps + 1): # list is 1-indexed 

849 perform_step( 

850 slice_number=slice_number, 

851 step_number=step_number, 

852 experiment_settings=experiment_settings, 

853 ) 

854 

855 # reset start_step to 1 at end of slice 

856 start_step = 1 

857 

858 ut.disconnect_microscope( 

859 microscope=experiment_settings.microscope, 

860 quiet_output=True, 

861 ) 

862 

863 print("\n\nExperiment complete.")