Coverage for src/pytribeam/utilities.py: 73%

239 statements  

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

1#!/usr/bin/python3 

2"""General-purpose utility functions for `pytribeam`. 

3 

4This module contains shared helper functions used throughout the package. The 

5utilities here support microscope connection management, beam-object dispatch, 

6YAML configuration parsing, nested-dictionary inspection, interval checks, 

7console-output suppression, simple user prompts, list formatting, filesystem 

8cleanup, and test-environment detection. 

9 

10Functions in this module are intentionally lightweight and generally avoid 

11owning workflow-specific logic. Higher-level modules use these helpers to keep 

12configuration parsing, validation, formatting, and microscope connection code 

13consistent across the package. 

14 

15## Utility categories 

16 

17| Category | Functions | 

18| --- | --- | 

19| Microscope connection | `connect_microscope`, `disconnect_microscope`, `valid_microscope_connection` | 

20| Beam dispatch | `beam_type` | 

21| YAML configuration | `yml_version`, `yml_format`, `yml_to_dict`, `dict_to_yml`, `general_settings`, `step_count`, `step_settings`, `step_type` | 

22| Dictionary helpers | `gen_dict_extract`, `nested_dictionary_location`, `nested_find_key_value_pair`, `none_value_dictionary` | 

23| Validation helpers | `in_interval`, `valid_enum_entry`, `enable_external_device` | 

24| Console helpers | `nostdout`, `yes_no`, `tabular_list`, `split_list` | 

25| Filesystem helpers | `remove_directory` | 

26| Test/environment helpers | `get_test_description`, `get_autoscript_version`, `is_laser_available` | 

27 

28## Typical usage 

29 

30Read and validate a YAML configuration file: 

31 

32```python 

33from pathlib import Path 

34 

35from pytribeam import utilities as ut 

36 

37version = ut.yml_version(Path("experiment.yml")) 

38fmt = ut.yml_format(version) 

39 

40settings = ut.yml_to_dict( 

41 yml_path_file=Path("experiment.yml"), 

42 version=version, 

43 required_keys=fmt.required_keys, 

44) 

45``` 

46 

47Check whether a value lies inside a closed interval: 

48 

49```python 

50import pytribeam.types as tbt 

51from pytribeam import utilities as ut 

52 

53is_valid = ut.in_interval( 

54 val=5.0, 

55 limit=tbt.Limit(min=0.0, max=10.0), 

56 type=tbt.IntervalType.CLOSED, 

57) 

58``` 

59 

60Connect to a microscope while suppressing connection output: 

61 

62```python 

63import pytribeam.types as tbt 

64from pytribeam import utilities as ut 

65 

66microscope = tbt.Microscope() 

67 

68ut.connect_microscope( 

69 microscope=microscope, 

70 quiet_output=True, 

71) 

72 

73# Use microscope... 

74 

75ut.disconnect_microscope(microscope) 

76``` 

77 

78Format a long list for display: 

79 

80```python 

81from pytribeam import utilities as ut 

82 

83print(ut.tabular_list(["CBS", "ETD", "TLD", "ABS"])) 

84``` 

85 

86## YAML configuration helpers 

87 

88The YAML helpers assume that experiment configuration files include a 

89`config_file_version` field and the package-specific top-level keys expected by 

90the selected `tbt.YMLFormatVersion`. 

91 

92`step_count` validates that the number of discovered steps matches the declared 

93step count in the general configuration section. `step_settings` retrieves the 

94settings dictionary for a specific step number and returns both the user-defined 

95step name and the step settings. 

96 

97## Beam dispatch 

98 

99`beam_type` is implemented with `functools.singledispatch` and maps 

100`pytribeam` beam wrapper types to the corresponding microscope beam property: 

101 

102| Input type | Returned microscope object | 

103| --- | --- | 

104| `tbt.ElectronBeam` | `microscope.beams.electron_beam` | 

105| `tbt.IonBeam` | `microscope.beams.ion_beam` | 

106 

107Unsupported beam types raise `NotImplementedError`. 

108 

109## Test and environment helpers 

110 

111The test helper functions detect whether the current system appears to be an 

112offline/simulated machine, microscope hardware machine, or laser-capable 

113hardware machine. These utilities are primarily intended for test selection, 

114test naming, and CI/local test reporting. 

115 

116## Notes 

117 

118This module is broad by design, but utilities that become tightly coupled to a 

119specific subsystem may be better placed in that subsystem's module over time. 

120 

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

122""" 

