Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/hardware/bliss/__init__.py: 84%

1#!/usr/bin/env python 

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

3import importlib 

4import logging 

5 

6from marshmallow import ValidationError, fields, validates_schema, post_load 

7from bliss.config.static import get_config 

8 

9from daiquiri.core.hardware.abstract import ProtocolHandler 

10from daiquiri.core.hardware.bliss.object import BlissDummyObject 

11from daiquiri.core.schema.hardware import HOConfigSchema 

12from daiquiri.core.exceptions import InvalidYAML 

13from daiquiri.core.utils import loader 

14 

15 

16logger = logging.getLogger(__name__) 

17 

18bl = logging.getLogger("bliss") 

19bl.setLevel(logging.WARNING) 

20bl.disabled = True 

21 

22bl = logging.getLogger("bliss.common.mapping") 

23bl.disabled = True 

24 

25 

26class BlissHOConfigSchema(HOConfigSchema): 

27 """The Bliss Hardware Object Config Schema""" 

28 

29 address = fields.Str(metadata={"description": "Beacon object id"}) 

30 type = fields.Str(metadata={"description": "Object type for objects without id"}) 

31 

32 @validates_schema 

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

34 if not (data.get("address") or data.get("url")): 

35 raise ValidationError( 

36 "Object must have either an `address` or `url` defined" 

37 ) 

38 

39 @post_load 

40 def populate(self, data, **kwargs): 

41 """Generate the device address from its url""" 

42 if data.get("url"): 

43 _, address = data["url"].split("://") 

44 data["address"] = address 

45 

46 return data 

47 

48 

49class BlissHandler(ProtocolHandler): 

50 """The bliss protocol handler 

51 

52 Returns an instance of an abstracted bliss object 

53 

54 The bliss protocol handler first checks the kwargs conform to the BlissHOConfigSchema 

55 defined above. This address is used to retrieve the bliss object. Its class is then mapped 

56 to an abstract class and a bliss specific instance is created (see hardware/bliss/motor.py) 

57 """ 

58 

59 library = "bliss" 

60 

61 _class_map = { 

62 "bliss.common.axis.Axis": "motor", 

63 "bliss.controllers.actuator.Actuator": "actuator", 

64 "bliss.controllers.multiplepositions.MultiplePositions": "multiposition", 

65 "bliss.common.shutter.BaseShutter": "shutter", 

66 "bliss.controllers.test.objectref.ObjectRef": "objectref", 

67 "bliss.controllers.intraled.Intraled": "intraled", 

68 "bliss.controllers.lima.lima_base.Lima": "lima", 

69 "tomo.procedures.base_procedure.SessionProcedure": "procedure", 

70 "tomo.procedures.user_script.UserScriptProcedure": "procedure", 

71 "tomo.controllers.tomo_config.TomoConfig": "tomoconfig", 

72 "tomo.controllers.tomo_detector.TomoDetector": "tomodetector", 

73 "tomo.controllers.tomo_detectors.TomoDetectors": "tomodetectors", 

74 "tomo.controllers.tomo_imaging.TomoImaging": "tomoimaging", 

75 "tomo.controllers.tomo_sample_stage.TomoSampleStage": "tomosamplestage", 

76 "tomo.controllers.flat_motion.FlatMotion": "tomoflatmotion", 

77 "tomo.controllers.holotomo.Holotomo": "tomoholo", 

78 "tomo.controllers.flat_motion.FlatMotion": "tomoflatmotion", 

79 "tomo.controllers.reference_position.ReferencePosition": "tomoreferenceposition", 

80 "tomo.optic.base_optic.BaseOptic": "tomooptic", 

81 } 

82 

83 _class_name_map = { 

84 "EBV": "beamviewer", 

85 "ChannelFromConfig": "channelfromconfig", 

86 "volpi": "volpi", 

87 "TestObject": "test", 

88 "Fshutter": "shutter", 

89 "transmission": "transmission", 

90 "tango_attr_as_counter": "tango_attr_as_counter", 

91 "ShimadzuCBM20": "shimadzucbm20", 

92 "ShimadzuPDA": "shimadzupda", 

93 "ID26Attenuator": "attenuator_wago", 

94 "PresetManager": "presetmanager", 

95 "LaserController": "laser", 

96 "LaserHeating": "laserheating", 

97 "IcePapTrack": "remotemotor", 

98 } 

99 

100 def get(self, **kwargs): 

101 try: 

102 kwargs = BlissHOConfigSchema().load(kwargs) 

103 except ValidationError as err: 

104 raise InvalidYAML( 

105 { 

106 "message": "Bliss hardware object definition is invalid", 

107 "file": "hardware.yml", 

108 "obj": kwargs, 

109 "errors": err.messages, 

110 } 

111 ) from err 

112 

113 obj_type = kwargs.get("type") 

114 

115 if obj_type == "activetomoconfig": 

116 # It's a global proxy without real instance 

117 class GlobalBlissObject: 

118 name = None 

119 

120 obj = GlobalBlissObject() 

121 else: 

122 config = get_config() 

123 try: 

124 obj = config.get(kwargs.get("address")) 

125 except Exception: 

126 logger.exception(f"Couldn't get bliss object {kwargs.get('address')}") 

127 return BlissDummyObject(**kwargs) 

128 kwargs["obj"] = obj 

129 

130 obj_type = kwargs.get("type") 

131 if obj_type is not None: 

132 return loader( 

133 "daiquiri.core.hardware.bliss", 

134 "", 

135 obj_type, 

136 **kwargs, 

137 ) 

138 

139 for bliss_mapping, mapped_class in self._class_map.items(): 

140 bliss_file, bliss_class_name = bliss_mapping.rsplit(".", 1) 

141 # Some classes may not be available depending on the bliss version 

142 try: 

143 bliss_module = importlib.import_module(bliss_file) 

144 bliss_class = getattr(bliss_module, bliss_class_name) 

145 except ModuleNotFoundError: 

146 logger.warning(f"Could not find bliss module {bliss_mapping}") 

147 continue 

148 

149 if isinstance(kwargs["obj"], bliss_class): 

150 return loader( 

151 "daiquiri.core.hardware.bliss", "", mapped_class, **kwargs 

152 ) 

153 

154 cls = kwargs["obj"].__class__.__name__ 

155 if cls in self._class_name_map: 

156 return loader( 

157 "daiquiri.core.hardware.bliss", 

158 "", 

159 self._class_name_map[cls], 

160 **kwargs, 

161 ) 

162 

163 logger.error("No class found for {cls}".format(cls=cls)) 

164 return BlissDummyObject(**kwargs)