Coverage for skema/program_analysis/gromet_wire_diagnosis.py: 25%

1import argparse 

2from skema.program_analysis.JSON2GroMEt import json2gromet 

3from skema.gromet.metadata import SourceCodeReference 

4 

5# Ways to expand 

6# Check loop, condition FN indices 

7# Check bf call FN indices 

8# Boxes associated with ports 

9 

10def disp_wire(wire): 

11 return f"src:{wire.src}<-->tgt:{wire.tgt}" 

12 

13def get_length(gromet_item): 

14 # For any gromet object we can generically retrieve the length, since they all exist 

15 # in lists 

16 return len(gromet_item) if gromet_item != None else 0 

17 

18def check_wire(gromet_wire, src_port_count, tgt_port_count, wire_type = "", metadata=None): 

19 # The current wiring checks are 

20 # Checking if the ports on both ends of the wire are below or over the bounds 

21 error_detected = False 

22 if gromet_wire.src < 0: 

23 error_detected = True 

24 print(f"Gromet Wire {wire_type} {disp_wire(gromet_wire)} has negative src port.") 

25 if gromet_wire.src == 0: 

26 error_detected = True 

27 print(f"Gromet Wire {wire_type} {disp_wire(gromet_wire)} has zero src port.") 

28 if gromet_wire.src > src_port_count: 

29 error_detected = True 

30 print(f"Gromet Wire {wire_type} {disp_wire(gromet_wire)} has a src port that goes over the boundary of {src_port_count} src ports.") 

31 

32 if gromet_wire.tgt < 0: 

33 error_detected = True 

34 print(f"Gromet Wire {wire_type} {disp_wire(gromet_wire)} has negative tgt port.") 

35 if gromet_wire.tgt == 0: 

36 error_detected = True 

37 print(f"Gromet Wire {wire_type} {disp_wire(gromet_wire)} has zero tgt port.") 

38 if gromet_wire.tgt > tgt_port_count: 

39 error_detected = True 

40 print(f"Gromet Wire {wire_type} {disp_wire(gromet_wire)} has a tgt port that goes over the boundary of {tgt_port_count} tgt ports.") 

41 

42 

43 if error_detected: 

44 if metadata == None: 

45 print("No line number information exists for this particular wire!") 

46 else: 

47 print(f"Wire is associated with source code lines start:{metadata.line_begin} end:{metadata.line_end}") 

48 print() 

49 

50 return error_detected 

51 

52def find_metadata_idx(gromet_fn): 

53 """ 

54 Attempts to find a metadata associated with this fn 

55 If it finds something, return it, otherwise return None 

56 """ 

57 if gromet_fn.b != None: 

58 for b in gromet_fn.b: 

59 if b.metadata != None: 

60 return b.metadata 

61 

62 if gromet_fn.bf != None: 

63 for bf in gromet_fn.bf: 

64 if bf.metadata != None: 

65 return bf.metadata 

66 

67 return None 

68 

69def analyze_fn_wiring(gromet_fn, metadata_collection): 

70 # Acquire information for all the ports, if they exist 

71 pif_length = get_length(gromet_fn.pif) 

72 pof_length = get_length(gromet_fn.pof) 

73 opi_length = get_length(gromet_fn.opi) 

74 opo_length = get_length(gromet_fn.opo) 

75 pil_length = get_length(gromet_fn.pil) 

76 pol_length = get_length(gromet_fn.pol) 

77 pic_length = get_length(gromet_fn.pic) 

78 poc_length = get_length(gromet_fn.poc) 

79 

80 # Find a SourceCodeReference metadata that we can extract line number information for 

81 # so we can display some line number information about potential errors in the wiring  

82 # NOTE: Can we make this extraction more accurate? 

83 metadata_idx = find_metadata_idx(gromet_fn) 

84 metadata = None 

85 if metadata_idx != None: 

86 for md in metadata_collection[metadata_idx - 1]: 

87 if isinstance(md, SourceCodeReference): 

88 metadata = md 

89 

90 wopio_length = get_length(gromet_fn.wopio) 

91 if wopio_length > 0: 

92 for wire in gromet_fn.wff: 

93 check_wire(wire, opo_length, opi_length, "wff", metadata) 

94 

95 ######################## loop (bl) wiring 

96 

97 wlopi_length = get_length(gromet_fn.wlopi) 

98 if wlopi_length > 0: 

99 for wire in gromet_fn.wlopi: 

100 check_wire(wire, pil_length, opi_length, "wlopi", metadata) 

