#!/usr/bin/env python3
# VICIfast Research, Report 01: The Two Bottlenecks Inside a Busy VICIdial Server
# Load rig (§7.2). FOR A THROWAWAY TEST SERVER, NOT A LIVE FLOOR. Read it before you run it.
# Per-step results of one run: answered calls lost as no-answer, people reaching agents, AMD misreads, answer-to-agent time,
# the AMI router queue, CPU, and more. Reads the exported VICIdial logs, the sampler logs, the fake carrier log and the agent log.
# Changes the server? No.
# AMI-wall rig 2026-10-23 analysis (one all-in-one dialer). Definitions follow the 10-15 rig:
#   answered      = numbers the fake carrier answers (last two digits 10-99): HUMAN if code % 8 == 0, else MACHINE
#   lost as NA    = answered at the carrier, but VICIdial logged the call NA (it never saw the answer in time)
#   answer->agent = carrier answer (sink log, ms) to the agent screen showing the call live (driver log, ms), people only
# Usage: analyze-al.py <run dir> [label]   -> <run dir>/result.json + a table on stdout
import collections, csv, datetime, glob, json, os, re, statistics as st, sys

D = sys.argv[1]
L = sys.argv[2] if len(sys.argv) > 2 else os.path.basename(D.rstrip('/'))
X = glob.glob(os.path.join(D, 'export-*'))[0]
WARM = 60
BOX_UTC_OFFSET = -4 * 3600  # the test dialer ran America/New_York (EDT on 2026-10-23); set yours


def kv(line):
    return {a: b for a, b in (x.split('=', 1) for x in line.split() if '=' in x)}


def num(v, d=None):
    try:
        return float(v)
    except Exception:
        return d


def pct(a, q):
    a = sorted(x for x in a if x is not None)
    return a[min(len(a) - 1, int(q * len(a)))] if a else None


def med(a):
    a = [x for x in a if x is not None]
    return round(st.median(a), 2) if a else None


def tsv(name):
    p = os.path.join(X, name)
    return list(csv.DictReader(open(p), delimiter='\t')) if os.path.exists(p) else []


def expected(phone):
    c = int(phone[-2:])
    if c < 6:
        return 'NOANSWER'
    if c < 8:
        return 'CONGESTION'
    if c == 8:
        return 'CHANUNAVAIL'
    if c == 9:
        return 'BUSY'
    return 'HUMAN' if c % 8 == 0 else 'MACHINE'


# ---- steps
steps, end = [], None
for l in open(os.path.join(X, 'steps.log')):
    t, rest = l.split(' ', 1)
    m = re.search(r'step cc=(\d+) agents=(\d+) level=([\d.]+)', rest)
    if m:
        steps.append({'cc': int(m[1]), 'agents': int(m[2]), 'level0': float(m[3]), 't0': float(t)})
    if ('dialing stopped' in rest or 'abort' in rest) and end is None:
        end = float(t)
        abort = rest.strip() if 'abort' in rest else None
for i, s in enumerate(steps):
    s['t1'] = steps[i + 1]['t0'] if i + 1 < len(steps) else end

# ---- samples
S = [kv(l) for l in open(os.path.join(X, 'sample.log'))]
for d in S:
    d['t'] = num(d.get('t'))
    d['calls_vd'] = sum(int(v) for k, v in d.items() if k.startswith('ac_'))
SINK = [kv(l) for l in open(os.path.join(D, 'sink', 'sample.log'))] if os.path.exists(os.path.join(D, 'sink', 'sample.log')) else []
AGS = [kv(l) for l in open(os.path.join(D, 'agents', 'sample.log'))] if os.path.exists(os.path.join(D, 'agents', 'sample.log')) else []
for arr in (SINK, AGS):
    for d in arr:
        d['t'] = num(d.get('t'))

CTRL = []
if os.path.exists(os.path.join(X, 'ctrl.log')):
    for l in open(os.path.join(X, 'ctrl.log')):
        v = l.split()
        CTRL.append({'t': float(v[0]), **kv(' '.join(v[1:]))})

VL = tsv('vicidial_log.tsv')
CL = tsv('carrier_log.tsv')
DROP = tsv('drop_log.tsv')

sink_ans = {}
p = os.path.join(D, 'sink', 'lt-messages.log')
if os.path.exists(p):
    for l in open(p):
        m = re.match(r'\[(\d{4}-\d\d-\d\d \d\d:\d\d:\d\d\.\d{3})\].*LTANS 1(\d{10}) ([HM])', l)
        if m:
            t = datetime.datetime.strptime(m[1], '%Y-%m-%d %H:%M:%S.%f').replace(tzinfo=datetime.timezone.utc).timestamp()
            sink_ans[m[2]] = (t, m[3])