123 

124__all__ = [ 

125 "beam_type", 

126 "connect_microscope", 

127 "dict_to_yml", 

128 "disconnect_microscope", 

129 "general_settings", 

130 "step_type", 

131 "in_interval", 

132 "gen_dict_extract", 

133 "nested_dictionary_location", 

134 "nested_find_key_value_pair", 

135 "none_value_dictionary", 

136 "nostdout", 

137 "step_count", 

138 "step_settings", 

139 "valid_microscope_connection", 

140 "enable_external_device", 

141 "valid_enum_entry", 

142 "yml_format", 

143 "yml_to_dict", 

144 "yml_version", 

145 "yes_no", 

146 "remove_directory", 

147 "split_list", 

148 "tabular_list", 

149 "get_test_description", 

150 "get_autoscript_version", 

151 "is_laser_available", 

152 "application_files", 

153] 

154 

155 

156# Default python modules 

157from pathlib import Path 

158from typing import Dict, Tuple, Any, List, Optional 

159from enum import Enum 

160import platform 

161from functools import singledispatch 

162import shutil 

163 

164# # Autoscript modules 

165import yaml 

166import contextlib 

167import sys 

168from pandas import json_normalize 

169 

170# # # 3rd party module 

171# from schema import Schema, And, Use, Optional, SchemaError 

172 

173# # Local scripts 

174import pytribeam.types as tbt 

175 

176# import pytribeam.constants as cs 

177from pytribeam.constants import Constants 

178 

179 

180@singledispatch 

181def beam_type(beam: Any, microscope: tbt.Microscope) -> property: 

182 """ 

183 Return the beam property object as ion and electron beams have the same internal hierarchy. 

184 

185 ## Parameters 

186 

187 - `beam` (`Any`): The beam object. 

188 

189 ## Returns 

190 

191 - `property`: The beam property object. 

192 

193 ## Raises 

194 

195 - `NotImplementedError`: If the beam type is not implemented. 

196 """ 

197 _ = beam # no operation 

198 raise NotImplementedError() 

199 

200 

201@beam_type.register 

202def _electron_beam_type(beam: tbt.ElectronBeam, microscope: tbt.Microscope) -> property: 

203 """ 

204 Return the electron beam property object. 

205 

206 ## Parameters 

207 

208 - `beam` (`tbt.ElectronBeam`): The electron beam object. 

209 - `microscope` (`tbt.Microscope`): The microscope object. 

210 

211 ## Returns 

212 

213 - `property`: The electron beam property object. 

214 """ 

215 return microscope.beams.electron_beam 

216 

217 

218@beam_type.register 

219def _ion_beam_type(beam: tbt.IonBeam, microscope: tbt.Microscope) -> property: 

220 """ 

221 Return the ion beam property object. 

222 

223 ## Parameters 

224 

225 - `beam` (`tbt.IonBeam`): The ion beam object. 

226 - `microscope` (`tbt.Microscope`): The microscope object. 

227 

228 ## Returns 

229 

230 - `property`: The ion beam property object. 

231 """ 

232 return microscope.beams.ion_beam 

233 

234 

235def connect_microscope( 

236 microscope: tbt.Microscope, 

237 quiet_output: bool = True, 

238 connection_host: str = None, 

239 connection_port: int = None, 

240) -> bool: 

241 """ 

242 Connect to the microscope with the option to suppress printout. 

243 

244 ## Parameters 

245 

246 - `microscope` (`tbt.Microscope`): The microscope object to connect. 

247 - `quiet_output` (`bool, optional`): Whether to suppress printout (default is True). 

248 - `connection_host` (`str, optional`): The connection host (default is None). 

249 - `connection_port` (`int, optional`): The connection port (default is None). 

250 

251 ## Returns 

252 

253 - `bool`: True if the connection is successful. 

254 

255 ## Raises 

256 

257 - `ConnectionError`: If the connection fails. 

258 """ 

259 

260 # TODO clean up inner function 

261 def connect( 

262 microscope: tbt.Microscope, 

263 connection_host: str = None, 

264 connection_port: int = None, 

265 ) -> bool: 

266 if connection_port is not None: 

267 microscope.connect(connection_host, connection_port) 

