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

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 if isinstance(prop, AbstractHardwareProperty2):

157 prop.connect_hardware(self._object)

158 else:

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

160

161 def __repr__(self) -> str:

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

163

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

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

166 if signal == prop.name:

167 self._update(name, prop, value)

168 break

169

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

171 obj = self._object

172 if isinstance(prop, AbstractHardwareProperty2):

173 prop.write_hardware(obj, value)

174 else:

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

176

177 def _do_get(self, prop: HardwareProperty):

178 obj = self._object

179 if isinstance(prop, AbstractHardwareProperty2):

180 try:

181 return prop.read_hardware(obj)

182 except (NotImplementedError, RuntimeError):

183 logger.info(

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

185 exc_info=True,

186 )

187 return None

188

189 getter = prop.getter

190 if getter is not None:

191 return getter(self)

192 try:

193 try:

194 return get_nested_attr(obj, prop.name)

195 except NotImplementedError:

196 if isinstance(prop, CouldBeNotImplementedProperty):

197 return prop.notImplementedValue()

198 raise

199 except (NotImplementedError, RuntimeError):

200 logger.info(

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

202 )

203 return None

204

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

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

207 if value is None:

208 return fn(**kwargs)

209 else:

210 return fn(value, **kwargs)

211

212 def _create_config_from_ref_property(self, name):

213 obj_name = self.get(name)

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

215 return None

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

217 if obj_name == "":

218 return None

219 return {

220 "protocol": "bliss",

221 "id": obj_name,

222 "name": obj_name,

223 "address": obj_name,

224 }

225

226 def _create_config_from_ref_list_property(self, name):

227 obj_names = self.get(name)

228 obj_names = [

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

230 ]

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

232 return [

233 {

234 "protocol": "bliss",

235 "id": obj_name,

236 "name": obj_name,

237 "address": obj_name,

238 }

239 for obj_name in obj_names

240 ]

241

242 def get_subobject_configs(self):

243 """

244 Create a list of configuration for each object reference composition

245 this object.

246

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

248 with the `compose` attribute to true.

249 """

250 configs = []

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

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

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

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

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

256

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

258

259

260class BlissDummyObject(HardwareObject):

261 """Dummy Bliss Object

262

263 Used when an object cannot be retrieved from Beacon

264 """

265

266 _type = "unknown"

267 _protocol = "bliss"

268 _online = False

269

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

271 pass

272

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

274 pass

275

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

277 pass