Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/implementors/imageviewer/poiscan.py: 35%

1#!/usr/bin/env python 

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

3import logging 

4from marshmallow import fields, validate, validates_schema, ValidationError 

5import mmh3 

6import numpy 

7 

8from bliss.config.static import get_config 

9from bliss.common.scans import ascan 

10from bliss.scanning.group import Sequence 

11from bliss.scanning.chain import AcquisitionChannel 

12 

13from daiquiri.core.components import ComponentActor, ComponentActorSchema 

14from daiquiri.core.utils import to_wavelength 

15from daiquiri.core.schema.components import RegionSchema 

16from daiquiri.core.hardware.bliss.session import * 

17from daiquiri.core.hardware.blissdata.helpers import get_scan_key 

18 

19from .createmap import CreatemapActor 

20from .beamlineparams import BeamlineParamsSchema 

21 

22 

23logger = logging.getLogger(__name__) 

24cfg = get_config() 

25 

26 

27class PoiscanSchema(ComponentActorSchema): 

28 subsampleid = fields.Int(required=True) 

29 repeats = fields.Int( 

30 required=True, 

31 metadata={"title": "No. Repeats"}, 

32 validate=validate.Range(min=1), 

33 dump_default=1, 

34 ) 

35 regions = fields.List( 

36 fields.Nested(RegionSchema), 

37 metadata={"title": "Energy Regions", "minItems": 1}, 

38 required=True, 

39 ) 

40 reference = fields.Float( 

41 metadata={"title": "Energy Reference", "unit": "keV"}, 

42 validate=validate.Range(min=0.1), 

43 required=True, 

44 ) 

45 

46 beamlineparams = fields.Nested( 

47 BeamlineParamsSchema, metadata={"title": "Beamline Parameters"} 

48 ) 

49 enqueue = fields.Bool(metadata={"title": "Queue Scan"}, dump_default=True) 

50 

51 @validates_schema 

52 def schema_validate(self, data, **kwargs): 

53 objs = data.get("objects") 

54 if objs: 

55 if len(objs) > 1 and data.get("enqueue") is False: 

56 raise ValidationError( 

57 f"Can only queue scan when more than one object is selected. {len(objs)} objects selected" 

58 ) 

59 

60 class Meta: 

61 uiorder = [ 

62 "subsampleid", 

63 "reference", 

64 "regions", 

65 "repeats", 

66 "beamlineparams", 

67 "enqueue", 

68 ] 

69 uischema = { 

70 "subsampleid": {"classNames": "hidden-row", "ui:widget": "hidden"}, 

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

72 "regions": {"ui:field": "arrayTable"}, 

73 "beamlineparams": {"ui:field": "optionalParams"}, 

74 } 

75 

76 

77class PoiscanActor(ComponentActor): 

78 schema = PoiscanSchema 

79 metatype = "Energy scan" 

80 name = "poiscan" 

81 

82 def method(self, **kwargs): 

83 print("Add poiscan", kwargs) 

84 

85 print("moving to poi") 

86 kwargs["absol"]["move_to"](kwargs["absol"]) 

87 

88 print("moving to addiional positions") 

89 kwargs["absol"]["move_to_additional"](kwargs["absol"]["positions"]) 

90 

91 print("capture params and image") 

92 kwargs["before_scan_starts"](self) 

93 

94 points = int( 

95 (kwargs["regions"][0]["end_e"] - kwargs["regions"][0]["start_e"]) 

96 / kwargs["regions"][0]["step"] 

97 ) 

98 

99 print("points", points) 

100 

101 mca = cfg.get("simu1") 

102 mca.block_size = 10 

103 

104 seq = Sequence() 

105 seq.add_custom_channel(AcquisitionChannel("avg_energy", float, ())) 

106 

107 rois = kwargs["get_rois"]()["rois"] 

108 roi_data = {} 

109 for roi in rois: 

110 seq.add_custom_channel(AcquisitionChannel(f"avg_{roi['name']}", float, ())) 

111 roi_data[roi["maproiid"]] = {"name": roi["name"], "data": []} 

112 

113 diodes = [] 

114 for i in range(4): 

115 diodes.append(config.get(f"diode{i+2}")) 

116 

117 with seq.sequence_context() as scan_seq: 

118 scans = [] 

119 for i in range(kwargs["repeats"]): 

120 scan = ascan( 

121 omega, 

122 kwargs["reference"], 

123 kwargs["reference"] + kwargs["regions"][0]["end_e"] / 1000, 

124 # Intervals vs points ! 

125 points - 1, 

126 kwargs["regions"][0]["dwell"], 

127 diode, 

128 diodes[0], 

129 diodes[1], 

130 diodes[2], 

131 diodes[3], 

132 mca, 

133 run=False, 

134 ) 

135 scans.append(scan) 

136 scan_seq.add(scan) 

137 

138 kwargs["update_datacollection"]( 

139 self, 

140 datacollectionnumber=mmh3.hash(get_scan_key(seq)) & 0xFFFFFFFF, 

141 imagecontainersubpath="1.1/measurement", 

142 exposuretime=kwargs["regions"][0]["dwell"], 

143 numberofimages=points, 

144 numberofpasses=kwargs["repeats"], 

145 wavelength=to_wavelength(kwargs["reference"] * 1e3), 

146 emit_start=True, 

147 ) 

148 

149 scan_number = None 

150 for i, scan in enumerate(scans): 

151 scan.run() 

152 

153 scan_number = mmh3.hash(get_scan_key(scan)) & 0xFFFFFFFF 

154 spectra = kwargs["scans"].get_scan_spectra(scan_number, allpoints=True) 

155 

156 mapsActor = CreatemapActor() 

157 maps = mapsActor.method(spectra=spectra, rois=rois) 

158 

159 if maps: 

160 for j, mroi in enumerate(maps[0]["maps"]): 

161 roi_data[mroi["maproiid"]]["data"].append(mroi["data"]) 

162 

163 if j == 0: 

164 kwargs["add_scanqualityindicators"]( 

165 self, point=i + 1, total=sum(mroi["data"]) 

166 ) 

167 

168 avg_len = 0 

169 avgs = {} 

170 for roi in roi_data.values(): 

171 avg = numpy.average(roi["data"], axis=0) 

172 avg_len = len(avg) 

173 seq.custom_channels[f"avg_{roi['name']}"].emit(avg) 

174 avgs[roi["name"]] = avg 

175 

176 energy_counter = "axis:omega" 

177 energy = [] 

178 scalars = kwargs["scans"].get_scan_data( 

179 scan_number, per_page=1e10, scalars=["axis:omega"] 

180 ) 

181 if energy_counter in scalars["data"]: 

182 energy = scalars["data"][energy_counter]["data"] 

183 else: 

184 logger.warning("Cannot find energy counter") 

185 energy = range(avg_len) 

186 

187 seq.custom_channels["avg_energy"].emit(energy) 

188 

189 with kwargs["open_attachment"]( 

190 self, "xy", suffix="_average", ext="asc" 

191 ) as avgf: 

192 headers = "\t".join(avgs.keys()) 

193 data = numpy.array([energy] + list(avgs.values())).T 

194 numpy.savetxt( 

195 avgf, 

196 data, 

197 delimiter="\t", 

198 header=f"{energy_counter}\t{headers}", 

199 )