268 elif connection_host is not None: 268 ↛ 272line 268 didn't jump to line 272 because the condition on line 268 was always true

269 microscope.connect(connection_host) 

270 

271 else: 

272 microscope.connect() 

273 

274 if quiet_output: 

275 with nostdout(): 

276 connect( 

277 microscope=microscope, 

278 connection_host=connection_host, 

279 connection_port=connection_port, 

280 ) 

281 else: 

282 connect( 

283 microscope=microscope, 

284 connection_host=connection_host, 

285 connection_port=connection_port, 

286 ) 

287 

288 if microscope.server_host is not None: 288 ↛ 291line 288 didn't jump to line 291 because the condition on line 288 was always true

289 return True 

290 else: 

291 raise ConnectionError( 

292 f"Connection failed with connection_host of '{connection_host}' and connection_port of '{connection_port}' microscope not connected." 

293 ) 

294 

295 

296def dict_to_yml(db: dict, file_path: Path) -> Path: 

297 """ 

298 Convert a dictionary to a YAML file. 

299 

300 ## Parameters 

301 

302 - `db` (`dict`): The dictionary to convert. 

303 - `file_path` (`Path`): The path to save the YAML file. 

304 

305 ## Returns 

306 

307 - `Path`: The path to the saved YAML file. 

308 """ 

309 with open(file_path, "w", encoding="utf-8") as out_file: 

310 yaml.dump( 

311 db, 

312 out_file, 

313 default_flow_style=False, 

314 sort_keys=False, 

315 ) 

316 

317 return file_path 

318 

319 

320def disconnect_microscope( 

321 microscope: tbt.Microscope, 

322 quiet_output: bool = True, 

323) -> bool: 

324 """ 

325 Disconnect from the microscope with the option to suppress printout. 

326 

327 ## Parameters 

328 

329 - `microscope` (`tbt.Microscope`): The microscope object to disconnect. 

330 - `quiet_output` (`bool, optional`): Whether to suppress printout (default is True). 

331 

332 ## Returns 

333 

334 - `bool`: True if the disconnection is successful. 

335 

336 ## Raises 

337 

338 - `ConnectionError`: If the disconnection fails. 

339 """ 

340 if quiet_output: 

341 with nostdout(): 

342 microscope.disconnect() 

343 else: 

344 microscope.disconnect() 

345 

346 if microscope.server_host is None: 346 ↛ 349line 346 didn't jump to line 349 because the condition on line 346 was always true

347 return True 

348 else: 

349 raise ConnectionError("Disconnection failed, microscope still connected") 

350 

351 

352def general_settings(exp_settings: dict, yml_format: tbt.YMLFormat) -> dict: 

353 """ 

354 Grab general experiment settings from a .yml file and return them as a dictionary. 

355 

356 ## Parameters 

357 

358 - `exp_settings` (`dict`): The experiment settings dictionary. 

359 - `yml_format` (`tbt.YMLFormat`): The YAML format version. 

360 

361 ## Returns 

362 

363 - `dict`: The general experiment settings as a dictionary. 

364 """ 

365 general_key = yml_format.general_section_key 

366 return exp_settings[general_key] 

367 

368 

369def step_type(settings: dict, yml_format: tbt.YMLFormat) -> tbt.StepType: 

370 """ 

371 Determine the step type for a specific step settings dictionary. 

372 

373 ## Parameters 

374 

375 - `settings` (`dict`): The step settings dictionary. 

376 - `yml_format` (`tbt.YMLFormat`): The YAML format version. 

377 

378 ## Returns 

379 

380 - `tbt.StepType`: The step type. 

381 """ 

382 step_type = tbt.StepType( 

383 settings[yml_format.step_general_key][yml_format.step_type_key] 

384 ) 

385 

386 return step_type 

387 

388 

389def in_interval(val: float, limit: tbt.Limit, type: tbt.IntervalType) -> bool: 

390 """ 

391 Test whether a value is within an interval, with the interval type defined by an enumerated IntervalType. 

392 

393 ## Parameters 

394 

395 - `val` (`float`): The input value to be compared against min and max. 

396 - `limit` (`tbt.Limit`): The bounds of the interval. 

397 - `type` (`tbt.IntervalType`): The type of interval. 

398 

399 ## Returns 

400 

401 - `bool`: True if within the interval, False otherwise. 

402 """ 

403 if type == tbt.IntervalType.OPEN: 

