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

135 statements  

« prev     ^ index     » next       coverage.py v7.6.5, created at 2024-11-15 02:12 +0000

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3 

4from __future__ import annotations 

5 

6import gevent 

7import logging 

8import numpy 

9from marshmallow import fields, validate 

10 

11from tomo.zseries import ZSeries 

12from . import tomo_helper 

13 

14from daiquiri.core.schema.validators import OneOf 

15from daiquiri.core.components import ( 

16 ComponentActor, 

17 ComponentActorSchema, 

18 ComponentActorKilled, 

19) 

20 

21_logger = logging.getLogger(__name__) 

22 

23 

24class ZseriesscanSchema(ComponentActorSchema): 

25 sampleid = fields.Int( 

26 required=True, 

27 metadata={"title": "Collection"}, 

28 ) 

29 dataset_name = fields.Str( 

30 required=True, 

31 metadata={"title": "Dataset"}, 

32 ) 

33 scan_type = OneOf( 

34 ["Continuous", "Step"], 

35 dump_default="Continuous", 

36 metadata={"title": "Scan type"}, 

37 ) 

38 nb_scans = fields.Int( 

39 dump_default=None, 

40 allow_none=True, 

41 # validate=validate.Range(min=1, max=999999), 

42 metadata={"title": "Nb scans", "readOnly": True}, 

43 ) 

44 z_height = fields.Float( 

45 dump_default=None, 

46 validate=validate.Range(min=0, max=999999), 

47 metadata={"title": "Height", "unit": "mm"}, 

48 ) 

49 z_step = fields.Float( 

50 dump_default=None, 

51 allow_none=True, 

52 metadata={"title": "Z step", "unit": "mm", "readOnly": True}, 

53 ) 

54 z_direction = OneOf( 

55 ["Up", "Down"], 

56 dump_default="Down", 

57 metadata={"title": "Direction"}, 

58 ) 

59 z_min_overlap = fields.Float( 

60 dump_default=10, 

61 validate=validate.Range(min=0, max=99.99999), 

62 metadata={"title": "Min overlap", "unit": "%"}, 

63 ) 

64 z_overlap = fields.Float( 

65 dump_default=10, 

66 validate=validate.Range(min=0, max=99.99999), 

67 required=False, 

68 metadata={"title": "Overlap", "unit": "%", "readOnly": True}, 

69 ) 

70 start_pos = fields.Float( 

71 dump_default=0, 

72 metadata={"title": "Start angle", "unit": "deg"}, 

73 ) 

74 end_pos = fields.Float( 

75 dump_default=360, 

76 metadata={"title": "End angle", "unit": "deg"}, 

77 ) 

78 tomo_n = fields.Int( 

79 dump_default=360, 

80 validate=validate.Range(min=1, max=999999), 

81 metadata={"title": "Nb steps"}, 

82 ) 

83 n_dark = fields.Int( 

84 dump_default=1, 

85 validate=validate.Range(min=0, max=999999), 

86 metadata={"title": "Nb darks"}, 

87 ) 

88 n_flat = fields.Int( 

89 dump_default=1, 

90 validate=validate.Range(min=0, max=999999), 

91 metadata={"title": "Nb flats"}, 

92 ) 

93 expo_time = fields.Float( 

94 validate=validate.Range(min=0.001, max=10), 

95 dump_default=None, 

96 metadata={"title": "Exposure time", "unit": "s"}, 

97 ) 

98 use_sinogram = fields.Bool( 

99 dump_default=True, 

100 metadata={ 

101 "title": "Sinogram", 

102 "description": "Generate or not a sinogram during the scan", 

103 }, 

104 ) 

105 axis_displacement = fields.Int( 

106 dump_default=0, 

107 validate=validate.Range(min=-300, max=300), 

108 metadata={ 

109 "title": "Axis displacement", 

110 "unit": "%", 

111 "description": "0% = centered, 100% = half detector width", 

112 }, 

113 ) 

114 comment = fields.Str( 

115 dump_default="", 

116 metadata={ 

117 "title": "Comment", 

118 }, 

119 ) 

120 enqueue = fields.Bool( 

121 dump_default=True, 

122 metadata={ 

123 "title": "Queue Scan", 

124 }, 

125 ) 

126 

127 class Meta: 

