Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 4 additions & 6 deletions archive/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ def __init__(self):
}

def scan_directory(self, root_path):
inventory = {
inventory: dict = {
"meta": {"root": str(root_path), "total_files": 0, "algorithms_found": 0},
"files": [],
}
Expand Down Expand Up @@ -59,15 +59,13 @@ def analyze_file(self, file_path):
# print(f"Error parsing {file_path}: {e}") # Reduce noise
return None

file_data = {"path": str(file_path), "functions": []}
file_data: dict = {"path": str(file_path), "functions": []}

# Find all function definitions (simplified traversal)
# Note: Ideally this should be language-agnostic node type checking
# For now relying on tree-sitter standard naming common in supported langs
cursor = tree.walk()

def traverse(node):
if node.type == "function_definition" or node.type == "method_declaration":
if node.type in {"function_definition", "method_declaration"}:
self._analyze_function_node(node, code_bytes, file_data)

for child in node.children:
Expand All @@ -80,7 +78,7 @@ def traverse(node):
def _analyze_function_node(self, node, code_bytes, file_data):
try:
func_name = node.child_by_field_name("name").text.decode("utf-8")
except:
except Exception:
func_name = "anonymous"

function_entry = {"name": func_name, "line": node.start_point[0] + 1, "algorithms": []}
Expand Down
8 changes: 5 additions & 3 deletions dissect/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
from .exporters.mermaid import save_mermaid
from .trace_receiver import parse_trace_file

TRACE_FILE_HELP = "Path to trace file (JSON)"


def explain_command(args):
"""Generate AI summary of the trace."""
Expand Down Expand Up @@ -138,12 +140,12 @@ def main():

# Trace command
trace_parser = subparsers.add_parser("trace", help="Parse and inspect a trace file")
trace_parser.add_argument("--file", "-f", required=True, help="Path to trace file (JSON)")
trace_parser.add_argument("--file", "-f", required=True, help=TRACE_FILE_HELP)
trace_parser.set_defaults(func=trace_command)

# Visualize command
viz_parser = subparsers.add_parser("visualize", help="Generate visualization from trace")
viz_parser.add_argument("--file", "-f", required=True, help="Path to trace file (JSON)")
viz_parser.add_argument("--file", "-f", required=True, help=TRACE_FILE_HELP)
viz_parser.add_argument(
"--format",
"-t",
Expand All @@ -160,7 +162,7 @@ def main():
explain_parser = subparsers.add_parser(
"explain", help="Generate AI-powered insights from trace"
)
explain_parser.add_argument("--file", "-f", required=True, help="Path to trace file (JSON)")
explain_parser.add_argument("--file", "-f", required=True, help=TRACE_FILE_HELP)
explain_parser.add_argument(
"--provider",
"-p",
Expand Down
8 changes: 4 additions & 4 deletions dissect/graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,18 +127,18 @@ def dfs(node_id: str, visited: set) -> tuple:

return (node_duration + max_child_duration, [node] + max_child_path)

roots = self.get_root_nodes()
if not roots:
return []
return self._best_root_path(self.get_root_nodes(), dfs)

@staticmethod
def _best_root_path(roots, dfs):
"""Racine au plus long chemin (extrait pour complexite)."""
max_duration = 0
critical_path = []
for root in roots:
duration, path = dfs(root.id, set())
if duration > max_duration:
max_duration = duration
critical_path = path

return critical_path

def to_dict(self) -> Dict[str, Any]:
Expand Down
23 changes: 11 additions & 12 deletions dissect/trace_receiver.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,18 @@
"""

import json
from datetime import datetime
from typing import Any, Dict, List, Optional

from .graph import Edge, Node, NodeType, OrchestrationGraph

UTC_SUFFIX = "+00:00"


def _parse_ts(ts):
"""ISO-8601 tolerant Z (meme regle aux 3 sites d appel)."""
return datetime.fromisoformat(ts.replace("Z", UTC_SUFFIX))


class TraceParser:
"""
Expand Down Expand Up @@ -210,11 +218,8 @@ def _parse_timestamp(self, ts: Any) -> Optional[float]:
return ts
if isinstance(ts, str):
# Try ISO format
from datetime import datetime

try:
dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
return dt.timestamp()
return _parse_ts(ts).timestamp()
except Exception:
return None
return None
Expand Down Expand Up @@ -352,11 +357,8 @@ def _parse_timestamp(self, ts: Any) -> Optional[float]:
return ts / 1000
return ts
if isinstance(ts, str):
from datetime import datetime

try:
dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
return dt.timestamp()
return _parse_ts(ts).timestamp()
except Exception:
return None
return None
Expand Down Expand Up @@ -454,11 +456,8 @@ def _parse_timestamp(self, ts: Any) -> Optional[float]:
return ts / 1000
return ts
if isinstance(ts, str):
from datetime import datetime

try:
dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
return dt.timestamp()
return _parse_ts(ts).timestamp()
except Exception:
return None
return None
Expand Down
Loading