404 return (val > limit.min) and (val < limit.max) 

405 if type == tbt.IntervalType.CLOSED: 

406 return (val >= limit.min) and (val <= limit.max) 

407 if type == tbt.IntervalType.LEFT_OPEN: 

408 return (val > limit.min) and (val <= limit.max) 

409 if type == tbt.IntervalType.RIGHT_OPEN: 409 ↛ exitline 409 didn't return from function 'in_interval' because the condition on line 409 was always true

410 return (val >= limit.min) and (val < limit.max) 

411 

412 

413def gen_dict_extract(key, var): 

414 """ 

415 Extract values from a nested dictionary by key. 

416 

417 ## Parameters 

418 

419 - `key` (`str`): The key to search for. 

420 - `var` (`dict`): The nested dictionary to search. 

421 

422 ## Yields 

423 

424 - `Any`: The values associated with the specified key. 

425 """ 

426 if hasattr(var, "items"): 

427 for k, v in var.items(): 

428 if k == key: 

429 yield v 

430 if isinstance(v, dict): 

431 for result in gen_dict_extract(key, v): 

432 yield result 

433 elif isinstance(v, list): 

434 for d in v: 

435 for result in gen_dict_extract(key, d): 

436 yield result 

437 

438 

439def nested_dictionary_location(d: dict, key: str, value: Any) -> List[str]: 

440 """ 

441 Find the nested location of a key-value pair in a dictionary. 

442 

443 This function returns a list of key values from the highest to the lowest level of nested dictionaries. 

444 

445 ## Parameters 

446 

447 - `d` (`dict`): The dictionary to search. 

448 - `key` (`str`): The key to search for. 

449 - `value` (`Any`): The value to search for. 

450 

451 ## Returns 

452 

453 - `List[str]`: The nested location of the key-value pair. 

454 

455 ## Raises 

456 

457 - `KeyError`: If the key-value pair is not found in the dictionary. 

458 """ 

459 nesting = nested_find_key_value_pair(d=d, key=key, value=value) 

460 if nesting is None: 

461 raise KeyError( 

462 f'Key : value pair of "{key} : {value}" not found in the provided dictionary.' 

463 ) 

464 return nesting 

465 

466 

467def nested_find_key_value_pair(d: dict, key: str, value: Any) -> Optional[List[str]]: 

468 """ 

469 Find a key-value pair in a nested dictionary. 

470 

471 This function returns a list of key values from the highest to the lowest level of nested dictionaries. 

472 

473 ## Parameters 

474 

475 - `d` (`dict`): The dictionary to search. 

476 - `key` (`str`): The key to search for. 

477 - `value` (`Any`): The value to search for. 

478 

479 ## Returns 

480 

481 - `List[str]`: The nested location of the key-value pair. 

482 - `None`: None if the key value pair does not exist 

483 """ 

484 for k, v in d.items(): 

485 if k == key: 

486 if v == value: 

487 return [k] 

488 if isinstance(v, dict): 

489 p = nested_find_key_value_pair(v, key, value) 

490 if p: 

491 return [k] + p 

492 

493 

494def _flatten(dictionary: dict) -> dict: 

495 """ 

496 Flatten a dictionary using pandas. 

497 

498 This function flattens a nested dictionary using pandas, which can be slow on large dictionaries. 

499 From https://stackoverflow.com/questions/6027558/flatten-nested-dictionaries-compressing-keys 

500 

501 ## Parameters 

502 

503 - `dictionary` (`dict`): The dictionary to flatten. 

504 

505 ## Returns 

506 

507 - `dict`: The flattened dictionary. 

508 """ 

509 data_frame = json_normalize(dictionary, sep="_") 

510 db_flat = data_frame.to_dict(orient="records")[0] 

511 return db_flat 

512 

513 

514def none_value_dictionary(dictionary: dict) -> bool: 

515 """ 

516 Check if all values in a dictionary are None. 

517 

518 This function returns True if all values in the dictionary are None, and False otherwise. 

519 

520 ## Parameters 

521 

522 - `dictionary` (`dict`): The dictionary to check. 

523 

524 ## Returns 

525 

526 - `bool`: True if all values in the dictionary are None, False otherwise. 

527 """ 

528 # flatten the dictionary first 

529 db_flat = _flatten(dictionary) 

530 return all([v is None for v in db_flat.values()]) 

