"""EVART-01: child-owned compressed cache and asynchronous durable Event admission."""
from collections import Counter, deque
import ctypes
import json
import os
from pathlib import Path
import queue
import struct
import threading
import time

from app.event_artifact_contract import ArtifactConfig
from app.event_artifact_store import ArtifactStore
from app.protected_packet_cache import ProtectedPacketCache

PACKET = struct.Struct('<QQQIIII')
MAX_PACKET = 2 * 1024 ** 2 + 65536 + PACKET.size
MAX_CACHE = 32 * 1024 ** 2
LIBRARY = '/opt/nvidia/deepstream/deepstream/lib/gst-plugins/libcontextra_rtp_time.so'


class EncodedCache:
    def __init__(self):
        self.lib = ctypes.CDLL(LIBRARY)
        self.lib.contextra_encoded_enable.argtypes = [ctypes.c_int]
        self.lib.contextra_encoded_pull.argtypes = [ctypes.c_void_p, ctypes.c_size_t]
        self.lib.contextra_encoded_pull.restype = ctypes.c_long
        self.lib.contextra_encoded_dropped.restype = ctypes.c_ulonglong
        self.buffer = ctypes.create_string_buffer(MAX_PACKET)
        self.lib.contextra_encoded_enable(1)

    def pull(self):
        length = self.lib.contextra_encoded_pull(self.buffer, len(self.buffer))
        if length < 0:
            raise RuntimeError('Compressed cache packet exceeded contract.')
        return self.buffer.raw[:length] if length else None

    def close(self):
        self.lib.contextra_encoded_enable(0)