101 

102 wll_length = get_length(gromet_fn.wll) 

103 if wll_length > 0: 

104 for wire in gromet_fn.wll: 

105 check_wire(wire, pil_length, pol_length, "wll", metadata) 

106 

107 wlf_length = get_length(gromet_fn.wlf) 

108 if wlf_length > 0: 

109 for wire in gromet_fn.wlf: 

110 check_wire(wire, pif_length, pol_length, "wlf", metadata) 

111 

112 wlc_length = get_length(gromet_fn.wlc) 

113 if wlc_length > 0: 

114 for wire in gromet_fn.wlc: 

115 check_wire(wire, pic_length, pol_length, "wlc", metadata) 

116 

117 wlopo_length = get_length(gromet_fn.wlopo) 

118 if wlopo_length > 0: 

119 for wire in gromet_fn.wlopo: 

120 check_wire(wire, opo_length, pol_length, "wlopo", metadata) 

121 

122 ######################## function (bf) wiring 

123 wfopi_length = get_length(gromet_fn.wfopi) 

124 if wfopi_length > 0: 

125 for wire in gromet_fn.wfopi: 

126 check_wire(wire, pif_length, opi_length, "wfopi", metadata) 

127 

128 wfl_length = get_length(gromet_fn.wfl) 

129 if wfl_length > 0: 

130 for wire in gromet_fn.wfl: 

131 check_wire(wire, pil_length, pof_length, "wfl", metadata) 

132 

133 wff_length = get_length(gromet_fn.wff) 

134 if wff_length > 0: 

135 for wire in gromet_fn.wff: 

136 check_wire(wire, pif_length, pof_length, "wff", metadata) 

137 

138 wfc_length = get_length(gromet_fn.wfc) 

139 if wfc_length > 0: 

140 for wire in gromet_fn.wfc: 

141 check_wire(wire, pic_length, pof_length, "wfc", metadata) 

142 

143 wfopo_length = get_length(gromet_fn.wfopo) 

144 if wfopo_length > 0: 

145 for wire in gromet_fn.wfopo: 

146 check_wire(wire, opo_length, pof_length, "wfopo", metadata) 

147 

148 ######################## condition (bc) wiring 

149 wcopi_length = get_length(gromet_fn.wcopi) 

150 if wcopi_length > 0: 

151 for wire in gromet_fn.wcopi: 

152 check_wire(wire, pic_length, opi_length, "wcopi", metadata) 

153 

154 wcl_length = get_length(gromet_fn.wcl) 

155 if wcl_length > 0: 

156 for wire in gromet_fn.wcl: 

157 check_wire(wire, pil_length, poc_length, "wcl", metadata) 

158 

159 wcf_length = get_length(gromet_fn.wcf) 

160 if wcf_length > 0: 

161 for wire in gromet_fn.wcf: 

162 check_wire(wire, pif_length, poc_length, "wcf", metadata) 

163 

164 wcc_length = get_length(gromet_fn.wcc) 

165 if wcc_length > 0: 

166 for wire in gromet_fn.wcc: 

167 check_wire(wire, pic_length, poc_length, "wcc", metadata) 

168 

169 wcopo_length = get_length(gromet_fn.wcopo) 

170 if wcopo_length > 0: 

171 for wire in gromet_fn.wcopo: 

172 check_wire(wire, opo_length, poc_length, "wcopo", metadata) 

173 

174 

175def wiring_analyzer(gromet_obj): 

176 # TODO: Multifiles 

177 

178 for module in gromet_obj.modules: 

179 # first_module = gromet_obj.modules[0] 

180 metadata = [] 

181 # Analyze base FN 

182 print(f"Analyzing {module.name}") 

183 analyze_fn_wiring(module.fn, module.metadata_collection) 

184 

185 # Analyze the rest of the FN_array 

186 for fn in module.fn_array: 

187 analyze_fn_wiring(fn, module.metadata_collection) 

188 

189def get_args(): 

190 parser = argparse.ArgumentParser( 

191 "Attempts to analyize GroMEt JSON for issues" 

192 ) 

193 parser.add_argument( 

194 "gromet_file_path", 

195 help="input GroMEt JSON file" 

196 ) 

197 

198 options = parser.parse_args() 

199 return options 

200 

201if __name__ == "__main__": 

202 args = get_args() 

203 gromet_obj = json2gromet.json_to_gromet(args.gromet_file_path) 

204 

205 wiring_analyzer(gromet_obj) 

206 

207 

208