531 

532 

533@contextlib.contextmanager 

534def nostdout(): 

535 """ 

536 Create a dummy file to suppress output. 

537 

538 This function creates a dummy file to suppress output. 

539 

540 ## Yields 

541 

542 - `None`: 

543 """ 

544 save_stdout = sys.stdout 

545 sys.stdout = tbt.DummyFile() 

546 try: 

547 yield 

548 finally: 

549 # Always restore stdout, even if KeyboardInterrupt or other exceptions occur 

550 sys.stdout = save_stdout 

551 

552 

553def step_count( 

554 exp_settings: dict, 

555 yml_format: tbt.YMLFormatVersion, 

556): 

557 """ 

558 Determine the maximum step number from a settings dictionary. 

559 

560 This function determines the maximum step number from a settings dictionary, as specified by the step_number_key. 

561 

562 ## Parameters 

563 

564 - `exp_settings` (`dict`): The experiment settings dictionary. 

565 - `yml_format` (`tbt.YMLFormatVersion`): The YAML format version. 

566 

567 ## Returns 

568 

569 - `int`: The maximum step number. 

570 

571 ## Raises 

572 

573 - `ValueError`: If the number of steps found does not match the expected step count. 

574 """ 

575 

576 step_number_key = yml_format.step_number_key 

577 non_step_sections = yml_format.non_step_section_count 

578 

579 # make sure dict from yml has correct section count 

580 # (steps should all be in one section) 

581 total_sections = len(exp_settings) 

582 if total_sections != non_step_sections + 1: 582 ↛ 583line 582 didn't jump to line 583 because the condition on line 582 was never true

583 raise ValueError( 

584 f"Invalid .yml file, {total_sections} sections were found but the input .yml should have {non_step_sections + 1} sections. Please verify that all top-level keys in the .yml have unique strings and that all steps are contained in a single top-level section." 

585 ) 

586 

587 expected_step_count = exp_settings[yml_format.general_section_key][ 

588 yml_format.step_count_key 

589 ] 

590 

591 found_step_count = 0 

592 while True: 

593 try: 

594 nested_dictionary_location( 

595 d=exp_settings, 

596 key=step_number_key, 

597 value=found_step_count + 1, 

598 ) 

599 except KeyError: 

600 break 

601 found_step_count += 1 

602 

603 # validate number of steps found with steps read by YAML loader 

604 # TODO YAML safeloader will ignore duplicate top level keys, so this check relies on unique step numbers in ascending order (no gaps) to be found. 

605 

606 if expected_step_count != found_step_count: 606 ↛ 607line 606 didn't jump to line 607 because the condition on line 606 was never true

607 raise ValueError( 

608 f"Invalid .yml file, {found_step_count} steps were found but the input .yml should have {expected_step_count} steps from the general setting key '{yml_format.step_count_key}' within the '{yml_format.general_section_key}' section. Please verify that all step_name keys in the .yml have unique strings and that step numbers are continuously-increasing positive integers starting at 1." 

609 ) 

610 

611 return found_step_count 

612 

613 

614def step_settings( 

615 exp_settings: dict, 

616 step_number_key: str, 

617 step_number_val: int, 

618 yml_format: tbt.YMLFormatVersion, 

619) -> Tuple[str, dict]: 

620 """ 

621 Grab specific step settings from an experimental dictionary and return them as a dictionary along with the user-defined step name. 

622 

623 ## Parameters 

624 

625 - `exp_settings` (`dict`): The experiment settings dictionary. 

626 - `step_number_key` (`str`): The key for the step number. 

627 - `step_number_val` (`int`): The value for the step number. 

628 - `yml_format` (`tbt.YMLFormatVersion`): The YAML format version. 

629 

630 ## Returns 

631 

632 - `Tuple[str, dict]`: The step name and the step settings dictionary. 

633 """ 

634 

635 nested_locations = nested_dictionary_location( 

636 d=exp_settings, 

637 key=step_number_key, 

638 value=step_number_val, 

639 ) 

640 ### top level dictionary key name is first index, need key name nested within it (second level, index = 1) 

641 step_name = nested_locations[1] 

642 step_section_key = yml_format.step_section_key 

643 return step_name, exp_settings[step_section_key][step_name] 

644 

645 

646def valid_microscope_connection(host: str, port: int) -> bool: 

