Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/implementors/samplescan/scan1.py: 62%

1#!/usr/bin/env python 

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

3import gevent 

4from marshmallow import fields, validate 

5import mmh3 

6 

7from bliss.config.static import get_config 

8from bliss.scanning.scan import ScanState 

9 

10from daiquiri.core.schema.validators import OneOf 

11from daiquiri.core.components import ( 

12 ComponentActor, 

13 ComponentActorSchema, 

14 ComponentActorKilled, 

15) 

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

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

18 

19 

20class Scan1Schema(ComponentActorSchema): 

21 sampleid = fields.Int(required=True) 

22 motor_start = fields.Float(required=True, metadata={"title": "Start Position"}) 

23 motor_end = fields.Float(required=True, metadata={"title": "End Position"}) 

24 npoints = fields.Int(required=True, metadata={"title": "No. Points"}) 

25 time = fields.Float( 

26 validate=validate.Range(min=0.1, max=5), 

27 required=True, 

28 metadata={"title": "Time(s) per Point"}, 

29 ) 

30 detectors = fields.List( 

31 OneOf(["diode", "simu1", "lima_simulator"]), 

32 required=True, 

33 metadata={"title": "Detectors", "uniqueItems": True, "minItems": 1}, 

34 ) 

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

36 

37 class Meta: 

38 uischema = {"sampleid": {"classNames": "hidden-row", "ui:widget": "hidden"}} 

39 

40 

41class Scan1Actor(ComponentActor): 

42 schema = Scan1Schema 

43 name = "scan1" 

44 saving_args = {"data_filename": "{sampleid.name}_scan{datacollectionid}"} 

45 

46 def method(self, **kwargs): 

47 cfg = get_config() 

48 

49 kwargs["motor"] = cfg.get("omega") 

50 kwargs["detectors"] = [cfg.get(d) for d in kwargs["detectors"]] 

51 

52 print("capture params") 

53 kwargs["before_scan_starts"](self) 

54 

55 scan = simple_scan(run=False, **kwargs) 

56 

57 greenlet = gevent.spawn(scan.run) 

58 

59 scan.wait_state(ScanState.STARTING) 

60 kwargs["update_datacollection"]( 

61 self, 

62 datacollectionnumber=mmh3.hash(get_scan_key(scan)) & 0xFFFFFFFF, 

63 imagecontainersubpath="1.1/measurement", 

64 numberofimages=kwargs["npoints"], 

65 exposuretime=kwargs["time"], 

66 ) 

67 

68 try: 

69 greenlet.join() 

70 except ComponentActorKilled as e: 

71 greenlet.kill() 

72 raise e 

73 

74 greenlet.get()