Coverage for src/galaxy_beep_frame/converter/parser.py: 100%

1"""Parser class.""" 

2 

3from typing import TYPE_CHECKING 

4 

5from galaxy_beep_frame.converter.token import TokenType 

6from galaxy_beep_frame.exception import ( 

7 InvalidTokenTypeError, 

8 NodeCountError, 

9 NodeNoneError, 

10 TokenMutableError, 

11 UnknownContinuationIndicatorError, 

12 UnknownMsgTypeError, 

13) 

14from galaxy_beep_frame.frame import Frame 

15from galaxy_beep_frame.frame.header_type.continuation_indicator import ( 

16 ContinuationIndicator, 

17) 

18from galaxy_beep_frame.header_type import PossibleHeaderType 

19 

20from .token import Token 

21 

22if TYPE_CHECKING: 

23 from galaxy_graph.node import Node 

24 

25 

26class Parser: 

27 """Convert a graph of token generated by lexer and validate it.""" 

28 

29 @staticmethod 

30 def _get_answer_number( 

31 header: Node[Token], 

32 message_type: PossibleHeaderType, 

33 ) -> int | None: 

34 """ 

35 Validate if the answer number should exists in a frame. 

36 

37 Answer number must exist if message type is "ANS". Else, the 

38 answer must not exist. 

39 

40 Parameters 

41 ---------- 

42 header: Node[Token] 

43 Header to check. 

44 message_type: MessageType 

45 Message type. 

46 

47 Returns 

48 ------- 

49 int | None: Answer number if it exists. 

50 

51 """ 

52 

53 results = header.get_outgoing_neighbors_by_connection_data( 

54 lambda data: data == "has answer number" 

55 ) 

56 

57 if message_type is PossibleHeaderType.ANS: 

58 if len(results) != 1: 

59 raise NodeCountError(len(results), "answer number") 

60 answer_number = next(iter(results)).data 

61 

62 if not isinstance(answer_number, Token): 

63 raise NodeNoneError 

64 if answer_number.type is not TokenType.ANSWER_NUMBER: 

65 raise InvalidTokenTypeError( 

66 answer_number.type, TokenType.ANSWER_NUMBER 

67 ) 

68 if answer_number.mutability.is_mutable(): 

69 raise TokenMutableError(answer_number.type) 

70 return int(answer_number.value.decode()) 

71 

72 if len(results) != 0: 

73 raise NodeCountError(len(results), "answer number", 0) 

74 return None 

75 

76 @staticmethod 

77 def _get_channel_number(header: Node[Token]) -> int: 

78 """ 

79 Validate and get the channel number. 

80 

81 Parameters 

82 ---------- 

83 header: Node[Token] 

84 Header to check. 

85 

86 Returns 

87 ------- 

88 int: channel number of the frame 

89 

90 """ 

91 results = header.get_outgoing_neighbors_by_connection_data( 

92 lambda data: data == "has channel number" 

93 ) 

94 

95 if len(results) != 1: 

96 raise NodeCountError(len(results), "channel number") 

97 channel_number = next(iter(results)).data 

98 if not isinstance(channel_number, Token): 

99 raise NodeNoneError 

100 if channel_number.type is not TokenType.CHANNEL_NUMBER: 

101 raise InvalidTokenTypeError( 

102 channel_number.type, TokenType.CHANNEL_NUMBER 

103 ) 

104 if channel_number.mutability.is_mutable(): 

105 raise TokenMutableError(channel_number.type) 

106 return int(channel_number.value.decode()) 

107 

108 @staticmethod 

109 def _get_continuation_indicator( 

110 header: Node[Token], 

111 ) -> ContinuationIndicator: 

112 """ 

113 Validate and get the continuation indicator. 

114 

115 Parameters 

116 ---------- 

117 header: Node[Token] 

118 Header to check. 

119 

120 Returns 

121 ------- 

122 ContinuationIndicator: continuation indicator of the frame 

123 

124 """ 

125 results = header.get_outgoing_neighbors_by_connection_data( 

126 lambda data: data == "has continuation indicator" 

127 ) 

