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

1#!/usr/bin/env python 

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

3from marshmallow import Schema, fields, validates_schema, ValidationError, post_load 

4 

5from daiquiri.core.schema.validators import ValidatedRegexp, Any 

6 

7import logging 

8 

9logger = logging.getLogger(__name__) 

10 

11 

12class HardwareObjectBaseSchema(Schema): 

13 """Hardware as exposed by the server to the web client""" 

14 

15 id = fields.Str() 

16 """Unique identifier for this object""" 

17 

18 protocol = fields.Str() 

19 """Protocol, like 'bliss', 'tango'...""" 

20 

21 type = fields.Str() 

22 """Normalized daiquiri hardware type""" 

23 

24 online = fields.Bool() 

25 """True if the hardware is online""" 

26 

27 errors = fields.List(fields.Dict()) 

28 """A list of errors collected during the execution of the hardware""" 

29 

30 name = fields.Str() 

31 """A fullname which identify the hardware for users""" 

32 

33 alias = fields.Str() 

34 """A short name which can be used to identify the hardware for users""" 

35 

36 require_staff = fields.Bool() 

37 """True if any action on the hardware requires staff privilege""" 

38 

39 callables = fields.List(fields.Str()) 

40 """Available commands provided by the hardware""" 

41 

42 properties = fields.Dict() 

43 """Exposed state of the hardware""" 

44 

45 user_tags = fields.List(fields.Str()) 

46 """List of user tags in which the hardware is categorized""" 

47 

48 locked = fields.Str(allow_none=True) 

49 """If not None, the device is locked for this reason, and no actions can be requested.""" 

50 

51 

52class SetObjectProperty(Schema): 

53 property = ValidatedRegexp("word", required=True) 

54 value = Any(required=True, allow_none=True) 

55 

56 

57class CallObjectFunction(Schema): 

58 function = ValidatedRegexp("word", required=True) 

59 value = Any() 

60 

61 

62class HardwareGroupSchema(Schema): 

63 groupid = fields.Str(required=True) 

64 name = fields.Str(required=True) 

65 description = fields.Str() 

66 objects = fields.List(fields.Str(), required=True) 

67 state = fields.Bool() 

68 

69 

70class HardwareTypeSchema(Schema): 

71 type = fields.Str() 

72 schema = fields.Str() 

73 

74 

75class HOConfigAttribute(Schema): 

76 """The Hardware Object Config Attribute Schema""" 

77 

78 id = fields.Str(required=True, metadata={"description": "Attribute id"}) 

79 name = fields.Str(metadata={"description": "Attribute name (can be customised)"}) 

80 ui_schema = fields.Dict( 

81 metadata={"description": "Define how the UI renders this attribute"} 

82 ) 

83 type = fields.Str( 

84 metadata={ 

85 "enum": ["float", "int", "bool", "str"], 

86 "description": "Attribute type", 

87 } 

88 ) 

89 step = fields.Float(metadata={"description": "Step size for attribute"}) 

90 min = fields.Float(metadata={"description": "Minimum value for attribute"}) 

91 max = fields.Float(metadata={"description": "Maximum value for attribute"}) 

92 

93 

94class HOConfigSchema(Schema): 

95 """HardwareObject base configuration schema""" 

96 

97 name = fields.Str(metadata={"description": "Object name"}) 

98 id = fields.Str(metadata={"description": "Object id"}) 

99 auto_id = fields.Bool( 

100 metadata={"description": "Whether the id was automatically generated"} 

101 ) 

102 protocol = fields.Str(metadata={"description": "Protocol handler to use"}) 

103 url = fields.Str( 

104 metadata={"description": "Url of the device including which protocol to use"} 

105 ) 

106 require_staff = fields.Bool( 

107 metadata={"description": "Whether this object requires staff to modify"} 

108 ) 

109 attributes = fields.Nested( 

110 HOConfigAttribute, 

111 many=True, 

112 metadata={"description": "Attribute configuration for run time schemas"}, 

113 ) 

114 

115 @validates_schema 

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

117 if not (data.get("protocol") or data.get("url")): 

118 raise ValidationError( 

119 "Object must have either a `protocol` or `url` defined" 

120 ) 

121 

122 @post_load 

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

124 if data.get("url"): 

125 protocol, rest = data["url"].split("://") 

126 data["protocol"] = protocol 

127 

128 if not data.get("id"): 

129 parts = rest.split("/") 

130 data["id"] = parts[-1] 

131 data["auto_id"] = True 

132 

133 if not data.get("name"): 

134 data["name"] = data.get("id") 

135 

136 return data 

137 

138 

139class HardwareSchema(Schema): 

140 def read_only(self, prop): 

141 return self.fields[prop].metadata.get("readOnly", False) 

142 

143 def __contains__(self, val): 

144 return val in self.fields 

145 

146 def __iter__(self): 

147 self.__iter = iter(self.fields.keys()) 

148 return self.__iter 

149 

150 def __next__(self): 

151 return next(self.__iter) 

152 

153 def validate(self, prop, value): 

154 data = {} 

155 data[prop] = value 

156 valid = self.load(data) 

157 

158 return valid[prop]