Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/components/tomo/scans/tiling_sequence_tomo_scan.py: 35%

1#!/usr/bin/env python 

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

3 

4import logging 

5import typing 

6import enum 

7from ..datatype import TomoScan 

8from .tomo_scan import IncompatibleScanData 

9from .. import tomovis_client 

10 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15class TilingSequenceTomoScan(TomoScan): 

16 def _init(self, metadata): 

17 tomotype = metadata.get("tomotype", None) 

18 if tomotype != "tilingseq": 

19 raise IncompatibleScanData("Not a tomo scan") 

20 

21 self._state: enum.Enum = None 

22 self._tomovisId: int = None 

23 self._nb_started_subscans = 0 

24 

25 def __start_tomovis_processing(self, metadata): 

26 tomovis_uri = self.component.tomovis_uri 

27 if tomovis_uri: 

28 filename = metadata.get("filename", None) 

29 scan_nb = metadata.get("scan_nb", 1) 

30 dataset = f"{scan_nb}.1" 

31 if filename is None or dataset is None: 

32 logger.error( 

33 "Missing metadata from tiling scan sequence. Call to Tomovis service skipped." 

34 ) 

35 else: 

36 self._tomovisId = tomovis_client.create_tiling( 

37 tomovis_uri, filename, dataset 

38 ) 

39 logger.info(f"Tomovis reconstruction created: {self._tomovisId}") 

40 

41 def to_rest(self) -> dict: 

42 return { 

43 "scanid": self.scanid, 

44 "state": self._state or "UNKNOWN", 

45 "tomovisId": self._tomovisId, 

46 } 

47 

48 def on_scan_started(self, metadata): 

49 scaninfos = self.component.get_scaninfos() 

50 scaninfos.last_tiling = self 

51 

52 # This is not anymore available in BLISS 2.0 

53 # FIXME: The API have to be adapted 

54 state = metadata.get("state", "RUNNING") 

55 self._state = state 

56 

57 self.__start_tomovis_processing(metadata) 

58 

59 self.component.emit( 

60 "update_scan_info", 

61 {"id": "lasttiling", "data": self.to_rest()}, 

62 ) 

63 

64 def on_subscan_started(self, scanid, scan: typing.Optional[TomoScan]): 

65 self._nb_started_subscans += 1 

66 

67 def on_scan_terminated(self, metadata): 

68 state = self._get_daiquiri_scan_state(metadata) 

69 self._state = state 

70 self.component.emit( 

71 "update_scan_info", 

72 {"id": "lasttiling", "data": self.to_rest()}, 

73 )