Coverage for /home/runner/work/zserio/zserio/compiler/extensions/python/runtime/src/zserio/creator.py: 100%

1""" 

2The module implements creator of zserio object tree based on type info. 

3""" 

4 

5import typing 

6import enum 

7 

8from zserio.typeinfo import ( 

9 TypeInfo, 

10 RecursiveTypeInfo, 

11 TypeAttribute, 

12 MemberInfo, 

13 MemberAttribute, 

14) 

15from zserio.exception import PythonRuntimeException 

16 

17 

18class ZserioTreeCreator: 

19 """ 

20 Allows to build zserio object tree defined by the given type info. 

21 """ 

22 

23 def __init__( 

24 self, 

25 type_info: typing.Union[TypeInfo, RecursiveTypeInfo], 

26 *arguments: typing.List[typing.Any], 

27 ) -> None: 

28 """ 

29 Constructor. 

30 

31 :param type_info: Type info defining the tree. 

32 :param arguments: Arguments for type which defines the tree. 

33 """ 

34 

35 self._root_type_info = type_info 

36 self._root_arguments = arguments 

37 self._member_info_stack: typing.List[MemberInfo] = [] 

38 self._value_stack: typing.List[typing.Any] = [] 

39 self._state = ZserioTreeCreator._State.BEFORE_ROOT 

40 

41 def begin_root(self) -> None: 

42 """ 

43 Creates the top level compound element and move to state of building its children. 

44 """ 

45 

46 if self._state != ZserioTreeCreator._State.BEFORE_ROOT: 

47 raise PythonRuntimeException(f"ZserioTreeCreator: Cannot begin root in state '{self._state}'!") 

48 

49 self._value_stack.append(self._root_type_info.py_type(*self._root_arguments)) 

50 self._state = ZserioTreeCreator._State.IN_COMPOUND 

51 

52 def end_root(self) -> typing.Any: 

53 """ 

54 Finishes building and returns the created tree. 

55 

56 :returns: Zserio object tree. 

57 :raises PythonRuntimeException: When the creator is not in state of building the root object. 

58 """ 

59 

60 if self._state != ZserioTreeCreator._State.IN_COMPOUND or len(self._value_stack) != 1: 

61 raise PythonRuntimeException(f"ZserioTreeCreator: Cannot end root in state '{self._state}'!") 

62 

63 self._state = ZserioTreeCreator._State.BEFORE_ROOT 

64 return self._value_stack.pop() 

65 

66 def begin_array(self, name: str) -> None: 

67 """ 

68 Creates an array field within the current compound. 

69 

70 :param name: Name of the array field. 

71 :raises PythonRuntimeException: When the field doesn't exist or when the creator is not in a compound. 

72 """ 

73 

74 if self._state != ZserioTreeCreator._State.IN_COMPOUND: 

75 raise PythonRuntimeException(f"ZserioTreeCreator: Cannot begin array in state '{self._state}'!") 

76 

77 member_info = self._find_member_info(self._get_type_info(), name) 

78 if not MemberAttribute.ARRAY_LENGTH in member_info.attributes: 

79 raise PythonRuntimeException( 

80 "ZserioTreeCreator: Field " f"'{member_info.schema_name}' is not an array!" 

81 ) 

82 

83 self._member_info_stack.append(member_info) 

84 self._value_stack.append([]) 

85 self._state = ZserioTreeCreator._State.IN_ARRAY 

86 

87 def end_array(self): 

88 """ 

89 Finishes the array field. 

90 

91 :raises PythonRuntimeException: When the creator is not in an array field. 

92 """ 

93 if self._state != ZserioTreeCreator._State.IN_ARRAY: 

94 raise PythonRuntimeException(f"ZserioTreeCreator: Cannot end array in state '{self._state}'!") 

95 

96 member_info = self._member_info_stack.pop() 

97 value = self._value_stack.pop() 

98 setattr( 

99 self._value_stack[-1], 

100 member_info.attributes[MemberAttribute.PROPERTY_NAME], 

101 value, 

102 ) 

103 self._state = ZserioTreeCreator._State.IN_COMPOUND 

104 

105 def begin_compound(self, name): 

106 """ 

107 Creates a compound field within the current compound. 

108 

109 :param name: Name of the compound field. 

110 :raises PythonRuntimeException: When the field doesn't exist or when the creator is not in a compound. 

111 """ 

112 