128 

129 if len(results) != 1: 

130 raise NodeCountError(len(results), "continuation indicator") 

131 continuation_indicator = next(iter(results)).data 

132 if not isinstance(continuation_indicator, Token): 

133 raise NodeNoneError 

134 if ( 

135 continuation_indicator.type 

136 is not TokenType.CONTINUATION_INDICATOR 

137 ): 

138 raise InvalidTokenTypeError( 

139 continuation_indicator.type, 

140 TokenType.CONTINUATION_INDICATOR, 

141 ) 

142 if continuation_indicator.mutability.is_mutable(): 

143 raise TokenMutableError(continuation_indicator.type) 

144 

145 match continuation_indicator.value: 

146 case b".": 

147 return ContinuationIndicator.COMPLETE 

148 case b"*": 

149 return ContinuationIndicator.INTERMEDIATE 

150 case _: 

151 raise UnknownContinuationIndicatorError( 

152 continuation_indicator.value 

153 ) 

154 

155 @staticmethod 

156 def _get_header(frame: Node[Token]) -> Node[Token]: 

157 """ 

158 Get header of the frame. 

159 

160 Parameters 

161 ---------- 

162 frame: Node[Token] 

163 Retrieved frame. 

164 

165 Returns 

166 ------- 

167 Node[Token]: Frame Node 

168 

169 """ 

170 

171 results = frame.get_outgoing_neighbors_by_connection_data( 

172 lambda data: data == "has header" 

173 ) 

174 

175 if len(results) != 1: 

176 raise NodeCountError(len(results), "header") 

177 header_node = next(iter(results)) 

178 header = header_node.data 

179 if not isinstance(header, Token): 

180 raise NodeNoneError 

181 if header.mutability.is_mutable(): 

182 raise TokenMutableError(header.type) 

183 if header.type is not TokenType.HEADER: 

184 raise InvalidTokenTypeError(header.type, TokenType.HEADER) 

185 return header_node 

186 

187 @staticmethod 

188 def _get_message_number(header: Node[Token]) -> int: 

189 """ 

190 Validate and get the message number. 

191 

192 Parameters 

193 ---------- 

194 header: Node[Token] 

195 Header to check. 

196 

197 Returns 

198 ------- 

199 int: message number of the frame 

200 

201 """ 

202 results = header.get_outgoing_neighbors_by_connection_data( 

203 lambda data: data == "has message number" 

204 ) 

205 

206 if len(results) != 1: 

207 raise NodeCountError(len(results), "message number") 

208 message_number = next(iter(results)).data 

209 if not isinstance(message_number, Token): 

210 raise NodeNoneError 

211 if message_number.type is not TokenType.MESSAGE_NUMBER: 

212 raise InvalidTokenTypeError( 

213 message_number.type, TokenType.MESSAGE_NUMBER 

214 ) 

215 if message_number.mutability.is_mutable(): 

216 raise TokenMutableError(message_number.type) 

217 return int(message_number.value.decode()) 

218 

219 @staticmethod 

220 def _get_message_type(header: Node[Token]) -> PossibleHeaderType: 

221 """ 

222 Get and validate if the message type is valid. 

223 

224 Parameters 

225 ---------- 

226 header: Node[Token] 

227 Header to check. 

228 

229 Returns 

230 ------- 

231 MessageType: Returns the message type of the frame. 

232 

233 """ 

234 results = header.get_outgoing_neighbors_by_connection_data( 

235 lambda data: data == "has message type" 

236 ) 

237 if len(results) != 1: 

238 raise NodeCountError(len(results), "message type") 

239 

240 message_type = next(iter(results)).data 

241 if not isinstance(message_type, Token): 

242 raise NodeNoneError 

243 if message_type.type is not TokenType.MESSAGE_TYPE: 

244 raise InvalidTokenTypeError( 

245 message_type.type, TokenType.MESSAGE_TYPE 

246 ) 

247 if message_type.mutability.is_mutable(): 

248 raise TokenMutableError(message_type.type) 

249 

250 match message_type.value: 

251 case b"ANS": 

252 return PossibleHeaderType.ANS 