647 """ 

648 Determine if a microscope connection can be made. 

649 

650 This function checks if a microscope connection can be made and disconnects if a connection can be made. 

651 

652 ## Parameters 

653 

654 - `host` (`str`): The connection host. 

655 - `port` (`str`): The connection port. 

656 

657 ## Returns 

658 

659 - `bool`: True if the connection can be made, False otherwise. 

660 """ 

661 microscope = tbt.Microscope() 

662 if connect_microscope( 662 ↛ 673line 662 didn't jump to line 673 because the condition on line 662 was always true

663 microscope=microscope, 

664 quiet_output=True, 

665 connection_host=host, 

666 connection_port=port, 

667 ): 

668 if disconnect_microscope( 668 ↛ 673line 668 didn't jump to line 673 because the condition on line 668 was always true

669 microscope=microscope, 

670 quiet_output=True, 

671 ): 

672 return True 

673 return False 

674 

675 

676def enable_external_device(oem: tbt.ExternalDeviceOEM) -> bool: 

677 """ 

678 Determine whether to enable external device control. 

679 

680 This function checks if the external device control should be enabled based on the OEM. 

681 

682 ## Parameters 

683 

684 - `oem` (`tbt.ExternalDeviceOEM`): The OEM of the external device. 

685 

686 ## Returns 

687 

688 - `bool`: True if the external device control should be enabled, False otherwise. 

689 

690 ## Raises 

691 

692 - `NotImplementedError`: If the OEM type is unsupported. 

693 """ 

694 if not isinstance(oem, tbt.ExternalDeviceOEM): 

695 raise NotImplementedError( 

696 f"Unsupported type of {type(oem)}, only 'ExternalDeviceOEM' types are supported." 

697 ) 

698 if oem != tbt.ExternalDeviceOEM.NONE: 698 ↛ 699line 698 didn't jump to line 699 because the condition on line 698 was never true

699 return True 

700 return False 

701 

702 

703def valid_enum_entry(obj: Any, check_type: Enum) -> bool: 

704 """ 

705 Determine if an object is a member of an Enum class. 

706 

707 This function checks if an object is a member of an Enum class. 

708 

709 ## Parameters 

710 

711 - `obj` (`Any`): The object to check. 

712 - `check_type` (`Enum`): The Enum class to check against. 

713 

714 ## Returns 

715 

716 - `bool`: True if the object is a member of the Enum class, False otherwise. 

717 """ 

718 try: 

719 check_type(obj) 

720 except ValueError: 

721 return False 

722 return True 

723 

724 

725def yml_format(version: float) -> tbt.YMLFormatVersion: 

726 """ 

727 Return the YML file format for a given version. 

728 

729 This function returns the YML file format for a given version. 

730 

731 ## Parameters 

732 

733 - `version` (`float`): The version of the YML file. 

734 

735 ## Returns 

736 

737 - `tbt.YMLFormatVersion`: The YML file format for the given version. 

738 

739 ## Raises 

740 

741 - `NotImplementedError`: If the YML file version is unsupported. 

742 """ 

743 supported_versions = [file.version for file in tbt.YMLFormatVersion] 

744 if not version in supported_versions: 

745 raise NotImplementedError( 

746 f'Unsupported YML file version for version "{version}". Valid formats include: {[i.value for i in tbt.YMLFormatVersion]}' 

747 ) 

748 yml_file_idx = supported_versions.index(version) 

749 yml_format = list(tbt.YMLFormatVersion)[yml_file_idx] 

750 return yml_format 

751 

752 

753def yml_to_dict( 

754 *, yml_path_file: Path, version: float, required_keys: Tuple[str, ...] 

755) -> Dict: 

756 """ 

757 Convert a YAML file to a dictionary. 

758 

759 This function reads a YAML file and returns the result as a dictionary. 

760 

761 ## Parameters 

762 

763 - `yml_path_file` (`Path`): The fully pathed location to the input file. 

764 - `version` (`float`): The version of the YAML file in x.y format. 

765 - `required_keys` (`Tuple[str, ...]`): The key(s) that must be in the YAML file for conversion to a dictionary to occur. 

766 

767 ## Returns 

768 

769 - `dict`: The YAML file represented as a dictionary. 

770 

771 ## Raises 

772 

773 - `TypeError`: If the file type is unsupported. 

774 - `OSError`: If the YAML file cannot be opened or decoded. 

775 - `KeyError`: If the required keys are not found in the YAML file. 

776 - `ValueError`: If the version specified in the file does not match the requested version or if the file is empty. 

777 """ 