AG = [json.loads(l) for l in open(os.path.join(D, 'agents', 'agents.jsonl')) if l.strip()] if os.path.exists(os.path.join(D, 'agents', 'agents.jsonl')) else []
live = [e for e in AG if e['ev'] == 'live']
http = [e for e in AG if e['ev'] == 'http']
rtc = collections.defaultdict(list)
for e in AG:
    if e['ev'] == 'rtc':
        rtc[e['agent']].append(e)

TPW = []
p = os.path.join(X, 'tp-warn.txt')
if os.path.exists(p):
    for l in open(p):
        m = re.match(r'\[(\d{4}-\d\d-\d\d \d\d:\d\d:\d\d)', l)
        if m and ('queue reached' in l or 'High water' in l or 'high water' in l):
            t = datetime.datetime.strptime(m[1], '%Y-%m-%d %H:%M:%S').replace(tzinfo=datetime.timezone.utc).timestamp() - BOX_UTC_OFFSET
            TPW.append((t, l.strip()[:160]))

res = {'label': L, 'abort': locals().get('abort'), 'steps': []}
for s in steps:
    lo, hi = s['t0'] + WARM, s['t1']
    inside = lambda t: s['t0'] <= num(t, 0) < s['t1']
    pts = [d for d in S if d['t'] and lo <= d['t'] < hi]
    r = {'cc': s['cc'], 'secs': round(s['t1'] - s['t0'])}
    g = lambda k, f=med: f([num(d.get(k)) for d in pts]) if pts else None
    r['calls_med'] = g('calls_vd')
    r['ast_calls_med'] = g('calls')
    r['chans_med'] = g('chans')
    r['cpu_pct'] = g('cpu')
    r['steal_pct'] = g('steal')
    r['cores'] = {k: g('c_' + k) for k in ('ast', 'db', 'vd', 'web', 'other')}
    r['daemons_pct_of_core'] = {k: g('d_' + k) for k in ('listen', 'ami2', 'send', 'autodial')}
    r['daemons_pct_of_core_max'] = {k: g('d_' + k, lambda a: max([x for x in a if x is not None], default=None)) for k in ('listen', 'ami2', 'autodial')}
    r['ast_top_thread_pct'] = g('thrmax')
    fns = collections.Counter(d.get('thrmaxfn') for d in pts)
    r['ast_top_thread_fn'] = fns.most_common(1)[0][0] if fns else None
    r['ast_rss_mb_max'] = g('astrssMB', lambda a: max([x for x in a if x is not None], default=None))
    q = [num(d.get('mgrq')) for d in pts]
    r['mgr_queue'] = {'p50': med(q), 'p95': pct(q, .95), 'max': max([x for x in q if x is not None], default=None)}
    r['mgr_maxdepth_end'] = num(pts[-1].get('mgrmax')) if pts else None
    pr = [(d['t'], num(d.get('mgrproc'))) for d in pts if d.get('mgrproc')]
    r['mgr_msgs_per_s'] = round((pr[-1][1] - pr[0][1]) / (pr[-1][0] - pr[0][0])) if len(pr) > 1 and pr[-1][0] > pr[0][0] else None
    lvl = [num(c.get('level')) for c in CTRL if lo <= c['t'] < hi]
    r['dial_level_med'] = med(lvl)
    cl = [x for x in CL if inside(x['t'])]
    r['attempts'] = len(cl)
    r['cps'] = round(len(cl) / (s['t1'] - s['t0']), 1)
    r['dialstatus'] = dict(collections.Counter(x['dialstatus'] for x in cl))
    vl = [x for x in VL if inside(x['t'])]
    r['status'] = dict(collections.Counter(x['status'] for x in vl).most_common(14))
    conf = collections.Counter()
    for x in vl:
        ph = x.get('phone_number') or ''
        if len(ph) == 10:
            conf[(expected(ph), x['status'])] += 1
    ans = sum(v for (e, _), v in conf.items() if e in ('HUMAN', 'MACHINE'))
    lost = sum(v for (e, s2), v in conf.items() if e in ('HUMAN', 'MACHINE') and s2 == 'NA')
    humans = sum(v for (e, _), v in conf.items() if e == 'HUMAN')
    r['answered'] = ans
    r['lost_as_na'] = lost
    r['lost_as_na_pct'] = round(100 * lost / ans, 2) if ans else None
    r['humans'] = humans
    r['human_lost_as_na'] = conf.get(('HUMAN', 'NA'), 0)
    r['human_to_agent'] = sum(v for (e, s2), v in conf.items() if e == 'HUMAN' and s2 not in ('AA', 'DROP', 'AB', 'ADC', 'ADCT', 'NA', 'PDROP', 'AL', 'AM'))
    r['human_dropped'] = sum(v for (e, s2), v in conf.items() if e == 'HUMAN' and s2 in ('DROP', 'PDROP'))
    r['human_dropped_pct'] = round(100 * r['human_dropped'] / humans, 2) if humans else None
    r['amd_human_as_machine'] = conf.get(('HUMAN', 'AA'), 0)
    r['amd_machine_as_human'] = sum(v for (e, s2), v in conf.items() if e == 'MACHINE' and s2 not in ('AA', 'NA'))
    r['confusion'] = {f'{e}->{s2}': v for (e, s2), v in conf.most_common(16)}
    lat = []
    for e in live:
        if not inside(e['t'] / 1000):
            continue
        a = sink_ans.get(e['phone'])
        if a and a[1] == 'H':
            lat.append(e['t'] / 1000 - a[0])
    r['answer_to_agent_s'] = {'n': len(lat), 'p50': round(pct(lat, .5), 2) if lat else None, 'p95': round(pct(lat, .95), 2) if lat else None,
                              'max': round(max(lat), 2) if lat else None}
    hs = collections.defaultdict(lambda: [0, [], 0])
    for e in http:
        if lo <= e['win'] / 1000 < hi:
            h = hs[e['ep']]; h[0] += e['n']; h[2] += e['fail']
            if e['p95'] is not None:
                h[1].append(e['p95'])
    r['agent_http_p95_ms'] = {ep: {'med': pct(v[1], .5), 'worst': max(v[1]) if v[1] else None, 'fail': v[2]} for ep, v in hs.items()
                              if ep in ('vdc_db_query', 'conf_exten_check', 'manager_send')}
    rx = lost_p = 0
    for a, evs in rtc.items():
        pts2 = [e for e in evs if lo <= e['t'] / 1000 < hi]
        if len(pts2) > 1:
            rx += max(0, pts2[-1]['rx'] - pts2[0]['rx']); lost_p += max(0, pts2[-1]['lost'] - pts2[0]['lost'])
    r['agent_webrtc_loss_pct'] = round(100 * lost_p / (rx + lost_p), 3) if rx + lost_p else None
    sk = [d for d in SINK if d['t'] and lo <= d['t'] < hi]
    r['sink'] = {'cpu': med([num(d.get('cpu')) for d in sk]), 'chans': med([num(d.get('chans')) for d in sk]),
                 'udprcvbuf': sum(num(d.get('udprcvbuf'), 0) for d in sk)}
    ag = [d for d in AGS if d['t'] and lo <= d['t'] < hi]
    r['agentsbox_cpu'] = med([num(d.get('cpu')) for d in ag])
    r['udprcvbuf_dialer'] = sum(num(d.get('udprcvbuf'), 0) for d in pts)
    r['tp_highwater_warnings'] = sum(1 for t, _ in TPW if s['t0'] <= t < s['t1'])
    res['steps'].append(r)