253 case b"ERR": 

254 return PossibleHeaderType.ERR 

255 case b"MSG": 

256 return PossibleHeaderType.MSG 

257 case b"NUL": 

258 return PossibleHeaderType.NUL 

259 case b"RPY": 

260 return PossibleHeaderType.RPY 

261 case _: 

262 raise UnknownMsgTypeError(message_type.value) 

263 

264 @staticmethod 

265 def _get_payload(frame: Node[Token]) -> bytes: 

266 """ 

267 Validate and get payload. 

268 

269 Parameters 

270 ---------- 

271 frame: Node[Token] 

272 Frame to check. 

273 

274 Returns 

275 ------- 

276 bytes: Payload of the frame. 

277 

278 """ 

279 results = frame.get_outgoing_neighbors_by_connection_data( 

280 lambda data: data == "has payload" 

281 ) 

282 if len(results) != 1: 

283 raise NodeCountError(len(results), "payload") 

284 

285 payload = next(iter(results)).data 

286 if not isinstance(payload, Token): 

287 raise NodeNoneError 

288 if payload.mutability.is_mutable(): 

289 raise TokenMutableError(payload.type) 

290 if payload.type is not TokenType.PAYLOAD: 

291 raise InvalidTokenTypeError(payload.type, TokenType.PAYLOAD) 

292 

293 return payload.value 

294 

295 @staticmethod 

296 def _get_sequence_number(header: Node[Token]) -> int: 

297 """ 

298 Validate and get sequence number. 

299 

300 Parameters 

301 ---------- 

302 header: Node[Token] 

303 Header to check. 

304 

305 Returns 

306 ------- 

307 int: sequence number of the frame 

308 

309 """ 

310 results = header.get_outgoing_neighbors_by_connection_data( 

311 lambda data: data == "has sequence number" 

312 ) 

313 

314 if len(results) != 1: 

315 raise NodeCountError(len(results), "sequence number") 

316 

317 sequence_number = next(iter(results)).data 

318 if not isinstance(sequence_number, Token): 

319 raise NodeNoneError 

320 if sequence_number.type is not TokenType.SEQUENCE_NUMBER: 

321 raise InvalidTokenTypeError( 

322 sequence_number.type, TokenType.SEQUENCE_NUMBER 

323 ) 

324 if sequence_number.mutability.is_mutable(): 

325 raise TokenMutableError(sequence_number.type) 

326 

327 return int(sequence_number.value.decode()) 

328 

329 @staticmethod 

330 def parse_frame(frame: Node[Token]) -> Frame: 

331 """ 

332 Convert a node from a graph to a Frame. 

333 

334 Parameters 

335 ---------- 

336 frame: Node[Token] 

337 Frame to convert. 

338 

339 """ 

340 

341 frame_token = frame.data 

342 if not isinstance(frame_token, Token): 

343 raise NodeNoneError 

344 if frame_token.mutability.is_mutable(): 

345 raise TokenMutableError(frame_token.type) 

346 if frame_token.type is not TokenType.FRAME: 

347 raise InvalidTokenTypeError( 

348 frame_token.type, TokenType.FRAME 

349 ) 

350 

351 header = Parser._get_header(frame) 

352 message_type = Parser._get_message_type(header) 

353 

354 return Frame( 

355 payload=Parser._get_payload(frame), 

356 channel_number=Parser._get_channel_number(header), 

357 message_number=Parser._get_message_number(header), 

358 sequence_number=Parser._get_sequence_number(header), 

359 answer_number=Parser._get_answer_number( 

360 header, message_type 

361 ), 

362 continuation_indicator=Parser._get_continuation_indicator( 

363 header 

364 ), 

365 header_type=message_type, 

366 ) 

367 

368 @staticmethod 

369 def parse(frames: list[Node[Token]]) -> list[Frame]: 

370 """ 

371 Parse a list of frames. 

372 

373 Parameters 

374 ---------- 

375 frames: list[Node[Token]] 

376 Frames to convert. 

377 

378 """ 

379 return [Parser.parse_frame(frame) for frame in frames]