778 

779 # Compared to the lower() method, the casefold() method is stronger. 

780 # It will convert more characters into lower case, and will find more 

781 # matches on comparison of two strings that are both are converted 

782 # using the casefold() method. 

783 file_type = yml_path_file.suffix.casefold() 

784 

785 supported_types = (".yaml", ".yml") 

786 

787 if file_type not in supported_types: 

788 raise TypeError("Only file types .yaml, and .yml are supported.") 

789 

790 try: 

791 with open(file=yml_path_file, mode="r", encoding="utf-8") as stream: 

792 # See deprecation warning for plain yaml.load(input) at 

793 # https://github.com/yaml/pyyaml/wiki/PyYAML-yaml.load(input)-Deprecation 

794 db = yaml.load(stream, Loader=yaml.SafeLoader) 

795 except yaml.YAMLError as error: 

796 print(f"Error with YAML file: {error}") 

797 # print(f"Could not open: {self.self.path_file_in}") 

798 print(f"Could not open or decode: {yml_path_file}") 

799 # raise yaml.YAMLError 

800 raise OSError from error 

801 

802 if db is None: 802 ↛ 803line 802 didn't jump to line 803 because the condition on line 802 was never true

803 raise ValueError(f"YAML file is empty: {yml_path_file}") 

804 

805 # check keys found in input file against required keys 

806 found_keys = tuple(db.keys()) 

807 keys_exist = tuple(map(lambda x: x in found_keys, required_keys)) 

808 has_required_keys = all(keys_exist) 

809 if not has_required_keys: 

810 raise KeyError(f"Input files must have these keys defined: {required_keys}") 

811 

812 version_specified = db["config_file_version"] 

813 version_requested = version 

814 

815 if version_specified != version_requested: 815 ↛ 816line 815 didn't jump to line 816 because the condition on line 815 was never true

816 ee = f"Version mismatch: specified in file was {version_specified}," 

817 ee += f"requested is {version_requested}" 

818 raise ValueError(ee) 

819 

820 return db 

821 

822 

823def yml_version( 

824 file: Path, 

825 key_name="config_file_version", 

826) -> float: 

827 """ 

828 Return the version of a YAML file if the proper key exists. 

829 

830 ## Parameters 

831 

832 - `file` (`Path`): The path to the YAML file. 

833 - `key_name` (`str, optional`): The key name for the version in the YAML file (default is "config_file_version"). 

834 

835 ## Returns 

836 

837 - `float`: The version of the YAML file. 

838 

839 ## Raises 

840 

841 - `KeyError`: If the version key is not found in the YAML file. 

842 - `ValueError`: If the version value is not a valid float. 

843 """ 

844 with open(file, "r") as stream: 

845 data = yaml.load(stream, Loader=yaml.SafeLoader) 

846 

847 try: 

848 version = data[key_name] 

849 except KeyError: 

850 # print(f"Error with version key: {error}") 

851 raise KeyError(f"Error with version key, '{key_name}' key not found in {file}.") 

852 try: 

853 version = float(version) 

854 except ValueError: 

855 raise ValueError( 

856 f"Could not find valid version in {file} for key {key_name}, found '{version}' which is not a float." 

857 ) 

858 return version 

859 

860 

861def yes_no(question): 

862 """ 

863 Simple Yes/No function. 

864 

865 ## Parameters 

866 

867 - `question` (`str`): The question to ask the user. 

868 

869 ## Returns 

870 

871 - `bool`: True if the user answers "yes", False otherwise. 

872 """ 

873 prompt = f"{question} (y/n): " 

874 while True: 

875 ans = input(prompt).strip().lower() 

876 if ans == "y": 876 ↛ 878line 876 didn't jump to line 878 because the condition on line 876 was always true

877 return True 

878 if ans == "n": 

879 return False 

880 print(f"{ans} is invalid, please try again...") 

881 

882 

883def remove_directory(directory: Path) -> None: 

884 """ 

885 Recursively remove a directory. 

886 

887 ## Parameters 

888 

889 - `directory` (`Path`): The path to the directory to remove. 

890 """ 

891 shutil.rmtree(directory) 

892 

893 

