File size: 11,019 Bytes
8a74c03
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
#!/usr/bin/env python3
"""
Final test of the optimized Person on Track Detector on all test videos
"""
import sys
import os
from io import BytesIO
import glob

# Add current directory to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

def test_person_on_track_final():
    """Test the optimized Person on Track Detector on all test videos"""
    print("FINAL PERSON ON TRACK DETECTOR TEST")
    print("=" * 50)
    print("Testing optimized detector based on comprehensive analysis")
    print()
    
    try:
        from local_models import get_local_model_manager
        from app import extract_frames_from_video, process_image_locally
        print("+ Components loaded successfully")
    except ImportError as e:
        print(f"- Import error: {e}")
        return
    
    # Find all test videos
    test_videos = glob.glob("test\\*.mp4")
    if not test_videos:
        print("- No MP4 files found in test folder")
        return
    
    print(f"+ Found {len(test_videos)} test videos")
    
    # Initialize models
    try:
        local_manager = get_local_model_manager()
        available_models = local_manager.get_available_models()
        print(f"+ Available models: {available_models}")
        
        if "Person on Track Detector" not in available_models:
            print("- Person on Track Detector not found!")
            return
        
        print("+ Person on Track Detector ready")
    except Exception as e:
        print(f"- Model initialization error: {e}")
        return
    
    all_results = []
    
    # Test each video
    for video_idx, video_path in enumerate(test_videos):
        video_name = os.path.basename(video_path)
        print(f"\n" + "=" * 60)
        print(f"TESTING VIDEO {video_idx + 1}: {video_name}")
        print("=" * 60)
        
        try:
            # Extract frames
            with open(video_path, 'rb') as f:
                video_data = f.read()
            
            video_file = BytesIO(video_data)
            frames = extract_frames_from_video(video_file, fps=0.5)
            
            if not frames:
                print(f"- No frames extracted from {video_name}")
                continue
            
            print(f"+ Extracted {len(frames)} frames from {video_name}")
            
            # Test first 3 frames
            test_frames = frames[:3]
            video_results = []
            
            for frame_idx, frame_data in enumerate(test_frames):
                frame_num = frame_idx + 1
                timestamp = frame_data['timestamp']
                
                print(f"\n  Frame {frame_num} ({timestamp:.1f}s):")
                print(f"  {'-' * 40}")
                
                try:
                    result = process_image_locally(
                        frame_data['frame'],
                        "Track Safety Analysis",  # Prompt is ignored for this detector
                        'Person on Track Detector',
                        local_manager
                    )
                    
                    if 'error' in result:
                        print(f"  ERROR: {result['error']}")
                        video_results.append({
                            'video': video_name,
                            'frame': frame_num,
                            'timestamp': timestamp,
                            'on_track': False,
                            'answer': 'ERROR',
                            'confidence': 0,
                            'reasoning': result['error']
                        })
                    elif 'person_on_track_detection' in result:
                        detection = result['person_on_track_detection']
                        
                        on_track = detection.get('person_on_track', False)
                        answer = detection.get('answer', 'UNKNOWN')
                        confidence = detection.get('confidence', 0)
                        reasoning = detection.get('reasoning', 'No reasoning')
                        detailed = detection.get('detailed_analysis', {})
                        
                        # Display results
                        status = "ON TRACK" if on_track else "CLEAR"
                        print(f"  Result: {status} ({answer})")
                        print(f"  Confidence: {confidence:.0%}")
                        print(f"  Reasoning: {reasoning}")
                        
                        # Show detailed analysis
                        if detailed:
                            print(f"  Details: Person={detailed.get('person_keywords_found', 0)}, " +
                                  f"Track={detailed.get('track_keywords_found', 0)}, " +
                                  f"Danger={detailed.get('danger_position_keywords', 0)}, " +
                                  f"Safety={detailed.get('safety_concern_keywords', 0)}")
                        
                        video_results.append({
                            'video': video_name,
                            'frame': frame_num,
                            'timestamp': timestamp,
                            'on_track': on_track,
                            'answer': answer,
                            'confidence': confidence,
                            'reasoning': reasoning,
                            'detailed_analysis': detailed
                        })
                        
                    else:
                        print(f"  Unexpected result format: {result}")
                        video_results.append({
                            'video': video_name,
                            'frame': frame_num,
                            'timestamp': timestamp,
                            'on_track': False,
                            'answer': 'UNKNOWN',
                            'confidence': 0,
                            'reasoning': 'Unknown result format'
                        })
                        
                except Exception as e:
                    print(f"  ERROR: {e}")
                    video_results.append({
                        'video': video_name,
                        'frame': frame_num,
                        'timestamp': timestamp,
                        'on_track': False,
                        'answer': 'ERROR',
                        'confidence': 0,
                        'reasoning': str(e)
                    })
            
            all_results.extend(video_results)
            
        except Exception as e:
            print(f"- Failed to process {video_name}: {e}")
            continue
    
    # Comprehensive summary
    print(f"\n" + "=" * 80)
    print("COMPREHENSIVE RESULTS SUMMARY")
    print("=" * 80)
    
    # Results table
    print(f"\nDETAILED RESULTS:")
    print("-" * 90)
    print(f"{'Video':<10} {'Frame':<6} {'Time':<6} {'On Track':<9} {'Answer':<7} {'Confidence':<11} {'Reasoning':<30}")
    print("-" * 90)
    
    total_frames = len(all_results)
    on_track_count = 0
    error_count = 0
    total_confidence = 0
    
    for result in all_results:
        video = result['video'][:8]
        frame = result['frame']
        timestamp = result['timestamp']
        on_track = "YES" if result['on_track'] else "NO"
        answer = result['answer']
        confidence = result['confidence']
        reasoning = result['reasoning'][:25] + "..." if len(result['reasoning']) > 25 else result['reasoning']
        
        print(f"{video:<10} {frame:<6} {timestamp:<6.1f} {on_track:<9} {answer:<7} {confidence:<11.0%} {reasoning:<30}")
        
        if result['on_track']:
            on_track_count += 1
        if result['answer'] == 'ERROR':
            error_count += 1
        total_confidence += confidence
    
    # Overall statistics
    print(f"\n" + "=" * 80)
    print("OVERALL PERFORMANCE")
    print("=" * 80)
    
    print(f"Total frames tested: {total_frames}")
    print(f"Videos tested: {len(test_videos)}")
    print(f"Person on track detections: {on_track_count}")
    print(f"Clear/safe detections: {total_frames - on_track_count - error_count}")
    print(f"Error responses: {error_count}")
    
    if total_frames > 0:
        detection_rate = on_track_count / total_frames * 100
        avg_confidence = total_confidence / total_frames
        error_rate = error_count / total_frames * 100
        
        print(f"Detection rate: {detection_rate:.1f}%")
        print(f"Average confidence: {avg_confidence:.0%}")
        print(f"Error rate: {error_rate:.1f}%")
    
    # Per-video breakdown
    print(f"\nPER-VIDEO ANALYSIS:")
    print("-" * 50)
    
    for video_path in test_videos:
        video_name = os.path.basename(video_path)
        video_results = [r for r in all_results if r['video'] == video_name]
        
        if video_results:
            on_track_frames = sum(1 for r in video_results if r['on_track'])
            total_video_frames = len(video_results)
            avg_video_confidence = sum(r['confidence'] for r in video_results) / len(video_results)
            
            print(f"{video_name}: {on_track_frames}/{total_video_frames} frames with person on track "
                  f"(avg confidence: {avg_video_confidence:.0%})")
    
    print(f"\n" + "=" * 80)
    print("FINAL ASSESSMENT")
    print("=" * 80)
    
    if error_rate < 10:
        print("+ EXCELLENT: Low error rate, detector is working reliably")
    elif error_rate < 25:
        print("+ GOOD: Acceptable error rate")
    else:
        print("- HIGH ERROR RATE: Needs improvement")
    
    if avg_confidence > 70:
        print("+ HIGH CONFIDENCE: Detector provides confident results")
    elif avg_confidence > 50:
        print("+ MODERATE CONFIDENCE: Results are reasonably confident")
    else:
        print("- LOW CONFIDENCE: Results may be unreliable")
    
    print(f"\nRECOMMENDATION:")
    if error_rate < 10 and avg_confidence > 70:
        print("✅ READY FOR PRODUCTION: Person on Track Detector is highly reliable")
        print("   - Use in Streamlit app for real-time track safety monitoring")
        print("   - Suitable for automated safety systems")
    elif error_rate < 25 and avg_confidence > 50:
        print("⚠️ SUITABLE WITH CAUTION: Good performance but monitor results")
        print("   - Use for preliminary screening")
        print("   - Consider human verification for critical decisions")
    else:
        print("❌ NEEDS IMPROVEMENT: Not reliable enough for production use")
        print("   - Improve keyword detection")
        print("   - Adjust confidence thresholds")
        print("   - Test with more diverse video content")
    
    print(f"\nNext steps:")
    print("1. Open http://localhost:8502")
    print("2. Select 'Person on Track Detector' from model dropdown")
    print("3. Upload test videos from test/ folder")
    print("4. Compare results with this analysis")
    
    return all_results

if __name__ == "__main__":
    test_person_on_track_final()