import ast
from pathlib import Path
import threading
import time
import unittest

from app import frame_ingress
from app.analysis_metrics import AnalysisMetrics


def collector_class():
    source = Path(frame_ingress.__file__).parent / 'multi_pipeline.py'
    node = next(n for n in ast.parse(source.read_text()).body
                if isinstance(n, ast.ClassDef) and n.name == 'ResultCollector')
    scope = {'threading': threading, 'time': time, 'PTS_TOL_NS': 1}
    exec(compile(ast.Module(body=[node], type_ignores=[]), str(source), 'exec'), scope)
    return scope['ResultCollector']


class CollectorMetricsTest(unittest.TestCase):
    def test_result_worker_reports_count_cpu_and_cleanup(self):
        from app.det_result_path import DetResultPath
        from app.frame_ingress import FrameIngress
        from app.contracts.frames import FrameEnvelope
        from app.native_metadata import NativeFrame
        ingress = FrameIngress('server', 'source', 'stream', 'run', 1)
        ingress.metrics.configure(True, 1)
        errors = []
        path = DetResultPath(ingress, None, 'RFDETRMedium', .5, collector_class()(['det']),
                             lambda: True, lambda: errors.append(True))
        try:
            path.start()
            frame = FrameEnvelope('server','source','stream','run',1,0,0,0,1280,720)
            path.submit(frame, NativeFrame(0,0,1280,720,(),576,576))
            deadline = time.monotonic() + 2
            while path.status().get('processed',0) < 1 and time.monotonic() < deadline:
                time.sleep(.005)
            status = path.status()
            self.assertEqual(1, status['processed'])
            self.assertEqual(1, status['collector_frames'])
            self.assertEqual([], path.snapshot()['objects'])
            for key in ('collector_wait','collector_work','result_work_cpu','result_lifecycle'):
                self.assertEqual(1, status['performance']['durations'][key]['samples'])
            self.assertEqual([], errors)
        finally:
            path.close(2)
        self.assertFalse(ingress.metrics.detail_enabled())

    def test_off_on_expiry_and_bounded_eviction(self):
        now = [0]
        metrics = AnalysisMetrics(clock=lambda: now[0])
        collector = collector_class()(['det'])
        self.assertFalse(metrics.detail_enabled())
        self.assertEqual(1, collector.add('det', 0, 0, {}, 1000, metrics))
        self.assertEqual({}, metrics.snapshot()['durations'])
        metrics.configure(True, 1)
        for i in range(1, 1400):
            self.assertEqual(min(i + 1, 1000), collector.add('det', i, i, {}, 1000, metrics))
        self.assertEqual(list(range(400, 1400)), list(collector.frames))
        timings = metrics.snapshot()['durations']
        for key in ('collector_wait', 'collector_work', 'collector_work_cpu',
                    'collector_evict', 'collector_evict_cpu'):
            self.assertEqual(256, timings[key]['samples'])
            self.assertGreaterEqual(timings[key]['mean_ms'], 0)
        now[0] = 1_000_000_001
        self.assertFalse(metrics.detail_enabled())
        collector.add('det', 1400, 1400, {}, 1000, metrics)
        self.assertEqual(timings, metrics.snapshot()['durations'])
        metrics.configure(False)
        self.assertEqual({}, metrics.snapshot()['durations'])

    def test_concurrent_reads_and_nonblocking_metrics(self):
        collector = collector_class()(['det'])
        metrics = AnalysisMetrics()
        metrics.configure(True)
        metrics._lock.acquire()
        worker = threading.Thread(target=lambda: collector.add('det', 1, 1, {}, 2, metrics))
        worker.start()
        worker.join(1)
        metrics._lock.release()
        self.assertFalse(worker.is_alive())
        self.assertGreater(metrics.snapshot()['observation_missed'], 0)
        with collector.lock:
            self.assertEqual([1], list(collector.frames))
        metrics.close()
        self.assertFalse(metrics.detail_enabled())


if __name__ == '__main__':
    unittest.main()