894def split_list(data: List, chunk_size: int) -> List: 

895 """ 

896 Split a list into equal-sized chunks. 

897 

898 ## Parameters 

899 

900 - `data` (`List`): The list to split. 

901 - `chunk_size` (`int`): The size of each chunk. 

902 

903 ## Returns 

904 

905 - `List`: A list of chunks. 

906 """ 

907 result = [] 

908 for i in range(0, len(data), chunk_size): 

909 result.append(data[i : i + chunk_size]) 

910 return result 

911 

912 

913def tabular_list( 

914 data: List, 

915 num_columns: int = Constants.default_column_count, 

916 column_width: int = Constants.default_column_width, 

917) -> str: 

918 """ 

919 Format a list into a tabular string. 

920 

921 ## Parameters 

922 

923 - `data` (`List`): The list to format. 

924 - `num_columns` (`int, optional`): The number of columns in the table (default is Constants.default_column_count). 

925 - `column_width` (`int, optional`): The width of each column in the table (default is Constants.default_column_width). 

926 

927 ## Returns 

928 

929 - `str`: The formatted tabular string. 

930 """ 

931 rows = split_list(data, chunk_size=num_columns) 

932 result = "" 

933 for sublist in rows: 

934 result += "\n" 

935 for item in sublist: 935 ↛ 933line 935 didn't jump to line 933 because the loop on line 935 didn't complete

936 result += f"{item:^{column_width}}" 

937 return result 

938 

939 

940### Functions for tests and CI/CD### 

941 

942 

943def get_test_description() -> str: 

944 """ 

945 Return a test-environment description string for the current machine. 

946 

947 The description combines the detected machine type with the installed 

948 AutoScript version. It is intended for test naming, reporting, or selecting 

949 expected test behavior. 

950 

951 ## Returns 

952 

953 - `description` (`str`): A string containing a description of the platform and the autoscript version. 

954 """ 

955 node = platform.uname().node.lower() 

956 offline_machine = any( 

957 node in machine.lower() or machine.lower() in node 

958 for machine in Constants.offline_machines 

959 ) 

960 hardware_machine = any( 

961 node in machine.lower() or machine.lower() in node 

962 for machine in Constants.microscope_machines 

963 ) 

964 

965 laser_machine = is_laser_available() 

966 

967 api_version = get_autoscript_version() 

968 

969 if offline_machine: 969 ↛ 970,   969 ↛ 9712 missed branches: 1) line 969 didn't jump to line 970 because the condition on line 969 was never true, 2) line 969 didn't jump to line 971 because the condition on line 969 was always true

970 description = "simulated_" 

971 elif hardware_machine and laser_machine: 

972 description = "laser_hardware_" 

973 else: 

974 description = "hardware_" 

975 

976 return description + api_version 

977 

978 

979def get_autoscript_version() -> str: 

980 """ 

981 Get the version of autoscript for the present system 

982 

983 ## Returns 

984 

985 - `version : str`: The version of autoscript 

986 """ 

987 try: 

988 import autoscript_sdb_microscope_client as asmc 

989 

990 version = asmc.build_information.INFO_VERSIONSHORT 

991 except ImportError: 

992 version = "none" 

993 return version 

994 

995 

996def is_laser_available() -> bool: 

997 """ 

998 Get the version of ThermoFisher Laser Control API for the present system 

999 

1000 ## Returns 

1001 

1002 - `version : str`: The version of the Laser API 

1003 """ 

1004 try: 

1005 import Laser.PythonControl as tfs_laser 

1006 

1007 return True 

1008 except ImportError: 

1009 return False 

1010 

1011 

1012def application_files(microscope: tbt.Microscope) -> List[str]: 

1013 """ 

1014 Get the list of application files from the current microscope. 

1015 

1016 This function retrieves the list of application files available on the current microscope, removes any "None" entries, and sorts the list. 

1017 

1018 ## Parameters 

1019 

1020 - `microscope` (`tbt.Microscope`): The microscope object from which to retrieve the application files. 

1021 

1022 ## Returns 

1023 

1024 - `List[str]`: A sorted list of application files available on the microscope. 

1025 """ 

1026 apps = microscope.patterning.list_all_application_files() 

1027 

1028 # Remove "None" entry 

1029 while "None" in apps: 

1030 apps.remove("None") 

1031 apps.sort(key=str.casefold) 

1032 

1033 return apps