Coverage for src / graphable / cli / commands / core.py: 91%

1from pathlib import Path 

2from typing import Any, Callable 

3 

4from ...enums import Engine 

5from ...graph import Graph 

6from ...registry import EXPORTERS, PARSERS 

7 

8 

9def get_parser(extension: str) -> Callable[..., Graph[Any]]: 

10 parser = PARSERS.get(extension.lower()) 

11 if not parser: 

12 raise ValueError(f"No parser available for extension: {extension}") 

13 return parser 

14 

15 

16def get_exporter(extension: str) -> Callable[..., None] | None: 

17 return EXPORTERS.get(extension.lower()) 

18 

19 

20def load_graph( 

21 path: Path, 

22 tag: str | None = None, 

23 upstream_of: str | None = None, 

24 downstream_of: str | None = None, 

25) -> Graph[Any]: 

26 parser = get_parser(path.suffix) 

27 g = parser(path) 

28 if tag: 

29 g = g.subgraph_tagged(tag) 

30 if upstream_of: 

31 g = g.upstream_of(g[upstream_of]) 

32 if downstream_of: 

33 g = g.downstream_of(g[downstream_of]) 

34 return g 

35 

36 

37def info_command( 

38 path: Path, 

39 tag: str | None = None, 

40 upstream_of: str | None = None, 

41 downstream_of: str | None = None, 

42) -> dict[str, Any]: 

43 g = load_graph(path, tag, upstream_of, downstream_of) 

44 

45 stats = { 

46 "nodes": len(g), 

47 "edges": sum(len(node.dependents) for node in g), 

48 "sources": [n.reference for n in g.sources], 

49 "sinks": [n.reference for n in g.sinks], 

50 "project_duration": None, 

51 "critical_path_length": 0, 

52 } 

53 

54 # Check if we should run CPM (if any node has duration > 0) 

55 if any(n.duration > 0 for n in g): 

56 analysis = g.cpm_analysis() 

57 stats["project_duration"] = max((v["EF"] for v in analysis.values()), default=0) 

58 stats["critical_path_length"] = len(g.critical_path()) 

59 

60 return stats 

61 

62 

63def check_command( 

64 path: Path, 

65 tag: str | None = None, 

66 upstream_of: str | None = None, 

67 downstream_of: str | None = None, 

68) -> dict[str, Any]: 

69 try: 

70 g = load_graph(path, tag, upstream_of, downstream_of) 

71 g.check_cycles() 

72 g.check_consistency() 

73 return {"valid": True, "error": None} 

74 except Exception as e: 

75 return {"valid": False, "error": str(e)} 

76 

77 

78def reduce_command( 

79 input_path: Path, 

80 output_path: Path, 

81 embed_checksum: bool = False, 

82 tag: str | None = None, 

83 upstream_of: str | None = None, 

84 downstream_of: str | None = None, 

85) -> None: 

86 g = load_graph(input_path, tag, upstream_of, downstream_of) 

87 g.write(output_path, transitive_reduction=True, embed_checksum=embed_checksum) 

88 

89 

90def convert_command( 

91 input_path: Path, 

92 output_path: Path, 

93 tag: str | None = None, 

94 upstream_of: str | None = None, 

95 downstream_of: str | None = None, 

96 **kwargs: Any, 

97) -> None: 

98 # Restrict convert to non-image formats 

99 ext = output_path.suffix.lower() 

100 if ext in (".png", ".svg"): 

101 raise ValueError( 

102 f"Extension '{ext}' is for images. Use the 'render' command instead." 

103 ) 

104 

105 g = load_graph(input_path, tag, upstream_of, downstream_of) 

106 g.write(output_path, **kwargs) 

107 

108 

109def checksum_command( 

110 path: Path, 

111 tag: str | None = None, 

112 upstream_of: str | None = None, 

113 downstream_of: str | None = None, 

114) -> str: 

115 g = load_graph(path, tag, upstream_of, downstream_of) 

116 return g.checksum() 

117 

118 

119def verify_command( 

120 path: Path, 

121 expected: str | None = None, 

122 tag: str | None = None, 

123 upstream_of: str | None = None, 

124 downstream_of: str | None = None, 

125) -> dict[str, Any]: 

126 g = load_graph(path, tag, upstream_of, downstream_of) 

127 actual = g.checksum() 

128 

129 if expected is None: 

130 # Check if there is an embedded checksum 

131 from ...parsers.utils import extract_checksum 

132 

133 expected = extract_checksum(path) 

134 

135 if expected is None: 

136 return {"valid": None, "actual": actual, "expected": None} 

137 

138 return {"valid": actual == expected, "actual": actual, "expected": expected} 

139 

140 

141def write_checksum_command( 

142 graph_path: Path, 

143 checksum_path: Path, 

144 tag: str | None = None, 

145 upstream_of: str | None = None, 

146 downstream_of: str | None = None, 

147) -> None: 

148 g = load_graph(graph_path, tag, upstream_of, downstream_of) 

149 g.write_checksum(checksum_path) 

150 

151 

152def diff_command(path1: Path, path2: Path, tag: str | None = None) -> dict[str, Any]: 

153 g1 = load_graph(path1, tag) 

154 g2 = load_graph(path2, tag) 

155 return g1.diff(g2) 

156 

157 

158def diff_visual_command( 

159 path1: Path, path2: Path, output_path: Path, tag: str | None = None 

160) -> None: 

161 g1 = load_graph(path1, tag) 

162 g2 = load_graph(path2, tag) 

163 dg = g1.diff_graph(g2) 

164 dg.write(output_path) 

165 

166 

167def render_command( 

168 input_path: Path, 

169 output_path: Path, 

170 engine: Engine | str | None = None, 

171 tag: str | None = None, 

172 upstream_of: str | None = None, 

173 downstream_of: str | None = None, 

174) -> None: 

175 """ 

176 Load a graph and render it as an image (SVG or PNG). 

177 

178 Args: 

179 input_path: Path to the input graph file. 

180 output_path: Path to the output image file. 

181 engine: The rendering engine to use (e.g., Engine.MERMAID, 'graphviz'). 

182 If None, it will be auto-detected based on system PATH. 

183 tag: Optional tag to filter the graph before rendering. 

184 upstream_of: Optional node reference to keep only its ancestors. 

185 downstream_of: Optional node reference to keep only its descendants. 

186 """ 

187 g = load_graph(input_path, tag, upstream_of, downstream_of) 

188 

189 from ...views.utils import get_image_exporter 

190 

191 exporter = get_image_exporter(engine) 

192 exporter(g, output_path) 

193 

194 

195def paths_command( 

196 path: Path, 

197 source: str, 

198 target: str, 

199 tag: str | None = None, 

200 upstream_of: str | None = None, 

201 downstream_of: str | None = None, 

202) -> list[list[str]]: 

203 """ 

204 Find all paths between two nodes. 

205 

206 Args: 

207 path: Path to the graph file. 

208 source: Reference of the source node. 

209 target: Reference of the target node. 

210 tag: Optional tag to filter the graph. 

211 upstream_of: Optional node reference to slice upstream. 

212 downstream_of: Optional node reference to slice downstream. 

213 

214 Returns: 

215 list[list[str]]: A list of paths, each being a list of node references. 

216 """ 

217 g = load_graph(path, tag, upstream_of, downstream_of) 

218 u = g[source] 

219 v = g[target] 

220 

221 paths = g.all_paths(u, v) 

222 return [[n.reference for n in path] for path in paths]