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
« 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`.
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.
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.
15## Utility categories
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` |
28## Typical usage
30Read and validate a YAML configuration file:
32```python
33from pathlib import Path
35from pytribeam import utilities as ut
37version = ut.yml_version(Path("experiment.yml"))
38fmt = ut.yml_format(version)
40settings = ut.yml_to_dict(
41 yml_path_file=Path("experiment.yml"),
42 version=version,
43 required_keys=fmt.required_keys,
44)
45```
47Check whether a value lies inside a closed interval:
49```python
50import pytribeam.types as tbt
51from pytribeam import utilities as ut
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```
60Connect to a microscope while suppressing connection output:
62```python
63import pytribeam.types as tbt
64from pytribeam import utilities as ut
66microscope = tbt.Microscope()
68ut.connect_microscope(
69 microscope=microscope,
70 quiet_output=True,
71)
73# Use microscope...
75ut.disconnect_microscope(microscope)
76```
78Format a long list for display:
80```python
81from pytribeam import utilities as ut
83print(ut.tabular_list(["CBS", "ETD", "TLD", "ABS"]))
84```
86## YAML configuration helpers
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`.
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.
97## Beam dispatch
99`beam_type` is implemented with `functools.singledispatch` and maps
100`pytribeam` beam wrapper types to the corresponding microscope beam property:
102| Input type | Returned microscope object |
103| --- | --- |
104| `tbt.ElectronBeam` | `microscope.beams.electron_beam` |
105| `tbt.IonBeam` | `microscope.beams.ion_beam` |
107Unsupported beam types raise `NotImplementedError`.
109## Test and environment helpers
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.
116## Notes
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.
121<hr style="height: 12px; background-color: #333; border: none;">
122"""
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]
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
164# # Autoscript modules
165import yaml
166import contextlib
167import sys
168from pandas import json_normalize
170# # # 3rd party module
171# from schema import Schema, And, Use, Optional, SchemaError
173# # Local scripts
174import pytribeam.types as tbt
176# import pytribeam.constants as cs
177from pytribeam.constants import Constants
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.
185 ## Parameters
187 - `beam` (`Any`): The beam object.
189 ## Returns
191 - `property`: The beam property object.
193 ## Raises
195 - `NotImplementedError`: If the beam type is not implemented.
196 """
197 _ = beam # no operation
198 raise NotImplementedError()
201@beam_type.register
202def _electron_beam_type(beam: tbt.ElectronBeam, microscope: tbt.Microscope) -> property:
203 """
204 Return the electron beam property object.
206 ## Parameters
208 - `beam` (`tbt.ElectronBeam`): The electron beam object.
209 - `microscope` (`tbt.Microscope`): The microscope object.
211 ## Returns
213 - `property`: The electron beam property object.
214 """
215 return microscope.beams.electron_beam
218@beam_type.register
219def _ion_beam_type(beam: tbt.IonBeam, microscope: tbt.Microscope) -> property:
220 """
221 Return the ion beam property object.
223 ## Parameters
225 - `beam` (`tbt.IonBeam`): The ion beam object.
226 - `microscope` (`tbt.Microscope`): The microscope object.
228 ## Returns
230 - `property`: The ion beam property object.
231 """
232 return microscope.beams.ion_beam
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.
244 ## Parameters
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).
251 ## Returns
253 - `bool`: True if the connection is successful.
255 ## Raises
257 - `ConnectionError`: If the connection fails.
258 """
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)
271 else:
272 microscope.connect()
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 )
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 )
296def dict_to_yml(db: dict, file_path: Path) -> Path:
297 """
298 Convert a dictionary to a YAML file.
300 ## Parameters
302 - `db` (`dict`): The dictionary to convert.
303 - `file_path` (`Path`): The path to save the YAML file.
305 ## Returns
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 )
317 return file_path
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.
327 ## Parameters
329 - `microscope` (`tbt.Microscope`): The microscope object to disconnect.
330 - `quiet_output` (`bool, optional`): Whether to suppress printout (default is True).
332 ## Returns
334 - `bool`: True if the disconnection is successful.
336 ## Raises
338 - `ConnectionError`: If the disconnection fails.
339 """
340 if quiet_output:
341 with nostdout():
342 microscope.disconnect()
343 else:
344 microscope.disconnect()
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")
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.
356 ## Parameters
358 - `exp_settings` (`dict`): The experiment settings dictionary.
359 - `yml_format` (`tbt.YMLFormat`): The YAML format version.
361 ## Returns
363 - `dict`: The general experiment settings as a dictionary.
364 """
365 general_key = yml_format.general_section_key
366 return exp_settings[general_key]
369def step_type(settings: dict, yml_format: tbt.YMLFormat) -> tbt.StepType:
370 """
371 Determine the step type for a specific step settings dictionary.
373 ## Parameters
375 - `settings` (`dict`): The step settings dictionary.
376 - `yml_format` (`tbt.YMLFormat`): The YAML format version.
378 ## Returns
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 )
386 return step_type
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.
393 ## Parameters
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.
399 ## Returns
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)
413def gen_dict_extract(key, var):
414 """
415 Extract values from a nested dictionary by key.
417 ## Parameters
419 - `key` (`str`): The key to search for.
420 - `var` (`dict`): The nested dictionary to search.
422 ## Yields
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
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.
443 This function returns a list of key values from the highest to the lowest level of nested dictionaries.
445 ## Parameters
447 - `d` (`dict`): The dictionary to search.
448 - `key` (`str`): The key to search for.
449 - `value` (`Any`): The value to search for.
451 ## Returns
453 - `List[str]`: The nested location of the key-value pair.
455 ## Raises
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
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.
471 This function returns a list of key values from the highest to the lowest level of nested dictionaries.
473 ## Parameters
475 - `d` (`dict`): The dictionary to search.
476 - `key` (`str`): The key to search for.
477 - `value` (`Any`): The value to search for.
479 ## Returns
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
494def _flatten(dictionary: dict) -> dict:
495 """
496 Flatten a dictionary using pandas.
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
501 ## Parameters
503 - `dictionary` (`dict`): The dictionary to flatten.
505 ## Returns
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
514def none_value_dictionary(dictionary: dict) -> bool:
515 """
516 Check if all values in a dictionary are None.
518 This function returns True if all values in the dictionary are None, and False otherwise.
520 ## Parameters
522 - `dictionary` (`dict`): The dictionary to check.
524 ## Returns
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()])
533@contextlib.contextmanager
534def nostdout():
535 """
536 Create a dummy file to suppress output.
538 This function creates a dummy file to suppress output.
540 ## Yields
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
553def step_count(
554 exp_settings: dict,
555 yml_format: tbt.YMLFormatVersion,
556):
557 """
558 Determine the maximum step number from a settings dictionary.
560 This function determines the maximum step number from a settings dictionary, as specified by the step_number_key.
562 ## Parameters
564 - `exp_settings` (`dict`): The experiment settings dictionary.
565 - `yml_format` (`tbt.YMLFormatVersion`): The YAML format version.
567 ## Returns
569 - `int`: The maximum step number.
571 ## Raises
573 - `ValueError`: If the number of steps found does not match the expected step count.
574 """
576 step_number_key = yml_format.step_number_key
577 non_step_sections = yml_format.non_step_section_count
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 )
587 expected_step_count = exp_settings[yml_format.general_section_key][
588 yml_format.step_count_key
589 ]
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
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.
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 )
611 return found_step_count
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.
623 ## Parameters
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.
630 ## Returns
632 - `Tuple[str, dict]`: The step name and the step settings dictionary.
633 """
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]
646def valid_microscope_connection(host: str, port: int) -> bool:
647 """
648 Determine if a microscope connection can be made.
650 This function checks if a microscope connection can be made and disconnects if a connection can be made.
652 ## Parameters
654 - `host` (`str`): The connection host.
655 - `port` (`str`): The connection port.
657 ## Returns
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
676def enable_external_device(oem: tbt.ExternalDeviceOEM) -> bool:
677 """
678 Determine whether to enable external device control.
680 This function checks if the external device control should be enabled based on the OEM.
682 ## Parameters
684 - `oem` (`tbt.ExternalDeviceOEM`): The OEM of the external device.
686 ## Returns
688 - `bool`: True if the external device control should be enabled, False otherwise.
690 ## Raises
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
703def valid_enum_entry(obj: Any, check_type: Enum) -> bool:
704 """
705 Determine if an object is a member of an Enum class.
707 This function checks if an object is a member of an Enum class.
709 ## Parameters
711 - `obj` (`Any`): The object to check.
712 - `check_type` (`Enum`): The Enum class to check against.
714 ## Returns
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
725def yml_format(version: float) -> tbt.YMLFormatVersion:
726 """
727 Return the YML file format for a given version.
729 This function returns the YML file format for a given version.
731 ## Parameters
733 - `version` (`float`): The version of the YML file.
735 ## Returns
737 - `tbt.YMLFormatVersion`: The YML file format for the given version.
739 ## Raises
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
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.
759 This function reads a YAML file and returns the result as a dictionary.
761 ## Parameters
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.
767 ## Returns
769 - `dict`: The YAML file represented as a dictionary.
771 ## Raises
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 """
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()
785 supported_types = (".yaml", ".yml")
787 if file_type not in supported_types:
788 raise TypeError("Only file types .yaml, and .yml are supported.")
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
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}")
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}")
812 version_specified = db["config_file_version"]
813 version_requested = version
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)
820 return db
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.
830 ## Parameters
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").
835 ## Returns
837 - `float`: The version of the YAML file.
839 ## Raises
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)
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
861def yes_no(question):
862 """
863 Simple Yes/No function.
865 ## Parameters
867 - `question` (`str`): The question to ask the user.
869 ## Returns
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...")
883def remove_directory(directory: Path) -> None:
884 """
885 Recursively remove a directory.
887 ## Parameters
889 - `directory` (`Path`): The path to the directory to remove.
890 """
891 shutil.rmtree(directory)
894def split_list(data: List, chunk_size: int) -> List:
895 """
896 Split a list into equal-sized chunks.
898 ## Parameters
900 - `data` (`List`): The list to split.
901 - `chunk_size` (`int`): The size of each chunk.
903 ## Returns
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
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.
921 ## Parameters
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).
927 ## Returns
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
940### Functions for tests and CI/CD###
943def get_test_description() -> str:
944 """
945 Return a test-environment description string for the current machine.
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.
951 ## Returns
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 )
965 laser_machine = is_laser_available()
967 api_version = get_autoscript_version()
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_"
976 return description + api_version
979def get_autoscript_version() -> str:
980 """
981 Get the version of autoscript for the present system
983 ## Returns
985 - `version : str`: The version of autoscript
986 """
987 try:
988 import autoscript_sdb_microscope_client as asmc
990 version = asmc.build_information.INFO_VERSIONSHORT
991 except ImportError:
992 version = "none"
993 return version
996def is_laser_available() -> bool:
997 """
998 Get the version of ThermoFisher Laser Control API for the present system
1000 ## Returns
1002 - `version : str`: The version of the Laser API
1003 """
1004 try:
1005 import Laser.PythonControl as tfs_laser
1007 return True
1008 except ImportError:
1009 return False
1012def application_files(microscope: tbt.Microscope) -> List[str]:
1013 """
1014 Get the list of application files from the current microscope.
1016 This function retrieves the list of application files available on the current microscope, removes any "None" entries, and sorts the list.
1018 ## Parameters
1020 - `microscope` (`tbt.Microscope`): The microscope object from which to retrieve the application files.
1022 ## Returns
1024 - `List[str]`: A sorted list of application files available on the microscope.
1025 """
1026 apps = microscope.patterning.list_all_application_files()
1028 # Remove "None" entry
1029 while "None" in apps:
1030 apps.remove("None")
1031 apps.sort(key=str.casefold)
1033 return apps