113 if self._state != ZserioTreeCreator._State.IN_COMPOUND: 

114 raise PythonRuntimeException(f"ZserioTreeCreator: Cannot begin compound in state '{self._state}'!") 

115 

116 member_info = self._find_member_info(self._get_type_info(), name) 

117 

118 if MemberAttribute.ARRAY_LENGTH in member_info.attributes: 

119 raise PythonRuntimeException( 

120 "ZserioTreeCreator: Field " f"'{member_info.schema_name}' is an array!" 

121 ) 

122 

123 if not TypeAttribute.FIELDS in member_info.type_info.attributes: 

124 raise PythonRuntimeException( 

125 f"ZserioTreeCreator: Field " f"'{member_info.schema_name}' is not a compound!" 

126 ) 

127 

128 compound = self._create_object(member_info, self._value_stack[-1]) 

129 

130 self._member_info_stack.append(member_info) 

131 self._value_stack.append(compound) 

132 self._state = ZserioTreeCreator._State.IN_COMPOUND 

133 

134 def end_compound(self): 

135 """ 

136 Finishes the compound. 

137 

138 :raises PythonRuntimeException: When the creator is not in a compound field. 

139 """ 

140 

141 if self._state != ZserioTreeCreator._State.IN_COMPOUND or not self._member_info_stack: 

142 raise PythonRuntimeException( 

143 f"ZserioTreeCreator: Cannot end compound in state '{self._state}'" 

144 + (", expecting end_root!" if not self._member_info_stack else "!") 

145 ) 

146 

147 if MemberAttribute.ARRAY_LENGTH in self._member_info_stack[-1].attributes: 

148 raise PythonRuntimeException("ZserioTreeCreator: Cannot end compound, it's an array element!") 

149 

150 member_info = self._member_info_stack.pop() 

151 value = self._value_stack.pop() 

152 setattr( 

153 self._value_stack[-1], 

154 member_info.attributes[MemberAttribute.PROPERTY_NAME], 

155 value, 

156 ) 

157 

158 def set_value(self, name: str, value: typing.Any): 

159 """ 

160 Sets field value within the current compound. 

161 

162 :param name: Name of the field. 

163 :param value: Value to set. 

164 

165 :raises PythonRuntimeException: When the field doesn't exist or when the creator is not in a compound. 

166 """ 

167 

168 if self._state != ZserioTreeCreator._State.IN_COMPOUND: 

169 raise PythonRuntimeException(f"ZserioTreeCreator: Cannot set value in state '{self._state}'!") 

170 

171 member_info = self._find_member_info(self._get_type_info(), name) 

172 

173 if value is not None: 

174 if MemberAttribute.ARRAY_LENGTH in member_info.attributes: 

175 raise PythonRuntimeException( 

176 "ZserioTreeCreator: Expecting array in field " f"'{member_info.schema_name}'!" 

177 ) 

178 

179 if not isinstance(value, member_info.type_info.py_type): 

180 raise PythonRuntimeException( 

181 f"ZserioTreeCreator: Unexpected value type '{type(value)}', " 

182 f"expecting '{member_info.type_info.py_type}'!" 

183 ) 

184 

185 setattr( 

186 self._value_stack[-1], 

187 member_info.attributes[MemberAttribute.PROPERTY_NAME], 

188 value, 

189 ) 

190 

191 def get_field_type(self, name: str) -> typing.Union[TypeInfo, RecursiveTypeInfo]: 

192 """ 

193 Gets type info of the expected field. 

194 

195 :param name: Field name. 

196 :returns: Type info of the expected field. 

197 :raises PythonRuntimeException: When the creator is not in a compound. 

198 """ 

199 

200 if self._state != ZserioTreeCreator._State.IN_COMPOUND: 

201 raise PythonRuntimeException(f"ZserioTreeCreator: Cannot get field type in state '{self._state}'!") 

202 

203 member_info = self._find_member_info(self._get_type_info(), name) 

204 return member_info.type_info 

205 

206 def begin_compound_element(self): 

207 """ 

208 Creates compound array element within the current array. 

209 

210 :raises PythonRuntimeException: When the creator is not in an array of compounds. 

211 """ 

212 

213 if self._state != ZserioTreeCreator._State.IN_ARRAY: 

214 raise PythonRuntimeException( 

215 "ZserioTreeCreator: Cannot begin compound element in state " f"'{self._state}'!" 

216 ) 

