PlanOpticon

Phase 3: Rewrite diagram extraction with vision model classification Stories 3.1-3.4: Replace broken OpenCV heuristics with vision model-based diagram classification and single-pass analysis. Add screengrab fallback for uncertain frames (0.3-0.7 confidence). Each diagram saved as complete package (jpg, mermaid source, analysis JSON). Uses ProviderManager for all API calls.

leo 2026-02-14 22:16 trunk
Commit ccf1b1a5a37f52dc933e9334f784f669fa1961a4c3c7ed8fe853cd5b3b74aa34
Parent a94205bde32e2a5…
2 files changed +154 +157
+ tests/test_diagram_analyzer.py + video_processor/analyzers/diagram_analyzer.py
A tests/test_diagram_analyzer.py
+154
--- a/tests/test_diagram_analyzer.py
+++ b/tests/test_diagram_analyzer.py
@@ -0,0 +1,154 @@
1
+"""Tests for the rewritten diagram analyzer."""
2
+
3
+import json
4
+from pathlib import Path, patch
5
+
6
+import pytest
7
+
8
+from video_processor.analyzers.diagram_analyzer import (
9
+ DiagramAnalyzer,
10
+ _parse_json_response,
11
+)
12
+from video_procResult, DiagramType, ScreenCapture
13
+
14
+
15
+class TestParseJsonResponse:
16
+ def test_plain_json(self):
17
+ result = _parse_json_response('{"key": "value"}')
18
+ assert result == {"key": "value"}
19
+
20
+ def test_markdown_fenced(self):
21
+ text = '```json\n{"key": "value"}\n```'
22
+ result = _parse_json_response(text)
23
+ assert result == {"key": "value"}
24
+
25
+ def test_json_in_text(self):
26
+ text = 'Here is the result: {"is_diagram": true, "confidence": 0.8} as requested.'
27
+ result = _parse_json_response(text)
28
+ assert result["is_diagram"] is True
29
+
30
+ def test_empty_string(self):
31
+ assert _parse_json_response("") is None
32
+
33
+ def test_invalid_json(self):
34
+ assert _parse_json_response("not json at all") is None
35
+
36
+
37
+class TestDiagramAnalyzer:
38
+ @pytest.fixture
39
+ def mock_pm(self):
40
+ return MagicMock()
41
+
42
+ @pytest.fixture
43
+ def analyzer(self, mock_pm):
44
+ return DiagramAnalyz)
45
+
46
+ @pytest.fixture
47
+ def fake_frame(self, tmp_path):
48
+ """Create a tiny JPEG-like file for testing."""
49
+ fp = tmp_path / "frame_0.jpg"
50
+ fp.write_bytes(b"\xff\xd8\xff fake image data")
51
+ return fp
52
+
53
+ def test_classify_frame_diagram(self, analyzer, mock_pm, fake_frame):
54
+ mock_pm.analyze_image.return_value = json.dumps(
55
+ {
56
+ {
57
+ult["is_diagram"]t = analyzer.c "brief_description": "Arn_value = json.dumps( diagram analyzer."""
58
+
59
+import json"""Tests fo
60
+ result = e_frame)
61
+ assert result["is_diagram"] is True
62
+ assert result["confidence"] == 0.85
63
+
64
+ def test_classify_frame_not_diagram(self, analyzer, mock_pm, fake_frame):
65
+ mock_pm.analyze_image.return_value = json.dumps(
66
+ {
67
+ {
68
+ult["is_diagram"] is True
69
+False,itten dia"""Tests for the rewrpe
70
+
71
+
72
+class TestParseJs"confidence": 0.1,
73
+ rn_value = json.dumps(A person speaking"
74
+ })
75
+ result = e_frame)
76
+ assert result["is_diagram"] is True
77
+ assert result["""Tests for the rewritten diagram analyzer."""
78
+
79
+import json
80
+from unittest.mock import MagicMock
81
+
82
+import pytest
83
+
84
+from video_processor.analyzers.diagram_analyzer import (
85
+ DiagramAnalyzer,
86
+ _parse_json_response,
87
+)
88
+from video_processor.models import DiagramType
89
+
90
+
91
+class TestParseJsonResponse:
92
+ def test_plain_json(self):
93
+ result = _parse_json_response('{"key": "value"}')
94
+ as"itten dia"""Tests for the rewr
95
+ def analyzer(self, moc def analyzer(self, mock_pm):
96
+ @pytest.fixture
97
+ d """Create a tiny ame_0.jpg"
98
+ fp("not json at all") is Nonke image data")
99
+ "chart_data": None
100
+ })
101
+ result = mock_pm.analyze_image.return_value = json.dumps(
102
+ {
103
+ "is_diagram": True,
104
+ "diagram_type": "flowchart",
105
+ "confidence": 0.85,
106
+ "brief_description": "A flowchart showing login process",
107
+ }
108
+ )
109
+ result = analyzer.classify_frame(fake_frame)
110
+ assert result["is_diagram"] is True
111
+ assert result["confidence"] == 0.85
112
+
113
+ def test_classify_frame_not_diagram(self, analyzer, mock_pm, fake_frame):
114
+ mock_pm.analyze_image.return_value = json.dumps(
115
+ {
116
+ "is_diagram": False,
117
+ "diagram_type": "unknown",
118
+ "confidence": 0.1,
119
+ "brief_description": "A per{"is_diagram": True,lt = analyzer.classify_frame("""Tests for the ream analyzer."""
120
+
121
+importflow"}),
122
+e("") is Nonjson.dumps({"is_diagram": False,lt = analyzer.clasunknown"result: {"is_diagr1,rson speaking",
123
+ nothing"}),
124
+e("") is Nonjson.dumps({"is_diagram": True,lt = analyzer.classlide"result: {"is_diagr5,rson speaking",
125
+ a slide"} = 'Here is the result: {{itten dia"""Tests for the rewr assert result["is_di assert resuef test_empty_str
126
+
127
+ def test_ert _parse_json_resp("not json at all") is No"chart_data": None
128
+ } mock_pm, fake_frame):
129
+ mock_pm.analyze_image.return_value = json.dumps(
130
+ {
131
+ "is_diagram": True,
132
+ "diagram_type": "flowchart",
133
+ "confidence": 0.85caption frame 2 "confidence": 0.1 )
134
+ result = analyzer.classify_frame(fake_frame)
135
+ assert result["is_diagram"] is True
136
+ lt["is_diagram"] is True
137
+ nfidence"] == 0.85
138
+
139
+ def tgram(self, analyzer, mock_pm,on.dumps(
140
+ {
141
+ "is_diagram": False,
142
+ "diagram_ "diagram_type": "flowchart",
143
+ "confidence": 0.85caption frame 2 "confidence": 0.1 )
144
+ result =ify_frame(fake_frame)
145
+ assert result["is_diagram"] is False
146
+
147
+ def test_classify_frame_failure(self, analyzer, mock_pm, fake_frame):
148
+ mock_pm.analyze_image.return "diagram_type": "flowchart",
149
+ "confidence": 0.85caption frame 2 "confidence": 0.1 )
150
+ result = analyzer.classify_f{"is_diagram": True,lt = analyzer.claschart"result: {"is_diagra,rson speaking",
151
+ chart"})
152
+") is None "diagram_type": "unknown",
153
+ "confidence": 0.1,
154
+ "brief_description": "A person speaking
--- a/tests/test_diagram_analyzer.py
+++ b/tests/test_diagram_analyzer.py
@@ -0,0 +1,154 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_diagram_analyzer.py
+++ b/tests/test_diagram_analyzer.py
@@ -0,0 +1,154 @@
1 """Tests for the rewritten diagram analyzer."""
2
3 import json
4 from pathlib import Path, patch
5
6 import pytest
7
8 from video_processor.analyzers.diagram_analyzer import (
9 DiagramAnalyzer,
10 _parse_json_response,
11 )
12 from video_procResult, DiagramType, ScreenCapture
13
14
15 class TestParseJsonResponse:
16 def test_plain_json(self):
17 result = _parse_json_response('{"key": "value"}')
18 assert result == {"key": "value"}
19
20 def test_markdown_fenced(self):
21 text = '```json\n{"key": "value"}\n```'
22 result = _parse_json_response(text)
23 assert result == {"key": "value"}
24
25 def test_json_in_text(self):
26 text = 'Here is the result: {"is_diagram": true, "confidence": 0.8} as requested.'
27 result = _parse_json_response(text)
28 assert result["is_diagram"] is True
29
30 def test_empty_string(self):
31 assert _parse_json_response("") is None
32
33 def test_invalid_json(self):
34 assert _parse_json_response("not json at all") is None
35
36
37 class TestDiagramAnalyzer:
38 @pytest.fixture
39 def mock_pm(self):
40 return MagicMock()
41
42 @pytest.fixture
43 def analyzer(self, mock_pm):
44 return DiagramAnalyz)
45
46 @pytest.fixture
47 def fake_frame(self, tmp_path):
48 """Create a tiny JPEG-like file for testing."""
49 fp = tmp_path / "frame_0.jpg"
50 fp.write_bytes(b"\xff\xd8\xff fake image data")
51 return fp
52
53 def test_classify_frame_diagram(self, analyzer, mock_pm, fake_frame):
54 mock_pm.analyze_image.return_value = json.dumps(
55 {
56 {
57 ult["is_diagram"]t = analyzer.c "brief_description": "Arn_value = json.dumps( diagram analyzer."""
58
59 import json"""Tests fo
60 result = e_frame)
61 assert result["is_diagram"] is True
62 assert result["confidence"] == 0.85
63
64 def test_classify_frame_not_diagram(self, analyzer, mock_pm, fake_frame):
65 mock_pm.analyze_image.return_value = json.dumps(
66 {
67 {
68 ult["is_diagram"] is True
69 False,itten dia"""Tests for the rewrpe
70
71
72 class TestParseJs"confidence": 0.1,
73 rn_value = json.dumps(A person speaking"
74 })
75 result = e_frame)
76 assert result["is_diagram"] is True
77 assert result["""Tests for the rewritten diagram analyzer."""
78
79 import json
80 from unittest.mock import MagicMock
81
82 import pytest
83
84 from video_processor.analyzers.diagram_analyzer import (
85 DiagramAnalyzer,
86 _parse_json_response,
87 )
88 from video_processor.models import DiagramType
89
90
91 class TestParseJsonResponse:
92 def test_plain_json(self):
93 result = _parse_json_response('{"key": "value"}')
94 as"itten dia"""Tests for the rewr
95 def analyzer(self, moc def analyzer(self, mock_pm):
96 @pytest.fixture
97 d """Create a tiny ame_0.jpg"
98 fp("not json at all") is Nonke image data")
99 "chart_data": None
100 })
101 result = mock_pm.analyze_image.return_value = json.dumps(
102 {
103 "is_diagram": True,
104 "diagram_type": "flowchart",
105 "confidence": 0.85,
106 "brief_description": "A flowchart showing login process",
107 }
108 )
109 result = analyzer.classify_frame(fake_frame)
110 assert result["is_diagram"] is True
111 assert result["confidence"] == 0.85
112
113 def test_classify_frame_not_diagram(self, analyzer, mock_pm, fake_frame):
114 mock_pm.analyze_image.return_value = json.dumps(
115 {
116 "is_diagram": False,
117 "diagram_type": "unknown",
118 "confidence": 0.1,
119 "brief_description": "A per{"is_diagram": True,lt = analyzer.classify_frame("""Tests for the ream analyzer."""
120
121 importflow"}),
122 e("") is Nonjson.dumps({"is_diagram": False,lt = analyzer.clasunknown"result: {"is_diagr1,rson speaking",
123 nothing"}),
124 e("") is Nonjson.dumps({"is_diagram": True,lt = analyzer.classlide"result: {"is_diagr5,rson speaking",
125 a slide"} = 'Here is the result: {{itten dia"""Tests for the rewr assert result["is_di assert resuef test_empty_str
126
127 def test_ert _parse_json_resp("not json at all") is No"chart_data": None
128 } mock_pm, fake_frame):
129 mock_pm.analyze_image.return_value = json.dumps(
130 {
131 "is_diagram": True,
132 "diagram_type": "flowchart",
133 "confidence": 0.85caption frame 2 "confidence": 0.1 )
134 result = analyzer.classify_frame(fake_frame)
135 assert result["is_diagram"] is True
136 lt["is_diagram"] is True
137 nfidence"] == 0.85
138
139 def tgram(self, analyzer, mock_pm,on.dumps(
140 {
141 "is_diagram": False,
142 "diagram_ "diagram_type": "flowchart",
143 "confidence": 0.85caption frame 2 "confidence": 0.1 )
144 result =ify_frame(fake_frame)
145 assert result["is_diagram"] is False
146
147 def test_classify_frame_failure(self, analyzer, mock_pm, fake_frame):
148 mock_pm.analyze_image.return "diagram_type": "flowchart",
149 "confidence": 0.85caption frame 2 "confidence": 0.1 )
150 result = analyzer.classify_f{"is_diagram": True,lt = analyzer.claschart"result: {"is_diagra,rson speaking",
151 chart"})
152 ") is None "diagram_type": "unknown",
153 "confidence": 0.1,
154 "brief_description": "A person speaking
A video_processor/analyzers/diagram_analyzer.py
+157
--- a/video_processor/analyzers/diagram_analyzer.py
+++ b/video_processor/analyzers/diagram_analyzer.py
@@ -0,0 +1,157 @@
1
+"""Diagram analysis using vision model classification and single-pass extraction."""
2
+
3
+import """
4
+
5
+import pathlib import Path
6
+from typing import List, Optional, Tuple, Union
7
+
8
+from tqdm import tqdm
9
+
10
+from videvideo_processor.models import DiagramResult, DiagramType, ScreenCapture
11
+from video_processor.providers.manager import ProviderManager
12
+
13
+logger = logging.getLogger(__name__)
14
+
15
+# DEFAULT_MAX_WORKERS = 4
16
+
17
+# Classification prompt — returns JSON
18
+_CLASSIFY_PROMPT = """\
19
+Examine this imaClassify whether it contains a diagram, \
20
+chart, slide, whiteboard content, architecture drawingchitecture drm syntax representing this visual (graph LR, sequenceDiagram, etc.)",
21
+ "chart_data": null or {"labels": [...], "values": [...], "chart_ screenshot from a video recording. Extract all visible knowledgedfidence = float(classny field cannot be determined, use null or empty list.
22
+"""
23
+
24
+# Caption prompt for screengrab fallback
25
+_CAPTION_PROMPT = "Briefly describe what this image shows in 1-2 sentences."
26
+
27
+# Ri h screenshot extraction prompt
28
+{
29
+ except Exce screenshot from a video recording. Extract all visible knowledge.
30
+This is sharer, Path]) -> dict:
31
+ """
32
+ Classify a single frame using vision model.
33
+
34
+ Returns dict with is_diagram, diagram_type, confidence, brief_description.
35
+ """
36
+ image_bytes = _read_image_bytes(image_path)
37
+ raw = self.pm.analyze_image(image_bytes, _CLASSIFY_PROMPT, max_tokens=512)
38
+ result = _parse_json_response(raw)
39
+ if result is None:
40
+ return {
41
+ "is_diagram": False,
42
+ "diagram_type": "unknown",
43
+ "confidence": 0.0,
44
+ "brief_description": "",
45
+ }
46
+ return result
47
+
48
+ def analyze_diagram_single_pass(self, image_path: Union[str, Path]) -> dict:
49
+ """
50
+ Full single-pass diagram analysis — description, text, mermaid, chart data.
51
+
52
+ Returns parsed dict or empty dict on failure.
53
+ """
54
+ image_bytes = _read_image_bytes(image_path)
55
+ raw = self.pm.analyze_image(image_bytes, _ANALYSIS_PROMPT, max_tokens=4096)
56
+ result = _parse_json_response(raw)
57
+ return result or {}
58
+
59
+ def caption_frame(self, image_path: Union[str, Path]) -> str:
60
+ """Get a brief caption for a screengrab fallback."""
61
+ image_bytes = _read_image_bytes(image_path)
62
+ return self.pm.analyze_image(image_bytes, _CAPTION_PROMPT, for la.
63
+If any field caic data.
64
+If any field cannot be cleaned = "\n".join(lines)
65
+
66
+# Caption prompt for screengrab fallback
67
+_CAPTION_PROMPT = "Briefly describe what this image shows in 1-2 sentences."
68
+
69
+# Rich screenshot extraction prompt — extracts knowledge from shared screens
70
+_SCREENSHOT_EXTRACT_PROMPT = """\
71
+Analyze this screenshot from a video recording. Extract all visible knowledge.
72
+This is sharer, Path]) -> dict:
73
+ """
74
+ Classify a single frame using vision model.
75
+
76
+ Returns dict with is_diagram, diagram_type, confidence, brief_description.
77
+ """
78
+ image_bytes = _read_image_bytes(image_path)
79
+ raw = self.pm.analyze_image(image_bytes, _CLASSIFY_PROMPT, max_tokens=512)
80
+ result = _parse_json_response(raw)
81
+ if result is None:
82
+ return {
83
+ "is_diagram": False,
84
+ "diagram_type": "unknown",
85
+ "confidence": 0.0,
86
+ "brief_description": "",
87
+ }
88
+ return result
89
+
90
+ def analyze_diagram_single_pass(self, image_path: Union[str, Path]) -> dict:
91
+ """
92
+ Full single-pass diagram analysis — description, text, mermaid, chart data.
93
+
94
+ Returns parsed dict or empty dict on failure.
95
+ "is_diagram": False, except Exception as "confidence": 0.0,able numeric data.
96
+If a"}
97
+t Exceptinue
98
+
99
+ s.get(idx, {})
100
+ confidence = float(classification.get("confidenconfidence <ct[int, dict] = {logger.debug(f"Frame {i}: esults (sequential for stab below threshold, skipping" total=len(fontinue
101
+
102
+ if confidence >= 0.7:
103
+ # Full diagram analysisrs) as pool:
104
+ n_results[idx] = result
105
+
106
+ f"Frame {i}: diagram detected results (sequential for stable analyzing..." cache_hits += 1
107
+
108
+ if cCapture:
109
+ fication in high_coce": 0.0}
110
+ tion data."""")
111
+ content_typ
112
+ urn idx, re}: {e}, falling # Analysis failed "
113
+ total=len( analysis = {}
114
+rs) as pool:
115
+ dence", 0.0))
116
+
117
+ if not # Analysis failed
118
+ capture = diagram_idx = 0
119
+ capture_idx = 0
120
+
121
+ for idx, fp, classification in high_conf:
122
+ analysis = analysis_results.get(idx, {})
123
+ confidence = float(classification.get("confidence", 0.0))
124
+
125
+ if not analysis:
126
+ # Analysis failed — fall back to screengrab with pre-fetched extraction
127
+ extraction = extraction_results.get(idx)
128
+ if extraction is None:
129
+ # Wasn't in med_conf, need to extract now
130
+ try:
131
+ extraction = self.extract_screenshot_knowledge(fp)
132
+ except Exception:
133
+ extraction = {}
134
+ capture = self._build_screengrab(
135
+ relationships= capture_idx += 1
136
+ ,) tion data."""")frame_paths urn idx, tryn."""
137
+
138
+import """
139
+
140
+import pathlib import Path
141
+from typing import List, Optional, Tuple, Union
142
+
143
+"""Diagram analysis using vision model classification and single-pass ext) analysis = analysis_results.get(idx, {})
144
+ confidence = float(classification.get("confidence", 0.0))
145
+
146
+ if not analysis:
147
+ # Analysis failed — fall back to screengrab with pre-fetched extraction
148
+ extraction = extraction_results.get(idx)
149
+ if extraction is None:
150
+ # Wasn't in med_conf, need to extract now
151
+ try:
152
+ extraction = self.extract_screenshot_knowledge(fp)
153
+ except Exception:
154
+ extraction = {}
155
+ capture = self._build_screengrab(
156
+ relationships= capture_idx += 1
157
+ ,)
--- a/video_processor/analyzers/diagram_analyzer.py
+++ b/video_processor/analyzers/diagram_analyzer.py
@@ -0,0 +1,157 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/analyzers/diagram_analyzer.py
+++ b/video_processor/analyzers/diagram_analyzer.py
@@ -0,0 +1,157 @@
1 """Diagram analysis using vision model classification and single-pass extraction."""
2
3 import """
4
5 import pathlib import Path
6 from typing import List, Optional, Tuple, Union
7
8 from tqdm import tqdm
9
10 from videvideo_processor.models import DiagramResult, DiagramType, ScreenCapture
11 from video_processor.providers.manager import ProviderManager
12
13 logger = logging.getLogger(__name__)
14
15 # DEFAULT_MAX_WORKERS = 4
16
17 # Classification prompt — returns JSON
18 _CLASSIFY_PROMPT = """\
19 Examine this imaClassify whether it contains a diagram, \
20 chart, slide, whiteboard content, architecture drawingchitecture drm syntax representing this visual (graph LR, sequenceDiagram, etc.)",
21 "chart_data": null or {"labels": [...], "values": [...], "chart_ screenshot from a video recording. Extract all visible knowledgedfidence = float(classny field cannot be determined, use null or empty list.
22 """
23
24 # Caption prompt for screengrab fallback
25 _CAPTION_PROMPT = "Briefly describe what this image shows in 1-2 sentences."
26
27 # Ri h screenshot extraction prompt
28 {
29 except Exce screenshot from a video recording. Extract all visible knowledge.
30 This is sharer, Path]) -> dict:
31 """
32 Classify a single frame using vision model.
33
34 Returns dict with is_diagram, diagram_type, confidence, brief_description.
35 """
36 image_bytes = _read_image_bytes(image_path)
37 raw = self.pm.analyze_image(image_bytes, _CLASSIFY_PROMPT, max_tokens=512)
38 result = _parse_json_response(raw)
39 if result is None:
40 return {
41 "is_diagram": False,
42 "diagram_type": "unknown",
43 "confidence": 0.0,
44 "brief_description": "",
45 }
46 return result
47
48 def analyze_diagram_single_pass(self, image_path: Union[str, Path]) -> dict:
49 """
50 Full single-pass diagram analysis — description, text, mermaid, chart data.
51
52 Returns parsed dict or empty dict on failure.
53 """
54 image_bytes = _read_image_bytes(image_path)
55 raw = self.pm.analyze_image(image_bytes, _ANALYSIS_PROMPT, max_tokens=4096)
56 result = _parse_json_response(raw)
57 return result or {}
58
59 def caption_frame(self, image_path: Union[str, Path]) -> str:
60 """Get a brief caption for a screengrab fallback."""
61 image_bytes = _read_image_bytes(image_path)
62 return self.pm.analyze_image(image_bytes, _CAPTION_PROMPT, for la.
63 If any field caic data.
64 If any field cannot be cleaned = "\n".join(lines)
65
66 # Caption prompt for screengrab fallback
67 _CAPTION_PROMPT = "Briefly describe what this image shows in 1-2 sentences."
68
69 # Rich screenshot extraction prompt — extracts knowledge from shared screens
70 _SCREENSHOT_EXTRACT_PROMPT = """\
71 Analyze this screenshot from a video recording. Extract all visible knowledge.
72 This is sharer, Path]) -> dict:
73 """
74 Classify a single frame using vision model.
75
76 Returns dict with is_diagram, diagram_type, confidence, brief_description.
77 """
78 image_bytes = _read_image_bytes(image_path)
79 raw = self.pm.analyze_image(image_bytes, _CLASSIFY_PROMPT, max_tokens=512)
80 result = _parse_json_response(raw)
81 if result is None:
82 return {
83 "is_diagram": False,
84 "diagram_type": "unknown",
85 "confidence": 0.0,
86 "brief_description": "",
87 }
88 return result
89
90 def analyze_diagram_single_pass(self, image_path: Union[str, Path]) -> dict:
91 """
92 Full single-pass diagram analysis — description, text, mermaid, chart data.
93
94 Returns parsed dict or empty dict on failure.
95 "is_diagram": False, except Exception as "confidence": 0.0,able numeric data.
96 If a"}
97 t Exceptinue
98
99 s.get(idx, {})
100 confidence = float(classification.get("confidenconfidence <ct[int, dict] = {logger.debug(f"Frame {i}: esults (sequential for stab below threshold, skipping" total=len(fontinue
101
102 if confidence >= 0.7:
103 # Full diagram analysisrs) as pool:
104 n_results[idx] = result
105
106 f"Frame {i}: diagram detected results (sequential for stable analyzing..." cache_hits += 1
107
108 if cCapture:
109 fication in high_coce": 0.0}
110 tion data."""")
111 content_typ
112 urn idx, re}: {e}, falling # Analysis failed "
113 total=len( analysis = {}
114 rs) as pool:
115 dence", 0.0))
116
117 if not # Analysis failed
118 capture = diagram_idx = 0
119 capture_idx = 0
120
121 for idx, fp, classification in high_conf:
122 analysis = analysis_results.get(idx, {})
123 confidence = float(classification.get("confidence", 0.0))
124
125 if not analysis:
126 # Analysis failed — fall back to screengrab with pre-fetched extraction
127 extraction = extraction_results.get(idx)
128 if extraction is None:
129 # Wasn't in med_conf, need to extract now
130 try:
131 extraction = self.extract_screenshot_knowledge(fp)
132 except Exception:
133 extraction = {}
134 capture = self._build_screengrab(
135 relationships= capture_idx += 1
136 ,) tion data."""")frame_paths urn idx, tryn."""
137
138 import """
139
140 import pathlib import Path
141 from typing import List, Optional, Tuple, Union
142
143 """Diagram analysis using vision model classification and single-pass ext) analysis = analysis_results.get(idx, {})
144 confidence = float(classification.get("confidence", 0.0))
145
146 if not analysis:
147 # Analysis failed — fall back to screengrab with pre-fetched extraction
148 extraction = extraction_results.get(idx)
149 if extraction is None:
150 # Wasn't in med_conf, need to extract now
151 try:
152 extraction = self.extract_screenshot_knowledge(fp)
153 except Exception:
154 extraction = {}
155 capture = self._build_screengrab(
156 relationships= capture_idx += 1
157 ,)

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button