class EventCapture:
    def __init__(self, config, camera_id, *, provider_key=None, training=None):
        self.config, self.camera_id = config, camera_id
        self.provider_key = provider_key
        self.store = ArtifactStore(config) if config and config.enabled else None
        self.training = training
        self.native = EncodedCache()
        self.results = queue.Queue(maxsize=128)
        self.cache = ProtectedPacketCache()
        self.packets = self.cache.packets
        self.cache_status = self.cache.status()
        self.metadata = deque(maxlen=300)
        self.pending = {}
        self.bytes = 0
        self.metrics = Counter()
        self.error = None
        self.stop = threading.Event()
        self.worker = threading.Thread(target=self._run, name='event-capture', daemon=True)
        self.worker.start()

    @classmethod
    def from_environment(cls, spec):
        from app.capture_trial import NoCaptureTrial, no_capture_enabled
        if no_capture_enabled():
            return NoCaptureTrial()
        config = ArtifactConfig.from_environment()
        region_enabled = any((spec.get(name) or {}).get('enabled', False)
                             for name in ('intrusion', 'loitering', 'fall', 'fight'))
        from app.fire_config import scenario_enabled
        from app.training_capture import TrainingCapture
        try:
            training = TrainingCapture(spec)
        except Exception as error:
            print('[TRAINING] INIT_' + type(error).__name__, flush=True)
            training = None
        from app.crowd_config import CrowdConfig
        crowd = CrowdConfig.parse(spec.get('crowd') or {})
        event_enabled = config and config.enabled and (region_enabled or scenario_enabled(spec)
                                                       or (crowd.enabled and crowd.event_enabled))
        if not event_enabled and (training is None or training.store is None):
            return None
        context = spec.get('_provider_runtime') or {}
        provider = context.get('provider')
        if provider is None:
            from app.provider_delivery import runtime_plan
            selected = runtime_plan(spec).providers
            if len(selected) == 1:
                provider = selected[0]
        return cls(config if event_enabled else None, spec.get('camera_id', spec['source_id']),
                   provider_key=provider.key if provider else None, training=training)

    def submit(self, result):
        if self.error or self.stop.is_set():
            self.metrics['unaccepted_events'] += len(result.events)
            return False
        try:
            self.results.put_nowait(result)
            return True
        except queue.Full:
            self.metrics['metadata_queue_drop'] += 1
            self.metrics['unaccepted_events'] += len(result.events)
            return False

    def status(self):
        return dict(self.metrics, error=self.error, pending=len(self.pending),
                    queued=self.results.qsize(), native_drop=self.native.lib.contextra_encoded_dropped(),
                    event_time_fallback=getattr(self, 'time_fallback_diagnostic', None),
                    protected_cache=self.cache_status,
                    training=self.training.status() if self.training else None)

    def _collect(self):
        for _ in range(128):
            packet = self.native.pull()
            if packet is None:
                break
            pts = PACKET.unpack_from(packet)[0]
            if self.packets and pts < self.packets[-1][0]:
                # PTS reordering is legal for B-frames; decoder order is retained.
                self.metrics['packet_pts_reordered'] += 1
            self.cache.append(pts, packet, bool(PACKET.unpack_from(packet)[5]))
            self.bytes = self.cache.bytes
        for _ in range(128):
            try:
                result = self.results.get_nowait()
            except queue.Empty:
                break
            snapshot = result.snapshot()
            if result.camera_id is None:
                if str(snapshot['cameraId']) != str(self.camera_id):
                    raise ValueError('Legacy capture camera identity differs.')
                snapshot['cameraId'] = self.camera_id
            elif type(snapshot['cameraId']) is not type(self.camera_id) or snapshot['cameraId'] != self.camera_id:
                raise ValueError('Capture camera identity differs.')
            self.metadata.append(snapshot)
            if self.training:
                self.training.consider(result)
            for event in result.events:
                if self.store is None:
                    continue
                payload = event.payload()
                payload['cameraId'] = self.camera_id
                payload['scenarioConfig'] = snapshot.get('scenarioConfig')
                from app.event_time_fallback import event_diagnostic, fallback_time
                diagnostic = event_diagnostic(event.frame, getattr(self, 'provider_key', None))
                if diagnostic is not None:
                    payload['eventTimeFallback'] = diagnostic
                    if fallback_time(payload) is not None:
                        self.metrics['event_time_fallback_candidates'] += 1
                        self.time_fallback_diagnostic = diagnostic
                        if self.metrics['event_time_fallback_candidates'] == 1:
                            print('[EVENT_TIME] ' + json.dumps(diagnostic, ensure_ascii=True), flush=True)
                accepted, reason = self.store.accept(payload)
                self.metrics['duplicate' if not accepted else 'accepted'] += 1
                if accepted and reason:
                    self.metrics['artifact_admission_failed'] += 1
                if accepted and not reason:
                    error = self.cache.protect(event.event_id, payload['ptsNs'])
                    self.pending[event.event_id] = (payload, time.monotonic()+6, error)
        self.cache_status = self.cache.status()

    def _save(self, event, reason, capture_error=None):
        event_id, pts = event['eventId'], event['ptsNs']
        capture_error = capture_error or self.cache.failures.get(event_id)
        if capture_error:
            self.store.captured(event_id, capture_error)
            return
        if type(pts) is not int or not self.packets:
            self.store.captured(event_id, 'NO_COMPRESSED_VIDEO')
            return
        lo, hi = max(0, pts-3_000_000_000), pts+3_000_000_000
        if event_id not in self.cache.pins:
            self.store.captured(event_id, 'NO_PRE_EVENT_KEYFRAME')
            return
        chosen = self.cache.rows(event_id, hi)
        stage = self.config.root / '.staging' / event_id
        stage.mkdir(mode=0o700, exist_ok=False)
        with (stage / 'capture.bin').open('xb') as target:
            for _, raw in chosen:
                target.write(struct.pack('<I', len(raw)))
                target.write(raw)
            target.flush()
            os.fsync(target.fileno())
        job = dict(event=event, startPtsNs=lo, endPtsNs=hi,
                   captureReason=reason, nativeDrop=self.native.lib.contextra_encoded_dropped(),
                   metadataQueueDrop=self.metrics['metadata_queue_drop'],
                   metadata=[s for s in self.metadata if s['ptsNs'] is not None and lo <= s['ptsNs'] <= hi])
        if getattr(self, 'provider_key', None) is not None:
            job['captureIdentity'] = dict(cameraId=self.camera_id, runId=event['runId'],
                                          providerKey=self.provider_key, eventId=event_id)
        with (stage / 'job.json').open('x') as target:
            json.dump(job, target, ensure_ascii=False, separators=(',', ':'))
            target.flush()
            os.fsync(target.fileno())
        self.store.captured(event_id)
        self.metrics['capture_ready'] += 1

    def _run(self):
        try:
            while True:
                self._collect()
                stopped = self.stop.is_set()
                if self.training:
                    try:
                        self.training.collect(self.packets, stopped=stopped)
                    except Exception as error:
                        self.metrics['training_error_'+type(error).__name__] += 1
                last_pts = max((p for p, _ in self.packets), default=-1)
                self.cache.expire()
                for event_id, (event, deadline, capture_error) in list(self.pending.items()):
                    complete = event['ptsNs'] is not None and last_pts >= event['ptsNs']+3_000_000_000
                    if capture_error or event_id in self.cache.failures or stopped or complete or time.monotonic() >= deadline:
                        try:
                            self._save(event, 'COMPLETE' if complete else 'STOP_PARTIAL' if stopped else 'POST_TIMEOUT_PARTIAL', capture_error)
                        finally:
                            self.cache.release(event_id)
                        del self.pending[event_id]
                self.cache_status = self.cache.status()
                if stopped:
                    return
                self.stop.wait(.05)
        except Exception as exc:
            self.error = 'EVENT_CAPTURE_' + type(exc).__name__.upper()
            print('[ARTIFACT] ' + self.error, flush=True)
            for event_id in self.pending:
                try:
                    self.store.captured(event_id, self.error)
                except Exception:
                    pass  # Durable CAPTURING expires in the parent, never vanishes.
        finally:
            self.cache.clear()
            self.cache_status = self.cache.status()
            self.pending.clear()
            self.native.close()
            if self.store is not None:
                self.store.close()

    def close(self, timeout):
        self.stop.set()
        self.worker.join(max(0, timeout))
        return not self.worker.is_alive()