217 

218 member_info = self._member_info_stack[-1] 

219 

220 if not TypeAttribute.FIELDS in member_info.type_info.attributes: 

221 raise PythonRuntimeException( 

222 f"ZserioTreeCreator: Field " f"'{self._member_info_stack[-1].schema_name}' is not a compound!" 

223 ) 

224 

225 compound = self._create_object(member_info, self._value_stack[-2], len(self._value_stack[-1])) 

226 

227 self._value_stack.append(compound) 

228 self._state = ZserioTreeCreator._State.IN_COMPOUND 

229 

230 def end_compound_element(self) -> None: 

231 """ 

232 Finishes the compound element. 

233 

234 :raises PythonRuntimeException: When the creator is not in a compound element. 

235 """ 

236 

237 if self._state != ZserioTreeCreator._State.IN_COMPOUND or not self._member_info_stack: 

238 raise PythonRuntimeException( 

239 "ZserioTreeCreator: Cannot end compound element in state " 

240 f"'{self._state}'" + (", expecting end_root!" if not self._member_info_stack else "!") 

241 ) 

242 

243 if not MemberAttribute.ARRAY_LENGTH in self._member_info_stack[-1].attributes: 

244 raise PythonRuntimeException("ZserioTreeCreator: Cannot end compound element, not in array!") 

245 

246 value = self._value_stack.pop() 

247 self._value_stack[-1].append(value) 

248 self._state = ZserioTreeCreator._State.IN_ARRAY 

249 

250 def add_value_element(self, value: typing.Any) -> None: 

251 """ 

252 Adds the value to the array. 

253 

254 :param value: Value to add. 

255 :raises PythonRuntimeException: When the creator is not in an array of simple values. 

256 """ 

257 

258 if self._state != ZserioTreeCreator._State.IN_ARRAY: 

259 raise PythonRuntimeException( 

260 "ZserioTreeCreator: Cannot add value element in state " f"'{self._state}'!" 

261 ) 

262 

263 element_type_info = self._member_info_stack[-1].type_info 

264 if value is not None and not isinstance(value, element_type_info.py_type): 

265 raise PythonRuntimeException( 

266 f"ZserioTreeCreator: Unexpected value type '{type(value)}', expecting " 

267 f"'{element_type_info.py_type}'!" 

268 ) 

269 

270 self._value_stack[-1].append(value) 

271 

272 def get_element_type(self) -> typing.Union[TypeInfo, RecursiveTypeInfo]: 

273 """ 

274 Gets type info of the expected array element. 

275 

276 :returns: Type info of the expected array element. 

277 :raises PythonRuntimeException: When the creator is not in an array. 

278 """ 

279 

280 if self._state != ZserioTreeCreator._State.IN_ARRAY: 

281 raise PythonRuntimeException("ZserioTreeCreator: Cannot get element type in state '{self._state}'!") 

282 

283 member_info = self._member_info_stack[-1] 

284 return member_info.type_info 

285 

286 def _get_type_info(self) -> typing.Union[TypeInfo, RecursiveTypeInfo]: 

287 return self._member_info_stack[-1].type_info if self._member_info_stack else self._root_type_info 

288 

289 @staticmethod 

290 def _find_member_info(type_info: typing.Union[TypeInfo, RecursiveTypeInfo], name: str) -> MemberInfo: 

291 members = type_info.attributes[TypeAttribute.FIELDS] 

292 for member in members: 

293 if member.schema_name == name: 

294 return member 

295 raise PythonRuntimeException( 

296 f"ZserioTreeCreator: Field '{name}' not found in " f"'{type_info.schema_name}'!" 

297 ) 

298 

299 @staticmethod 

300 def _create_object( 

301 member_info: MemberInfo, 

302 parent: typing.Any, 

303 element_index: typing.Optional[int] = None, 

304 ) -> typing.Any: 

305 args = [] 

306 if MemberAttribute.TYPE_ARGUMENTS in member_info.attributes: 

307 for argument_lambda in member_info.attributes[MemberAttribute.TYPE_ARGUMENTS]: 

308 args.append(argument_lambda(parent, element_index)) 

309 

310 return member_info.type_info.py_type(*args) 

311 

312 class _State(enum.Enum): 

313 BEFORE_ROOT = enum.auto() 

314 IN_COMPOUND = enum.auto() 

315 IN_ARRAY = enum.auto()