"""Bounded opt-in publisher timings; caller owns the publisher condition lock."""

from collections import Counter, deque
from copy import deepcopy
import math
import time


class PublisherMetrics:
    def __init__(self, clock=time.monotonic):
        self.clock = clock
        self.deadline = 0
        self.revision = 0
        self.channels = {}
        self.pending = {}
        self.active = {}

    def configure(self, enabled, seconds=30):
        if type(enabled) is not bool or type(seconds) is not int or not 1 <= seconds <= 60:
            raise ValueError('Invalid publisher measurement window.')
        self.revision += 1
        self.channels.clear()
        self.pending.clear()
        self.active.clear()
        self.deadline = self.clock() + seconds if enabled else 0

    def token(self):
        return self.revision if self.clock() < self.deadline else None

    def count(self, key, name):
        if self.token() is None:
            return
        row = self.channels.setdefault(key, dict(counts=Counter(), timings={}))
        row['counts'][name] += 1

    def _event(self, key, event):
        row = self.channels.setdefault(key, dict(counts=Counter(), timings={}))
        events = row.setdefault('events', deque(maxlen=256))
        if len(events) == events.maxlen:
            row['counts']['timeline_evicted'] += 1
        events.append(event)

    def enqueue(self, key, sequence, reason, previous, worker=0):
        token = self.token()
        if token is None:
            return
        now = int(self.clock() * 1e9)
        row = self.channels.setdefault(key, dict(counts=Counter(), timings={}))
        if reason == 'frame':
            last = row.get('last_enqueue_ns')
            if last is not None:
                self.duration(token, key, 'enqueue_gap', (now-last)/1e9)
            row['last_enqueue_ns'] = now
        if previous and previous[1] is not None:
            samples = row.setdefault('replacements', deque(maxlen=32))
            if len(samples) == samples.maxlen:
                row['counts']['replacement_samples_evicted'] += 1
            # Keep a bounded reference so the worker can fill its actual completion time.
            samples.append(dict(at_ns=now, old_sequence=self.pending.get(key, {}).get('sequence'),
                new_sequence=sequence, old_enqueued_ns=int(previous[0]*1e9), reason=reason,
                inflight=self.active.get(worker)))
        point = dict(at_ns=now, sequence=sequence, kind=reason)
        self.pending[key] = point
        self._event(key, dict(point, event='enqueue'))

    def dequeue(self, key, kind, worker=0):
        if self.token() is None:
            return None
        now = int(self.clock() * 1e9)
        queued = self.pending.pop(key, {})
        point = dict(key=key, sequence=queued.get('sequence'), kind=kind, worker=worker,
                     dequeued_ns=now, enqueued_ns=queued.get('at_ns'))
        self.active[worker] = point
        self._event(key, dict(point, event='dequeue', at_ns=now))
        return point

    def complete(self, token, point, started_ns, finished_ns, cpu_ns, accounted_ns):
        if token is None or token != self.token() or point is None:
            return
        point.update(sink_started_ns=started_ns, sink_finished_ns=finished_ns,
                     sink_cpu_ns=cpu_ns, accounted_ns=accounted_ns)
        self._event(point['key'], dict(point, event='complete', at_ns=finished_ns))
        if self.active.get(point['worker']) is point:
            self.active.pop(point['worker'])

    def duration(self, token, key, name, seconds):
        if token is None or token != self.token():
            return
        row = self.channels.setdefault(key, dict(counts=Counter(), timings={}))
        stat = row['timings'].setdefault(name, dict(count=0, total=0., maximum=0., recent=deque(maxlen=256)))
        ms = max(0., seconds * 1000)
        stat['count'] += 1
        stat['total'] += ms
        stat['maximum'] = max(stat['maximum'], ms)
        stat['recent'].append(ms)

    def snapshot(self, include_timeline=False):
        channels = {}
        for key, row in self.channels.items():
            timings = {}
            for name, stat in row['timings'].items():
                values = sorted(stat['recent'])
                timings[name] = dict(count=stat['count'], mean_ms=stat['total']/stat['count'],
                    max_ms=stat['maximum'], recent_samples=len(values),
                    recent_p95_ms=values[max(0, math.ceil(len(values)*.95)-1)])
            channels[key] = dict(counts=dict(row['counts']), timings=timings)
            if include_timeline:
                channels[key].update(timeline=deepcopy(list(row.get('events', ()))),
                    replacements=deepcopy(list(row.get('replacements', ()))))
        return dict(enabled=self.token() is not None, revision=self.revision, channels=channels,
                    timing_scope='sink_only_v2', timeline_limit_per_channel=256, replacement_limit_per_channel=32)