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

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 protocol = fields.Str() 

17 type = fields.Str() 

18 online = fields.Bool() 

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

20 name = fields.Str() 

21 alias = fields.Str() 

22 require_staff = fields.Bool() 

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

24 properties = fields.Dict() 

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

26 

27 

28class SetObjectProperty(Schema): 

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

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

31 

32 

33class CallObjectFunction(Schema): 

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

35 value = Any() 

36 

37 

38class HardwareGroupSchema(Schema): 

39 groupid = fields.Str(required=True) 

40 name = fields.Str(required=True) 

41 description = fields.Str() 

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

43 state = fields.Bool() 

44 

45 

46class HardwareTypeSchema(Schema): 

47 type = fields.Str() 

48 schema = fields.Str() 

49 

50 

51class HOConfigAttribute(Schema): 

52 """The Hardware Object Config Attribute Schema""" 

53 

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

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

56 ui_schema = fields.Dict( 

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

58 ) 

59 type = fields.Str( 

60 metadata={ 

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

62 "description": "Attribute type", 

63 } 

64 ) 

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

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

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

68 

69 

70class HOConfigSchema(Schema): 

71 """HardwareObject base configuration schema""" 

72 

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

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

75 auto_id = fields.Bool( 

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

77 ) 

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

79 url = fields.Str( 

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

81 ) 

82 require_staff = fields.Bool( 

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

84 ) 

85 attributes = fields.Nested( 

86 HOConfigAttribute, 

87 many=True, 

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

89 ) 

90 

91 @validates_schema 

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

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

94 raise ValidationError( 

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

96 ) 

97 

98 @post_load 

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

100 if data.get("url"): 

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

102 data["protocol"] = protocol 

103 

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

105 parts = rest.split("/") 

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

107 data["auto_id"] = True 

108 

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

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

111 

112 return data 

113 

114 

115class HardwareSchema(Schema): 

116 def read_only(self, prop): 

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

118 

119 def __contains__(self, val): 

120 return val in self.fields 

121 

122 def __iter__(self): 

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

124 return self.__iter 

125 

126 def __next__(self): 

127 return next(self.__iter) 

128 

129 def validate(self, prop, value): 

130 data = {} 

131 data[prop] = value 

132 valid = self.load(data) 

133 

134 return valid[prop]