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

1#!/usr/bin/env python

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

3from bliss.config.static import get_config

4from bliss.common import event

5from bliss import global_map

6from collections.abc import MutableSequence

8from daiquiri.core.hardware.abstract import (

9 MappedHardwareObject,

10 HardwareObject,

11 HardwareProperty,

12 AbstractHardwareProperty2,

13)

14from daiquiri.core.utils import get_nested_attr

16import logging

18logger = logging.getLogger(__name__)

21class ObjectRefProperty(HardwareProperty):

22 """Attribute read from BLISS as another BLISS object and exposed to Daiquiri

23 as an object name

25 Attributes:

26 name: Name of the attribute in the remote hardware device

27 compose: If true the referenced object is part of this component and will

28 automatically be exposed registered in Daiquiri and exposed to

29 the front end.

30 """

32 def __init__(self, name, compose: bool = False):

33 HardwareProperty.__init__(self, name)

34 self._compose = compose

36 @property

37 def compose(self) -> bool:

38 """If true the referenced object compose this hardware object.

40 As result if this object is registered in Daiquiri, the referenced

41 object will be registered too.

42 """

43 return self._compose

45 def translate_from(self, value):

46 if value is None:

47 name = ""

48 elif isinstance(value, str):

49 name = value

50 else:

51 name = value.name

52 return "hardware:" + name

55class ObjectRefListProperty(HardwareProperty):

56 """Attribute read from BLISS as another BLISS object and exposed to Daiquiri

57 as a list of object name

59 Attributes:

60 name: Name of the attribute in the remote hardware device

61 compose: If true the referenced object is part of this component and will

62 automatically be exposed registered in Daiquiri and exposed to

63 the front end.

65 """

67 def __init__(self, name, compose: bool = False):

68 HardwareProperty.__init__(self, name)

69 self._compose = compose

71 @property

72 def compose(self) -> bool:

73 """If true the referenced object compose this hardware object.

75 As result if this object is registered in Daiquiri, the referenced

76 object will be registered too.

77 """

78 return self._compose

80 def _device_to_name(self, device):

81 if device is None:

82 name = ""

83 elif isinstance(device, str):

84 name = device

85 else:

86 name = device.name

87 return "hardware:" + name

89 def translate_from(self, value):

90 if value is None:

91 return []

92 return [self._device_to_name(v) for v in value]

95class EnumProperty(HardwareProperty):

96 """Attribute read from BLISS as a python enum and exposed to Daiquiri as a

97 name

98 """

100 def __init__(self, name: str, enum_type, getter: callable = None):

101 HardwareProperty.__init__(self, name, getter=getter)

102 self.__enum_type = enum_type

103

104 def translate_from(self, value):

105 if isinstance(value, str):

106 value = self.__enum_type(value)

107 state = value.name

108 return state.upper()

109

110

111class CouldBeNotImplementedProperty(HardwareProperty):

112 """Attribute which can not be implemented in the hardware object.

113

114 In this case a default value is returned.

115 name

116 """

117

118 def notImplementedValue(self):

119 """Returned value when the property is not implemented."""

120 return None

121

122

123class BlissObject(MappedHardwareObject):

124 _protocol = "bliss"

125 _online = True

126

127 def __init__(self, obj=None, **kwargs):

128 super().__init__(**kwargs)

129

130 if obj is None:

131 c = get_config()

132 self._object = c.get(kwargs["address"])

133 else:

134 self._object = obj

135

136 aliases = global_map.aliases

137 self._alias = aliases.get_alias(obj)

138

139 user_tags = []

140 if hasattr(obj, "config"):

141 config = obj.config

142 tags = config.get("user_tag")

143 if tags is not None:

144 if isinstance(tags, str):

145 user_tags.append(tags)

146 elif isinstance(tags, MutableSequence):

147 for tag in tags:

148 user_tags.append(tag)

149 else:

150 raise ValueError("Unsupported BLISS tag from object %s", obj.name)

151 self._user_tags = user_tags

152

