Coverage for apio/common/proto_util.py: 89%

1# This file is part of the Apio project 

2# (C) 2016-present FPGAwars 

3# Project authors: 

4# * Jesús Arroyo (Jesus89) (2016-2019) 

5# * Juan Gonzalez (obijuan) (2019-2024) 

6# * The Apio Team (2024-present) 

7# License: GPLv2 

8 

9"""Utilities related to the Apio Protocol Buffers objects.""" 

10 

11import re 

12from typing import Any, Dict, TypeVar 

13from google.protobuf.json_format import ParseDict 

14from google.protobuf.message import Message 

15from google.protobuf.unknown_fields import UnknownFieldSet 

16from google.protobuf.json_format import MessageToDict 

17from apio.common.apio_console import fatal_error, cerror 

18 

19# Placeholder for a concrete protobuf message *class* (e.g. MyProto), 

20# not an instance. bound=Message restricts it to subclasses of 

21# google.protobuf.message.Message. A TypeVar (vs plain type[Message]) 

22# lets the type checker keep the specific class: pass MyProto, get MyProto. 

23MessageClass = TypeVar("MessageClass", bound=Message) 

24 

25 

26def is_initialized(proto_msg: Message, use_json_naming: bool = False) -> bool: 

27 """Test if the protocol buffer message is fully initializes. If not, 

28 print an error message and return False, otherwise return True. 

29 """ 

30 

31 assert isinstance(proto_msg, Message), type(proto_msg) 

32 

33 # -- Check 1: All required fields should present. 

34 if not proto_msg.IsInitialized(): 

35 # -- Report the first missing required field. 

36 # missing_field: str = proto_msg.FindInitializationErrors()[0] 

37 find = getattr(proto_msg, "FindInitializationErrors", None) 

38 assert callable(find) 

39 missing_field = str(find()[0]) 

40 if use_json_naming: 40 ↛ 42line 40 didn't jump to line 42 because the condition on line 40 was always true

41 missing_field = missing_field.replace("_", "-") 

42 cerror(f"Missing required field '{missing_field}'") 

43 return False 

44 

45 # -- Check 2: Should not carry unknown fields. 

46 unknown_fields = list(UnknownFieldSet(proto_msg)) 

47 if len(unknown_fields) > 0: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true

48 cerror(f'Unknown fields: {", ".join(unknown_fields)}') 

49 return False 

50 

51 # -- All is good. 

52 return True 

53 

54 

55def check_is_initialized( 

56 proto_msg: Message, error_msg: str, *, use_json_naming: bool = False 

57) -> None: 

58 """Check that a proto message is fully populated""" 

59 if not is_initialized(proto_msg, use_json_naming=use_json_naming): 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 fatal_error(error_msg) 

61 

62 

63def check_is_required(proto_msg: Message, *fields_names: str) -> None: 

64 """Check that all the names are of required fields of the proto 

65 object proto_msg. Names may be nested, e.g. "field1.field2". 

66 """ 

67 assert isinstance(proto_msg, Message), type(proto_msg) 

68 

69 for name in fields_names: 

70 descriptor = proto_msg.DESCRIPTOR 

71 for segment in name.split("."): 

72 field = descriptor.fields_by_name.get(segment) 

73 if field is None: 

74 fatal_error( 

75 f"Field '{name}' is not a field of " 

76 "protocol buffer message " 

77 f"'{proto_msg.DESCRIPTOR.full_name}'" 

78 ) 

79 if not field.is_required: 

80 fatal_error( 

81 f"Field '{name}' of '{proto_msg.DESCRIPTOR.full_name}' " 

82 f"is not required" 

83 ) 

84 descriptor = field.message_type 

85 

86 

87def check_not_required(proto_msg: Message, *fields_names: str) -> None: 

88 """Check that none of the names is a not fully required path of proto_msg. 

89 Names may be nested, e.g. "field1.field2". A path is not required if 

90 any segment along it is not required. So if "b" is optional, "a.b.c" 

91 is not required. Repeated fields are not required since they can have 

92 zero members. 

93 """ 

94 assert isinstance(proto_msg, Message), type(proto_msg) 

95 

96 for name in fields_names: 

97 descriptor = proto_msg.DESCRIPTOR 

98 all_required = True 

99 for segment in name.split("."): 

100 if descriptor is None: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true

101 fatal_error( 

102 f"Field '{name}' is not a field of " 

103 "protocol buffer message " 

104 f"'{proto_msg.DESCRIPTOR.full_name}'" 

105 ) 

106 field = descriptor.fields_by_name.get(segment) 

107 if field is None: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true

108 fatal_error( 

109 f"Field '{name}' is not a field of " 

110 "protocol buffer message " 

111 f"'{proto_msg.DESCRIPTOR.full_name}'" 

112 ) 

113 if not field.is_required: 

114 all_required = False 

115 descriptor = field.message_type 

116 if all_required: 

117 fatal_error( 

118 f"Field '{name}' of '{proto_msg.DESCRIPTOR.full_name}' " 

119 f"is required" 

120 ) 

121 

122 

123def proto_from_json_dict( 

124 json_dict: Dict[str, Any], 

125 proto_class: type[MessageClass], 

126 *, 

127 use_json_naming: bool = True, 

128) -> MessageClass | None: 

129 """Create and return an object of proto message class 'proto_class' 

130 populated with values from json dict json_dict. If error, print an error 

131 message and return None. 

132 """ 

133 # pylint: disable=broad-exception-caught 

134 

135 try: 

136 proto_msg = ParseDict(json_dict, proto_class()) 

137 

138 except Exception as e: 

139 parsing_error = str(e) 

140 

141 # -- Try to improve the error message. 

142 pattern = re.compile(r'has no field named "([^"]+)" at') 

143 match = pattern.search(parsing_error) 

144 if match: 144 ↛ 148line 144 didn't jump to line 148 because the condition on line 144 was always true

145 parsing_error = f"Unknown field '{match.group(1)}'" 

146 

147 # -- Print an error message and exit. 

148 cerror(parsing_error) 

149 return None 

150 

151 if not is_initialized(proto_msg, use_json_naming=use_json_naming): 

152 return None 

153 

154 return proto_msg 

155 

156 

157def proto_to_json_dict(proto_msg: Message) -> Dict[str, Any]: 

158 """Given a proto object, convert it to a json dict.""" 

159 assert isinstance(proto_msg, Message), type(proto_msg) 

160 return MessageToDict(proto_msg)