Coverage for aiocoap/optiontypes.py: 90%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

103 statements  

1# This file is part of the Python aiocoap library project. 

2# 

3# Copyright (c) 2012-2014 Maciej Wasilak <http://sixpinetrees.blogspot.com/>, 

4# 2013-2014 Christian Amsüss <c.amsuess@energyharvesting.at> 

5# 

6# aiocoap is free software, this file is published under the MIT license as 

7# described in the accompanying LICENSE file. 

8 

9import abc 

10import collections 

11 

12from .numbers.contentformat import ContentFormat 

13 

14def _to_minimum_bytes(value): 

15 return value.to_bytes((value.bit_length() + 7) // 8, 'big') 

16 

17class OptionType(metaclass=abc.ABCMeta): 

18 """Interface for decoding and encoding option values 

19 

20 Instances of :class:`OptionType` are collected in a list in a 

21 :attr:`.Message.opt` :class:`.Options` object, and provide a translation 

22 between the CoAP octet-stream (accessed using the 

23 :meth:`encode()`/:meth:`decode()` method pair) and the interpreted value 

24 (accessed via the :attr:`value` attribute). 

25 

26 Note that OptionType objects usually don't need to be handled by library 

27 users; the recommended way to read and set options is via the Options 

28 object'sproperties (eg. ``message.opt.uri_path = ('.well-known', 

29 'core')``).""" 

30 

31 @abc.abstractmethod 

32 def __init__(self, number, value): 

33 """Set the `self.name` and `self.value` attributes""" 

34 

35 @abc.abstractmethod 

36 def encode(self): 

37 """Return the option's value in serialzied form""" 

38 

39 @abc.abstractmethod 

40 def decode(self, rawdata): 

41 """Set the option's value from the bytes in rawdata""" 

42 

43class StringOption(OptionType): 

44 """String CoAP option - used to represent string options. Always encoded in 

45 UTF8 per CoAP specification.""" 

46 

47 def __init__(self, number, value=""): 

48 self.value = value 

49 self.number = number 

50 

51 def encode(self): 

52 # FIXME: actually, this should be utf8 of the net-unicode form (maybe it is) 

53 rawdata = self.value.encode('utf-8') 

54 return rawdata 

55 

56 def decode(self, rawdata): 

57 self.value = rawdata.decode('utf-8') 

58 

59 def __str__(self): 

60 return self.value 

61 

62class OpaqueOption(OptionType): 

63 """Opaque CoAP option - used to represent options that just have their 

64 uninterpreted bytes as value.""" 

65 

66 def __init__(self, number, value=b""): 

67 self.value = value 

68 self.number = number 

69 

70 def encode(self): 

71 rawdata = self.value 

72 return rawdata 

73 

74 def decode(self, rawdata): 

75 self.value = rawdata 

76 

77 def __str__(self): 

78 return repr(self.value) 

79 

80class UintOption(OptionType): 

81 """Uint CoAP option - used to represent integer options.""" 

82 

83 def __init__(self, number, value=0): 

84 self.value = value 

85 self.number = number 

86 

87 def encode(self): 

88 return _to_minimum_bytes(int(self.value)) 

89 

90 def decode(self, rawdata): 

91 self.value = int.from_bytes(rawdata, 'big') 

92 

93 def __str__(self): 

94 return str(self.value) 

95 

96class TypedOption(OptionType, metaclass=abc.ABCMeta): 

97 @property 

98 @abc.abstractmethod 

99 def type(self) -> type: 

100 """Checked type of the option""" 

101 

102 def __init__(self, number, value=None): 

103 self.number = number 

104 # FIXME when is this ever initialized without value? 

105 if value is not None: 

106 self.value = value 

107 

108 value = property(lambda self: self._value, lambda self, value: self._set_from_opt_value(value)) 

109 

110 def _set_from_opt_value(self, value: object): 

111 """Convert a value set as ``message.opt.option_name = value`` into the 

112 stored value. By default, this does an eager isinstance check on the 

113 value (anticipating that encoding an unsuitable value would otherwise 

114 fail at a hard-to-debug location).""" 

115 if not isinstance(value, self.type): 

116 raise ValueError("Setting values of type %s is not supported on this option" % type(value)) 

117 self._value = value 

118 

119 def __str__(self): 

120 return str(self.value) 

121 

122class BlockOption(TypedOption): 

123 """Block CoAP option - special option used only for Block1 and Block2 options. 

124 Currently it is the only type of CoAP options that has 

125 internal structure. 

126 

127 That structure (BlockwiseTuple) covers not only the block options of 

128 RFC7959, but also the BERT extension of RFC8323. If the reserved size 

129 exponent 7 is used for purposes incompatible with BERT, the implementor 

130 might want to look at the context dependent option number 

131 interpretations which will hopefully be in place for Signaling (7.xx) 

132 messages by then.""" 

133 class BlockwiseTuple(collections.namedtuple('_BlockwiseTuple', ['block_number', 'more', 'size_exponent'])): 

134 @property 

135 def size(self): 

136 return 2 ** (min(self.size_exponent, 6) + 4) 

137 

138 @property 

139 def start(self): 

140 """The byte offset in the body indicated by block number and size. 

141 

142 Note that this calculation is only valid for descriptive use and 

143 Block2 control use. The semantics of block_number and size in 

144 Block1 control use are unrelated (indicating the acknowledged block 

145 number in the request Block1 size and the server's preferred block 

146 size), and must not be calculated using this property in that 

147 case.""" 

148 return self.block_number * self.size 

149 

150 @property 

151 def is_bert(self): 

152 """True if the exponent is recognized to signal a BERT message.""" 

153 return self.size_exponent == 7 

154 

155 def is_valid_for_payload_size(self, payloadsize): 

156 if self.is_bert: 

157 if self.more: 

158 return payloadsize % 1024 == 0 

159 return True 

160 else: 

161 if self.more: 

162 return payloadsize == self.size 

163 else: 

164 return payloadsize <= self.size 

165 

166 def reduced_to(self, maximum_exponent): 

167 """Return a BlockwiseTuple whose exponent is capped to the given 

168 maximum_exponent 

169 

170 >>> initial = BlockOption.BlockwiseTuple(10, 0, 5) 

171 >>> initial == initial.reduced_to(6) 

172 True 

173 >>> initial.reduced_to(3) 

174 BlockwiseTuple(block_number=40, more=0, size_exponent=3) 

175 """ 

176 if maximum_exponent >= self.size_exponent: 

177 return self 

178 if maximum_exponent == 6 and self.size_exponent == 7: 

179 return (self.block_number, self.more, 6) 

180 increasednumber = self.block_number << (min(self.size_exponent, 6) - maximum_exponent) 

181 return type(self)(increasednumber, self.more, maximum_exponent) 

182 

183 type = BlockwiseTuple 

184 

185 def encode(self): 

186 as_integer = (self.value.block_number << 4) + (self.value.more * 0x08) + self.value.size_exponent 

187 return _to_minimum_bytes(as_integer) 

188 

189 def decode(self, rawdata): 

190 as_integer = int.from_bytes(rawdata, 'big') 

191 self.value = self.BlockwiseTuple(block_number=(as_integer >> 4), more=bool(as_integer & 0x08), size_exponent=(as_integer & 0x07)) 

192 

193 def _set_from_opt_value(self, value): 

194 # Casting it through the constructor makes it easy to set the option as 

195 # `(num, more, sz)` without having to pick the type out of a very long 

196 # module name 

197 super()._set_from_opt_value(self.type(*value)) 

198 

199class ContentFormatOption(TypedOption): 

200 """Type of numeric options whose number has :class:`ContentFormat` 

201 semantics""" 

202 

203 type = ContentFormat 

204 

205 def encode(self): 

206 return _to_minimum_bytes(int(self.value)) 

207 

208 def decode(self, rawdata): 

209 as_integer = int.from_bytes(rawdata, 'big') 

210 self._value = ContentFormat(as_integer) 

211 

212 def _set_from_opt_value(self, value): 

213 super()._set_from_opt_value(ContentFormat(value))