153 logger.debug("Connecting to object %s", self._object.name)

154 for name, prop in self._property_map.items():

155 logger.debug(" - Property %s", name)

156 self._connect_event(prop)

157

158 def _connect_event(self, prop: HardwareProperty):

159 if isinstance(prop, AbstractHardwareProperty2):

160 prop.connect_hardware(self._object)

161 else:

162 event.connect(self._object, prop.name, self._event)

163

164 def __repr__(self) -> str:

165 return f"<Bliss: {self.name()} ({self.__class__.__name__}/{self._object.__class__.__name__})>"

166

167 def _event(self, value, *args, signal=None, **kwargs):

168 for name, prop in self._property_map.items():

169 if signal == prop.name:

170 self._update(name, prop, value)

171 break

172

173 def _do_set(self, prop: HardwareProperty, value):

174 obj = self._object

175 if isinstance(prop, AbstractHardwareProperty2):

176 prop.write_hardware(obj, value)

177 else:

178 return setattr(obj, prop.name, value)

179

180 def _do_get(self, prop: HardwareProperty):

181 obj = self._object

182 if isinstance(prop, AbstractHardwareProperty2):

183 try:

184 return prop.read_hardware(obj)

185 except (NotImplementedError, RuntimeError):

186 logger.info(

187 f"Could not get property {prop.name} from {self.name()}",

188 exc_info=True,

189 )

190 return None

191

192 getter = prop.getter

193 if getter is not None:

194 return getter(self)

195 try:

196 try:

197 return get_nested_attr(obj, prop.name)

198 except NotImplementedError:

199 if isinstance(prop, CouldBeNotImplementedProperty):

200 return prop.notImplementedValue()

201 raise

202 except (NotImplementedError, RuntimeError):

203 logger.info(

204 f"Could not get property {prop.name} from {self.name()}", exc_info=True

205 )

206 return None

207

208 def _do_call(self, function, value, **kwargs):

209 fn = getattr(self._object, self._callable_map[function])

210 if value is None:

211 return fn(**kwargs)

212 else:

213 return fn(value, **kwargs)

214

215 def _create_config_from_ref_property(self, name):

216 obj_name = self.get(name)

217 if not obj_name.startswith("hardware:"):

218 return None

219 obj_name = obj_name[len("hardware:") :]

220 if obj_name == "":

221 return None

222 return {

223 "protocol": "bliss",

224 "id": obj_name,

225 "name": obj_name,

226 "address": obj_name,

227 }

228

229 def _create_config_from_ref_list_property(self, name):

230 obj_names = self.get(name)

231 obj_names = [

232 n[len("hardware:") :] for n in obj_names if n.startswith("hardware:")

233 ]

234 obj_names = [n for n in obj_names if n != ""]

235 return [

236 {

237 "protocol": "bliss",

238 "id": obj_name,

239 "name": obj_name,

240 "address": obj_name,

241 }

242 for obj_name in obj_names

243 ]

244

245 def get_subobject_configs(self):

246 """

247 Create a list of configuration for each object reference composition

248 this object.

249

250 It is based on properties `ObjectRefProperty` and `ObjectRefListProperty`

251 with the `compose` attribute to true.

252 """

253 configs = []

254 for name, prop in self._property_map.items():

255 if isinstance(prop, ObjectRefListProperty) and prop.compose:

256 configs.extend(self._create_config_from_ref_list_property(name))

257 if isinstance(prop, ObjectRefProperty) and prop.compose:

258 configs.append(self._create_config_from_ref_property(name))

259

260 return [c for c in configs if c is not None]

261

262

263class BlissDummyObject(HardwareObject):

264 """Dummy Bliss Object

265

266 Used when an object cannot be retrieved from Beacon

267 """

268

269 _type = "unknown"

270 _protocol = "bliss"

271 _online = False

272

273 def _call(self, *args, **kwargs):

274 pass

275

276 def _get(self, *args, **kwargs):

277 pass

278

279 def _set(self, *args, **kwargs):

280 pass