res['tp_warnings_sample'] = [w for _, w in TPW[:5]]
res['mgr_final'] = open(os.path.join(X, 'mgr-final.txt')).read().strip().splitlines()[-1] if os.path.exists(os.path.join(X, 'mgr-final.txt')) else None
json.dump(res, open(os.path.join(D, 'result.json'), 'w'), indent=1)

print(f"== {L}  abort={res['abort']}  final router: {res['mgr_final']}")
print('target calls  cps  lvl  | lostNA%  (n/ans)  hDrop%  a2agent p50/p95  | mgrQ p50/p95/max  maxdepth msgs/s | cpu% ast db vd | listen ami2 autodial(%core, max) | topthr% | rss')
for r in res['steps']:
    a2 = r['answer_to_agent_s']; q = r['mgr_queue']; dm = r['daemons_pct_of_core']; dx = r['daemons_pct_of_core_max']
    print(f"{r['cc']:>5} {r['calls_med']!s:>6} {r['cps']:>4} {r['dial_level_med']!s:>5} | {r['lost_as_na_pct']!s:>6} ({r['lost_as_na']}/{r['answered']}) "
          f"{r['human_dropped_pct']!s:>5} {a2['p50']!s:>5}/{a2['p95']!s:<5} | {q['p50']!s:>5}/{q['p95']!s:>6}/{q['max']!s:>7} {r['mgr_maxdepth_end']!s:>7} {r['mgr_msgs_per_s']!s:>5} | "
          f"{r['cpu_pct']!s:>5} {r['cores']['ast']!s:>5} {r['cores']['db']!s:>4} {r['cores']['vd']!s:>4} | {dm['listen']!s:>4} {dm['ami2']!s:>4} {dm['autodial']!s:>4} ({dx['listen']!s},{dx['ami2']!s},{dx['autodial']!s}) | "
          f"{r['ast_top_thread_pct']!s:>4} {(r['ast_top_thread_fn'] or '')[:18]} | {r['ast_rss_mb_max']} tpw={r['tp_highwater_warnings']}")