128 uiorder = [ 

129 "sampleid", 

130 "dataset_name", 

131 "scan_type", 

132 "expo_time", 

133 "nb_scans", 

134 "z_step", 

135 "z_min_overlap", 

136 "z_overlap", 

137 "z_height", 

138 "z_direction", 

139 "start_pos", 

140 "end_pos", 

141 "tomo_n", 

142 "use_sinogram", 

143 "n_dark", 

144 "n_flat", 

145 "axis_displacement", 

146 "comment", 

147 "enqueue", 

148 ] 

149 uigroups = [ 

150 {"Data policy": ["sampleid", "dataset_name"], "ui:minwidth": 12}, 

151 "scan_type", 

152 { 

153 "Main": ["expo_time", "start_pos", "end_pos", "tomo_n"], 

154 "ui:minwidth": 12, 

155 }, 

156 { 

157 "Z-series": [ 

158 "z_height", 

159 "nb_scans", 

160 "z_direction", 

161 "z_step", 

162 "z_min_overlap", 

163 "z_overlap", 

164 ], 

165 "ui:minwidth": 6, 

166 }, 

167 {"Options": ["use_sinogram", "n_dark", "n_flat"], "ui:minwidth": 12}, 

168 { 

169 "Half acquisition": ["axis_displacement"], 

170 "ui:minwidth": 12, 

171 }, 

172 "comment", 

173 "enqueue", 

174 ] 

175 uischema = { 

176 "sampleid": {"ui:widget": "SampleId"}, 

177 "dataset_name": {"ui:widget": "DatasetName"}, 

178 "comment": {"ui:widget": "textarea", "ui:options": {"rows": 3}}, 

179 "enqueue": {"classNames": "hidden-row", "ui:widget": "hidden"}, 

180 } 

181 

182 def calculated(self, data): 

183 return self._calculated(**data) 

184 

185 def _calculated( 

186 self, 

187 z_direction: str, 

188 expo_time: float | None = None, 

189 z_height: None | float = None, 

190 z_min_overlap: float | None = None, 

191 **kwargs, 

192 ): 

193 """Returns the calculated values 

194 

195 Arguments: 

196 data: Dictionary containing the actual parameters of the form 

197 """ 

198 result: dict[str, object] = {} 

199 

200 if expo_time is None: 

201 result["expo_time"] = self.calculated_expo_time() 

202 vfov = self.calculated_vfov() 

203 nb_scans = self.calculated_nb_scans( 

204 z_height=z_height, z_min_overlap=z_min_overlap, vfov=vfov 

205 ) 

206 if nb_scans is not None: 

207 result["nb_scans"] = nb_scans 

208 result["z_step"] = self.calculated_z_step( 

209 z_height=z_height, nb_scans=nb_scans, vfov=vfov, z_direction=z_direction 

210 ) 

211 result["z_overlap"] = self.calculated_z_overlap( 

212 z_height=z_height, nb_scans=nb_scans, vfov=vfov 

213 ) 

214 print(result, "vfov", vfov, "nb_scans", nb_scans) 

215 return result 

216 

217 def calculated_expo_time(self) -> float: 

218 # Trick to feed the initial expo time based on the tomo imaging device 

219 tomo_config = tomo_helper.get_active_tomo_config() 

220 if tomo_config is None: 

221 raise RuntimeError("No ACTIVE_TOMOCONFIG selected") 

222 imaging = tomo_config.tomo_imaging 

223 if imaging is None: 

224 return 1.0 

225 return imaging.exposure_time 

226 

227 def calculated_vfov(self) -> None | float: 

228 """Vertical field of view in milimeter""" 

229 tomo_config = tomo_helper.get_active_tomo_config() 

230 tomocam = tomo_config.detectors.active_detector 

231 if tomocam is None: 

232 return None 

233 px = tomocam.sample_pixel_size 

234 if px is None or px == 0: 

235 return None 

236 size = tomocam.actual_size 

237 if size is None: 

238 return None 

239 return px * size[1] * 0.001 

240 

241 def calculated_nb_scans( 

242 self, z_height: None | float, vfov: None | float, z_min_overlap: float | None 

243 ) -> None | int: 

244 """ 

245 n > height / (vfov * (1 - min_overlap)) - min_overlap / (1 - min_overlap) 

246 """ 

247 if z_height is None: 

248 return None 

249 if z_min_overlap is None: 

