Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/implementors/tomo/tomo_helper.py: 0%

1from __future__ import annotations 

2 

3from typing import Any 

4import contextlib 

5import logging 

6from bliss.common import cleanup 

7from tomo.controllers.tomo_config import TomoConfig 

8from tomo.utils import configure_shift, ShiftType 

9from bliss.config.static import get_config 

10from daiquiri.core.components import ( 

11 ComponentActorSchema, 

12) 

13from bliss.common.measurementgroup import MeasurementGroup 

14 

15_logger = logging.getLogger(__name__) 

16 

17 

18try: 

19 from tomo.globals import ACTIVE_TOMOCONFIG 

20except ImportError: 

21 ACTIVE_TOMOCONFIG = None 

22 

23 

24def get_active_tomo_config() -> TomoConfig | None: 

25 tomo_config = ( 

26 ACTIVE_TOMOCONFIG is not None and ACTIVE_TOMOCONFIG.deref_active_object() 

27 ) 

28 return tomo_config 

29 

30 

31def patch_actor_validator(func): 

32 """Patch the validation function from the actor schema to make is a bit more 

33 easy to use 

34 

35 .. code-block:: 

36 

37 @tomo_helper.patch_actor_validator 

38 def calculated(self, expo_time: float | None = None, **kwargs): 

39 ... 

40 """ 

41 

42 def new_func(self, data): 

43 try: 

44 try: 

45 return func(self, **data) 

46 except TypeError as e: 

47 if f"{func.name()}" in e.args[0]: 

48 # The function signature is not right, it means we expect 

49 # params which are not yet part of the function 

50 _logger.debug(f"{func.name} failed with: {e}", exc_info=True) 

51 else: 

52 raise 

53 except Exception as e: 

54 # Use an error instead of a debug 

55 _logger.error(f"{func.name} failed with: {e}", exc_info=True) 

56 raise 

57 return None 

58 

59 return new_func 

60 

61 

62def get_sequence_from_name( 

63 tomo_config: TomoConfig, object_names: str | list[str], expected_class 

64): 

65 if isinstance(object_names, str): 

66 object_names = [object_names] 

67 

68 bliss_config = get_config() 

69 

70 for name in object_names: 

71 sequence = bliss_config.get(name) 

72 if not isinstance(sequence, expected_class): 

73 raise RuntimeError( 

74 f"'Object {name} is expected to be a {expected_class.name}. Found {type(sequence)}" 

75 ) 

76 if sequence.tomo_config is tomo_config: 

77 break 

78 else: 

79 raise RuntimeError( 

80 f"No {expected_class.__name__} object was found for tomo config '{tomo_config.name}'" 

81 ) 

82 return sequence 

83 

84 

85def create_daiquiri_scan_info( 

86 actor: ComponentActorSchema, 

87 share_data_collection: bool = False, 

88 share_data_collection_group: bool = False, 

89) -> dict[str, Any]: 

90 scan_info = { 

91 "daiquiri": {"sampleid": actor["sampleid"], "sessionid": actor["sessionid"]} 

92 } 

93 if share_data_collection: 

94 datacollectionid = actor.get("datacollectionid") 

95 if datacollectionid: 

96 scan_info["daiquiri"]["datacollectionid"] = datacollectionid 

97 if share_data_collection_group: 

98 datacollectiongroupid = actor.get("datacollectiongroupid") 

99 if datacollectiongroupid is not None: 

100 scan_info["daiquiri"]["datacollectiongroupid"] = datacollectiongroupid 

101 return scan_info 

102 

103 

104@contextlib.contextmanager 

105def y_axis_at(config: TomoConfig, position: float, enabled: bool = True): 

106 """ 

107 Move and restore the rotation axis 

108 """ 

109 if enabled: 

110 context = cleanup.cleanup( 

111 config.y_axis, 

112 restore_list=(cleanup.axis.POS,), 

113 ) 

114 else: 

115 context = contextlib.nullcontext 

116 

117 with context: 

118 if enabled: 

119 config.y_axis.move(position) 

120 yield 

121 

122 

123def create_mg(tomo_config: TomoConfig): 

124 tomocam = tomo_config.detectors.active_detector 

125 if tomocam is None: 

126 raise RuntimeError(f"No active detector selected in `{tomo_config.name}`") 

127 cam = tomocam.detector 

128 mg = MeasurementGroup("daiquiri_mg", {"counters": []}) 

129 try: 

130 mg.add(cam.image) 

131 except Exception: # nosec 

132 pass 

133 mg.enable(cam.image.fullname) 

134 if tomo_config.machinfo is not None: 

135 try: 

136 mg.add(tomo_config.machinfo.counters.current) 

137 except Exception: # nosec 

138 pass 

139 mg.enable(tomo_config.machinfo.counters.current.fullname) 

140 return mg 

141 

142 

143def setup_main_pars( 

144 sequence, 

145 scan_type: str, 

146 use_sinogram: bool, 

147 n_dark: int, 

148 n_flat: int, 

149 axis_displacement: float, 

150 comment: str, 

151): 

152 sequence.pars.scan_type = scan_type.upper() 

153 sequence.pars.activate_sinogram = use_sinogram 

154 if use_sinogram: 

155 # A sinogram is expected so a dark will be acquired 

156 sequence.pars.dark_at_start = True 

157 sequence.pars.flat_at_start = True 

158 sequence.pars.flat_n = n_flat 

159 sequence.pars.dark_n = n_dark 

160 sequence.pars.comment = comment 

161 if axis_displacement == 0: 

162 sequence.pars.half_acquisition = False 

163 sequence.pars.shift_in_fov_factor = 0 

164 sequence.pars.shift_in_mm = 0 

165 else: 

166 configure_shift(sequence, axis_displacement / 100, ShiftType.FOV_FACTOR) 

167 sequence.pars.half_acquisition = True