250 return None 

251 if vfov is None: 

252 return None 

253 min_overlap = z_min_overlap * 0.01 

254 n = z_height / (vfov * (1 - min_overlap)) - min_overlap / (1 - min_overlap) 

255 return max(1, int(numpy.ceil(n))) 

256 

257 def calculated_z_overlap( 

258 self, 

259 z_height: None | float, 

260 vfov: None | float, 

261 nb_scans: int | None, 

262 ) -> None | float: 

263 """ 

264 overlap = (vfov * n - height) / (vfov * (n - 1)) 

265 """ 

266 if z_height is None: 

267 return None 

268 if vfov is None: 

269 return None 

270 if nb_scans is None: 

271 return None 

272 if nb_scans == 1: 

273 return 0 

274 return (vfov * nb_scans - z_height) / (vfov * (nb_scans - 1)) * 100 

275 

276 def calculated_z_step( 

277 self, 

278 z_height: None | float, 

279 vfov: None | float, 

280 nb_scans: int | None, 

281 z_direction: str, 

282 ) -> None | float: 

283 if z_height is None: 

284 return None 

285 if vfov is None: 

286 return None 

287 if nb_scans is None: 

288 return None 

289 if nb_scans == 1: 

290 return 0 

291 step = (z_height - vfov * 0.5) / (nb_scans - 1) 

292 if z_direction == "Up": 

293 return -step 

294 elif z_direction == "Down": 

295 return step 

296 else: 

297 raise ValueError(f"Unexpected value for z_direction. Found: {z_direction}") 

298 

299 

300class ZseriesscanActor(ComponentActor): 

301 schema = ZseriesscanSchema 

302 name = "[tomo] full tomo scan" 

303 

304 metatype = "tomo" 

305 saving_args = {"dataset": "{dataset_name}"} 

306 

307 def method( 

308 self, 

309 scan_type, 

310 dataset_name, 

311 nb_scans, 

312 z_step, 

313 start_pos, 

314 end_pos, 

315 tomo_n, 

316 expo_time, 

317 use_sinogram, 

318 n_dark, 

319 n_flat, 

320 axis_displacement, 

321 comment, 

322 before_scan_starts, 

323 update_datacollection, 

324 **kwargs, 

325 ): 

326 if len(kwargs): 

327 print("Unused params", kwargs) 

328 

329 tomo_config = tomo_helper.get_active_tomo_config() 

330 if tomo_config is None: 

331 raise RuntimeError("No ACTIVE_TOMOCONFIG selected") 

332 

333 actor_config = self.get_config() 

334 zseries_object_names = actor_config.get("zseries_object") 

335 if zseries_object_names is None: 

336 raise RuntimeError( 

337 "'fullfield_name' config field was not setup in this samplescan actor description" 

338 ) 

339 

340 zseries = tomo_helper.get_sequence_from_name( 

341 tomo_config, zseries_object_names, ZSeries 

342 ) 

343 

344 mg = tomo_helper.create_mg(tomo_config) 

345 mg.set_active() 

346 

347 tomo_helper.setup_main_pars( 

348 zseries, 

349 scan_type=scan_type, 

350 use_sinogram=use_sinogram, 

351 n_dark=n_dark, 

352 n_flat=n_flat, 

353 axis_displacement=axis_displacement, 

354 comment=comment, 

355 ) 

356 if use_sinogram: 

357 zseries.pars.dark_flat_for_each_scan = True 

358 

359 scan_info = tomo_helper.create_daiquiri_scan_info( 

360 self, share_data_collection_group=True 

361 ) 

362 

363 before_scan_starts(self) 

364 

365 def run(): 

366 zseries.basic_scan( 

367 collection_name=None, 

368 dataset_name=dataset_name, 

369 delta_pos=z_step, 

370 nb_scans=nb_scans, 

371 tomo_start_pos=start_pos, 

372 tomo_end_pos=end_pos, 

373 tomo_n=tomo_n, 

374 expo_time=expo_time, 

375 step_start_pos=None, 

376 scan_info=scan_info, 

377 trust_data_policy_location=True, 

378 ) 

379 

380 greenlet = gevent.spawn(run) 

381 try: 

382 greenlet.join() 

383 except ComponentActorKilled: 

384 greenlet.kill() 

385 raise 

386 

387 greenlet.get()