#!/usr/bin/env bash
# HERZA Bench — MIT License — Herza, Ltd.
set -eu
command -v python3 >/dev/null 2>&1 || { echo 'HERZA Bench requires Python 3.6+.' >&2; exit 1; }
python3 -I -c 'import sys; sys.exit(0 if sys.version_info >= (3,6) else 1)' || { echo 'Python 3.6+ required.' >&2; exit 1; }
exec python3 -I - "$@" <<'HERZA_ENGINE'
import argparse, datetime, hashlib, io, json, math, mmap, multiprocessing, os, platform, random, re, shutil, signal, statistics, struct, subprocess, sys, tempfile, time, tarfile
VERSION='1.0.0-rc.1'
MIB=1024*1024
ENDPOINTS=[('Cloudflare edge','https://speed.cloudflare.com/__down?bytes=33554432'),('Falkenstein, DE','https://fsn1-speed.hetzner.com/100MB.bin'),('Ashburn, US','https://ash-speed.hetzner.com/100MB.bin'),('Hillsboro, US','https://hil-speed.hetzner.com/100MB.bin')]
def read(path):
    try:
        with open(path) as f:return f.read().strip()
    except OSError:return ''
def run(cmd,timeout=10,**kw):
    return subprocess.run(cmd,stdout=subprocess.PIPE,stderr=subprocess.PIPE,universal_newlines=True,timeout=timeout,**kw)
def clean(s):return re.sub(r'[\x00-\x1f\x7f-\x9f]','?',str(s))
def size(n):return '{:.2f} GiB'.format(n/1024**3)
def skipped(reason):return {'status':'skipped','reason':reason}
def prime_worker(seconds,q):
    start=time.monotonic();events=0
    while time.monotonic()-start<seconds:
        for number in range(10000,11000):
            prime=True
            for divisor in range(2,math.isqrt(number)+1 if hasattr(math,'isqrt') else int(math.sqrt(number))+1):
                if number%divisor==0:prime=False;break
            if prime: events+=1
    q.put((events,time.monotonic()-start))
def cpu_test(workers):
    q=multiprocessing.Queue();children=[];start=time.monotonic()
    try:
        for _ in range(workers):
            p=multiprocessing.Process(target=prime_worker,args=(2,q));p.start();children.append(p)
        values=[q.get(timeout=12) for _ in children]
        for p in children:p.join(timeout=2)
        elapsed=time.monotonic()-start
        return {'status':'ok','workers':workers,'events_per_second':round(sum(v[0] for v in values)/elapsed,2),'elapsed_seconds':round(elapsed,3)}
    finally:
        for p in children:
            if p.is_alive():p.terminate()
            p.join(timeout=2)
        q.close();q.join_thread()
def system_info(disk_dir):
    cpu=read('/proc/cpuinfo');fields={}
    for line in cpu.splitlines():
        if ':' in line:
            k,v=line.split(':',1)
            if k.strip() not in fields:fields[k.strip()]=v.strip()
    mem={}
    for line in read('/proc/meminfo').splitlines():
        k,v=line.split(':',1);mem[k]=int(v.split()[0])*1024
    total=mem.get('MemTotal',0);available=mem.get('MemAvailable',mem.get('MemFree',0))
    swap=mem.get('SwapTotal',0);up=float(read('/proc/uptime').split()[0] or 0)
    try:affinity=len(os.sched_getaffinity(0))
    except AttributeError:affinity=os.cpu_count() or 1
    quota=None
    raw=read('/sys/fs/cgroup/cpu.max').split()
    if len(raw)==2 and raw[0]!='max':quota=float(raw[0])/float(raw[1])
    else:
        q=read('/sys/fs/cgroup/cpu/cpu.cfs_quota_us');period=read('/sys/fs/cgroup/cpu/cpu.cfs_period_us')
        if q and period and int(q)>0:quota=float(q)/float(period)
    mem_limit=read('/sys/fs/cgroup/memory.max') or read('/sys/fs/cgroup/memory/memory.limit_in_bytes')
    memory_quota=int(mem_limit) if mem_limit.isdigit() and int(mem_limit)<total else None
    virt='Unknown'
    if shutil.which('systemd-detect-virt'):
        v=run(['systemd-detect-virt'],5);virt=v.stdout.strip() or ('none detected' if v.returncode==1 else 'Unknown')
    disk=shutil.disk_usage(disk_dir);flags=fields.get('flags',fields.get('Features','')).split()
    return {'cpu_model':fields.get('model name',fields.get('Hardware','Unknown')),'logical_cpus':os.cpu_count(),'available_cpus':affinity,'cpu_quota':quota,'cpu_mhz':fields.get('cpu MHz','Unknown'),'cache':fields.get('cache size','Unknown'),'aes':bool('aes' in flags),'virtualization_extensions':bool('vmx' in flags or 'svm' in flags),'architecture':platform.machine(),'kernel':platform.release(),'os':next((x.split('=',1)[1].strip('"') for x in read('/etc/os-release').splitlines() if x.startswith('PRETTY_NAME=')),'Linux'),'memory_total':total,'memory_used':max(0,total-available),'memory_quota':memory_quota,'swap_total':swap,'swap_used':swap-mem.get('SwapFree',0),'uptime_seconds':int(up),'load_average':list(os.getloadavg()),'virtualization':virt,'tcp_congestion':read('/proc/sys/net/ipv4/tcp_congestion_control'),'disk_total':disk.total,'disk_used':disk.used,'disk_free':disk.free,'disk_path':os.path.abspath(disk_dir)}
def disk_test(directory, progress, display=None):
    if shutil.disk_usage(directory).free<256*MIB:return skipped('At least 256 MiB free space required')
    result={'status':'ok','file_mib':64,'block_mib':1,'write_method':'sequential buffered writes + fsync; repeated random 1 MiB buffer','write_mib_s':[]}
    with tempfile.TemporaryDirectory(prefix='.herza-bench-',dir=directory) as tmp:
        filename=os.path.join(tmp,'disk.bin');buf=os.urandom(MIB)
        for trial in range(3):
            progress('Disk: sequential write {}/3 (64 MiB + fsync)'.format(trial+1))
            start=time.monotonic()
            with open(filename,'wb',buffering=0) as f:
                for _ in range(64):
                    view=memoryview(buf)
                    while view:
                        written=f.write(view)
                        if not written:raise IOError('Short write')
                        view=view[written:]
                os.fsync(f.fileno())
            speed = round(64/(time.monotonic()-start), 2)
            result['write_mib_s'].append(speed)
            if display is not None:
                display.field('Write test {}'.format(trial + 1), '{:.1f} MiB/s'.format(speed))
        result['write_median_mib_s']=statistics.median(result['write_mib_s'])
        if display is not None:
            display.field('Write speed (median)', '{:.1f} MiB/s'.format(result['write_median_mib_s']))
        fd=None;aligned=None
        try:
            fd=os.open(filename,os.O_RDWR|os.O_DIRECT);aligned=mmap.mmap(-1,MIB);aligned[:]=buf
            progress('Disk: direct sequential read (64 MiB)')
            start=time.monotonic();count=0
            for _ in range(64):
                n=os.readv(fd,[aligned]);count+=n
                if n!=MIB:raise IOError('Short direct read')
            result['read_mib_s']=round(count/MIB/(time.monotonic()-start),2)
            if display is not None:
                display.field('Read speed (direct)', '{:.1f} MiB/s'.format(result['read_mib_s']))
            random_workloads = (
                ('Small-block I/O (4K)', 4096, 32768, 'random_'),
                ('Large-block I/O (1 MiB)', MIB, 512, 'large_random_'),
            )
            if display is not None:
                display.line()
            for label, block_bytes, max_operations, key_prefix in random_workloads:
                if display is not None:
                    display.line('  {:<24} {:>12} {:>12} {:>12}'.format(label, 'Ops/sec', 'MiB/s', 'p95 µs'))
                progress('Disk: {} random direct read/write (up to 2 seconds each)'.format(label))
                rng = random.Random(42)
                block_count = result['file_mib'] * MIB // block_bytes
                for direction in ('read', 'write'):
                    is_read = direction == 'read'
                    operations = 0
                    latencies = []
                    start = time.monotonic()
                    with memoryview(aligned)[:block_bytes] as view:
                        while operations < max_operations and time.monotonic() - start < 2:
                            os.lseek(fd, rng.randrange(block_count) * block_bytes, os.SEEK_SET)
                            call_start = time.monotonic()
                            transferred = os.readv(fd, [view]) if is_read else os.writev(fd, [view])
                            if transferred != block_bytes:
                                raise IOError('Short random I/O')
                            latencies.append((time.monotonic() - call_start) * 1e6)
                            operations += 1
                        if not is_read:
                            os.fsync(fd)
                    elapsed = time.monotonic() - start
                    if not latencies:
                        raise IOError('No random I/O completed')
                    latencies.sort()
                    p95_index = math.ceil(.95 * len(latencies)) - 1
                    measurement = {
                        'iops': round(operations / elapsed, 1),
                        'mib_s': round(operations * block_bytes / MIB / elapsed, 2),
                        'p95_us': round(latencies[p95_index], 1),
                        'operations': operations,
                        'elapsed_seconds': round(elapsed, 3),
                    }
                    result[key_prefix + direction] = measurement
                    if display is not None:
                        display.line('  {:<24} {:>12,.0f} {:>12.1f} {:>12.1f}'.format(
                            'Random ' + direction, measurement['iops'], measurement['mib_s'], measurement['p95_us']))
            result['random_method']='Linux O_DIRECT, 4 KiB, QD1, single process; write fsync included in throughput, not syscall latency'
            result['large_random_method'] = 'Linux O_DIRECT, 1 MiB, QD1, up to 512 operations per direction; write fsync included in throughput, not syscall latency'
        except (OSError,AttributeError) as e:
            result['direct_io_error']='{}: {}'.format(type(e).__name__,str(e));result['status']='partial'
        finally:
            if fd is not None:os.close(fd)
            if aligned is not None:aligned.close()
    return result
def curl_measure(url,limit=0,family=None,upload=None):
    if not shutil.which('curl'):raise RuntimeError('curl required for network measurements')
    cmd=['curl','--silent','--show-error','--fail','--noproxy','*','--connect-timeout','4','--max-time','12','--proto','=https','-A','HERZA-Bench/'+VERSION,'-o','/dev/null','-w','%{http_code} %{time_namelookup} %{time_connect} %{time_starttransfer} %{time_total} %{size_download} %{size_upload}']
    if family:cmd.append('-'+str(family))
    if upload:cmd+=['-H','Content-Type: application/octet-stream','-H','Expect:','--data-binary','@'+upload]
    else:cmd+=['--range','0-'+str(max(0,limit-1)),'--max-filesize',str(max(limit,1024)),'-H','Accept-Encoding: identity']
    p=run(cmd+[url],16)
    x=p.stdout.strip().split()
    timed_download=(p.returncode==28 and not upload and limit>1024 and len(x)==7 and x[0] in ('200','206') and float(x[5])>=MIB)
    if p.returncode and not timed_download:raise RuntimeError('curl exit {}'.format(p.returncode))
    if len(x)!=7 or x[0] not in ('200','206'):raise RuntimeError('Unexpected HTTP response')
    v=list(map(float,x[1:]));return {'time_limited':timed_download,'http_status':int(x[0]),'tcp_ms':max(0,(v[1]-v[0])*1000),'ttfb_ms':v[2]*1000,'seconds':v[3],'download_bytes':int(v[4]),'upload_bytes':int(v[5])}
def network_test(progress, on_node=None):
    if not shutil.which('curl'):return {'status':'skipped','reason':'curl is not installed','nodes':[]}
    result={'status':'ok','nodes':[],'method':'single HTTPS transfer, 32 MiB download / 8 MiB upload; TCP connect time excludes DNS; no ICMP test'}
    for family in (4,6):
        try:curl_measure('https://speed.cloudflare.com/__down?bytes=0',family=family);result['ipv'+str(family)]='reachable'
        except Exception:result['ipv'+str(family)]='unreachable to test endpoint'
    for name,url in ENDPOINTS:
        row={'name':name,'status':'ok','upload_status':'not supported','endpoint':url.split('?')[0]};progress('Network: '+name)
        samples=[]
        for _ in range(3):
            try:samples.append(curl_measure('https://speed.cloudflare.com/__down?bytes=0' if name=='Cloudflare edge' else url,limit=1)['tcp_ms'])
            except Exception:break
        if samples:
            row['tcp_median_ms']=round(statistics.median(samples),2)
            row['tcp_jitter_ms']=round(statistics.mean(abs(samples[i]-samples[i-1]) for i in range(1,len(samples))),2) if len(samples)>1 else None
        try:
            m=curl_measure(url,limit=32*MIB)
            if m['download_bytes']<=0:raise RuntimeError('Empty download')
            row['download_mbps']=round(m['download_bytes']*8/max(m['seconds'],.000001)/1e6,2);row['download_bytes']=m['download_bytes'];row['download_seconds']=m['seconds'];row['time_limited']=m['time_limited']
        except Exception as e:row['status']='error';row['reason']=str(e)
        if name=='Cloudflare edge':
            with tempfile.TemporaryDirectory(prefix='herza-net-') as tmp:
                p=os.path.join(tmp,'payload.bin')
                with open(p,'wb') as f:f.write(os.urandom(8*MIB))
                try:
                    m=curl_measure('https://speed.cloudflare.com/__up',upload=p)
                    if m['upload_bytes']!=8*MIB:raise RuntimeError('Incomplete upload')
                    row['upload_mbps']=round(m['upload_bytes']*8/max(m['seconds'],.000001)/1e6,2);row['upload_status']='ok';row['upload_bytes']=m['upload_bytes']
                except Exception as e:row['upload_status']='error';row['upload_error']=str(e);row['status']='partial' if row['status']=='ok' else 'error'
        result['nodes'].append(row)
        if on_node is not None:
            on_node(row)
    if any(n['status']!='ok' for n in result['nodes']):result['status']='partial'
    return result
def find_ookla(explicit=None):
    candidate=explicit or shutil.which('speedtest')
    if not candidate:return None
    try:
        v=run([candidate,'--version'],8)
        if v.returncode==0 and 'ookla' in (v.stdout+v.stderr).lower():return candidate
    except (OSError,subprocess.SubprocessError):pass
    return None

def parse_ookla(raw, selection):
    # Retain measurements and server metadata, never client IP/MAC/result URL.
    try:
        payload = json.loads(raw)
    except (ValueError, RecursionError):
        raise ValueError('Malformed Ookla JSON') from None
    if not isinstance(payload, dict) or payload.get('type') != 'result':
        raise ValueError('Expected an Ookla result object')
    for field in ('server', 'upload', 'download', 'ping'):
        if not isinstance(payload.get(field), dict):
            raise ValueError('Invalid Ookla result field: ' + field)

    def number(value, field):
        try:
            if isinstance(value, bool):
                raise ValueError
            measurement = float(value)
            if not math.isfinite(measurement) or measurement < 0:
                raise ValueError
        except (TypeError, ValueError, OverflowError):
            raise ValueError('Invalid Ookla measurement: ' + field) from None
        return measurement

    server = payload['server']
    server_id = server.get('id')
    if isinstance(server_id, str) and re.fullmatch(r'[0-9]{1,12}', server_id):
        server_id = int(server_id)
    if type(server_id) is not int or not 0 < server_id < 10**12:
        raise ValueError('Invalid Ookla server ID')

    for field in ('name', 'location', 'country'):
        if field in server and not isinstance(server[field], str):
            raise ValueError('Invalid Ookla server field: ' + field)

    packet_loss = payload.get('packetLoss')
    if packet_loss is not None:
        packet_loss = number(packet_loss, 'packet loss')
        if packet_loss > 100:
            raise ValueError('Invalid Ookla measurement: packet loss')
    return {
        'status': 'ok',
        'selection': selection,
        'name': clean(server.get('name', 'Unknown')),
        'location': clean(server.get('location', 'Unknown')),
        'country': clean(server.get('country', 'Unknown')),
        'server_id': server_id,
        'upload_status': 'ok',
        'upload_mbps': round(number(payload['upload'].get('bandwidth'), 'upload') * (8 / 1e6), 2),
        'download_mbps': round(number(payload['download'].get('bandwidth'), 'download') * (8 / 1e6), 2),
        'latency_ms': round(number(payload['ping'].get('latency'), 'latency'), 2),
        'jitter_ms': round(number(payload['ping'].get('jitter'), 'jitter'), 2),
        'packet_loss_percent': packet_loss,
    }

OOKLA_CITIES = (
    ('Singapore, SG', 13623),
    ('Kuala Lumpur, MY', 9454),
    ('Tokyo, JP', 48463),
    ('Hong Kong, HK', 32155),
    ('Paris, FR', 61933),
    ('Amsterdam, NL', 41423),
    ('London, UK', 11445),
    ('Los Angeles, US', 7190),
    ('Dallas, US', 22288),
)

def ookla_network(binary, server_ids, progress, cities=(), on_node=None):
    result = {
        'engine': 'ookla',
        'status': 'ok',
        'nodes': [],
        'method': 'Official Ookla Speedtest CLI; automatic best-server selection first, then sequential tests by server ID',
    }
    targets = [(None, 'Automatic local/best server', None)]
    targets.extend((ident, label, label.split(',')[0]) for label, ident in cities)
    seen = {ident for ident, _, _ in targets}
    for ident in server_ids:
        if ident not in seen:
            targets.append((ident, 'Server ID ' + str(ident), None))
            seen.add(ident)

    for index, (ident, label, expected_city) in enumerate(targets, 1):
        if result.get('rate_limited'):
            node = {'name': label, 'selection': label, 'server_id': ident,
                    'status': 'skipped', 'reason': 'Skipped after Ookla rate limit'}
            result['nodes'].append(node)
            if on_node is not None:
                on_node(node)
            continue
        progress('Ookla {}/{}: {} (upload + download)'.format(index, len(targets), label))
        cmd = [binary, '--format=json', '--progress=no', '--accept-license', '--accept-gdpr']
        if ident is not None:
            cmd.append('--server-id=' + str(ident))
        try:
            response = run(cmd, 150, stdin=subprocess.DEVNULL)
            if response.returncode:
                diagnostic = (response.stderr + response.stdout).lower()
                if 'too many requests' in diagnostic or 'limit reached' in diagnostic:
                    result['rate_limited'] = True
                    result['reason'] = 'Ookla rate limit reached. Try again later.'
                    raise ValueError('Ookla rate limit reached')
                raise ValueError('Speedtest failed (exit {})'.format(response.returncode))
            node = parse_ookla(response.stdout, label)
            if expected_city:
                actual_city = node['location'].split(',')[0].strip()
                if node['server_id'] != ident or actual_city.casefold() != expected_city.casefold():
                    raise ValueError('Server returned {} #{}; expected {} #{}'.format(
                        node['location'], node['server_id'], expected_city, ident))
        except (OSError, ValueError, KeyError, subprocess.SubprocessError) as exc:
            reason = 'Speedtest exceeded 150 seconds' if isinstance(exc, subprocess.TimeoutExpired) else str(exc)
            node = {'name': label, 'selection': label, 'server_id': ident,
                    'status': 'error', 'reason': reason}
        result['nodes'].append(node)
        if on_node is not None:
            on_node(node)
    if any(node['status'] != 'ok' for node in result['nodes']):
        result['status'] = 'partial'
    return result

OOKLA_ARCHIVES={
    'x86_64':('x86_64','5690596c54ff9bed63fa3732f818a05dbc2db19ad36ed68f21ca5f64d5cfeeb7'),
    'aarch64':('aarch64','3953d231da3783e2bf8904b6dd72767c5c6e533e163d3742fd0437affa431bd3')}

def prepare_ookla(directory,progress):
    arch=platform.machine().lower();arch={'amd64':'x86_64','arm64':'aarch64'}.get(arch,arch)
    if arch not in OOKLA_ARCHIVES:raise ValueError('Automatic Ookla download supports x86_64 and aarch64; use --speedtest-bin for this architecture')
    target,expected=OOKLA_ARCHIVES[arch]
    url='https://install.speedtest.net/app/cli/ookla-speedtest-1.2.0-linux-'+target+'.tgz'
    archive=os.path.join(directory,'cli.tgz');binary=os.path.join(directory,'speedtest')
    progress('Preparing temporary Ookla CLI 1.2.0 (official download + SHA-256 verification)')
    if shutil.which('curl'):
        cmd=['curl','-fSs','--proto','=https','--connect-timeout','10','--max-time','45','--max-filesize',str(8*MIB),'-o',archive,url]
    elif shutil.which('wget'):
        cmd=['wget','-q','--timeout=15','--tries=1','--max-redirect=0','-O',archive,url]
    else:raise ValueError('curl or wget is required to download Ookla CLI')
    downloaded=run(cmd,50)
    if downloaded.returncode:raise ValueError('Official Ookla download failed; retry or use --network-engine https explicitly')
    if os.path.getsize(archive)>8*MIB:raise ValueError('Ookla archive exceeds size limit')
    with open(archive,'rb') as f:digest=hashlib.sha256(f.read()).hexdigest()
    if digest!=expected:raise ValueError('Ookla SHA-256 mismatch; refusing to execute download')
    # Extract only the expected regular file, never arbitrary archive paths/links.
    with tarfile.open(archive,'r:gz') as tar:
        member=tar.getmember('speedtest')
        if not member.isfile() or not 0<member.size<=16*MIB:raise ValueError('Invalid Ookla binary in archive')
        with tar.extractfile(member) as src,open(binary,'xb') as dst:shutil.copyfileobj(src,dst)
    os.chmod(binary,0o700)
    try:
        version=run([binary,'--version'],8)
    except OSError as exc:
        raise ValueError('Cannot execute Ookla CLI in {}: {} (errno {}). Use --work-dir on a writable filesystem that permits execution.'.format(directory,exc.strerror,exc.errno))
    except subprocess.SubprocessError:
        raise ValueError('Ookla version check timed out in '+directory)
    if version.returncode!=0 or 'ookla' not in (version.stdout+version.stderr).lower():
        detail=clean((version.stderr or version.stdout).strip())[:240]
        raise ValueError('Ookla version check failed (exit {}): {}'.format(version.returncode,detail or 'no diagnostic output'))
    return binary

def selected_network(args, progress, on_node=None):
    if args.network_engine=='https':
        result=network_test(progress, on_node);result['engine']='https';return result
    cities = () if getattr(args, 'local_only', False) else OOKLA_CITIES
    binary=find_ookla(args.speedtest_bin)
    if binary:return ookla_network(binary, args.server_id, progress, cities, on_node)
    if args.speedtest_bin:
        return {'engine':'ookla','status':'error','nodes':[],'reason':'The specified official Ookla CLI could not be executed'}
    try:
        with tempfile.TemporaryDirectory(prefix='.herza-ookla-',dir=os.path.abspath(getattr(args,'work_dir','.'))) as directory:
            binary=prepare_ookla(directory,progress)
            return ookla_network(binary, args.server_id, progress, cities, on_node)
    except (OSError,ValueError,KeyError,tarfile.TarError,subprocess.SubprocessError) as exc:
        return {'engine':'ookla','status':'error','nodes':[],'reason':str(exc)}

class Display:
    def __init__(self,enabled,color):self.enabled=enabled;self.color=color
    def paint(self,s,n='36'):return '\033['+n+'m'+str(s)+'\033[0m' if self.color else str(s)
    def line(self,s=''):
        if self.enabled:print(s,flush=True)
    def section(self,title):self.line('\n'+self.paint('  '+title.upper(),'1;36'));self.line('  '+'─'*72)
    def field(self,k,v):self.line('  {:<24} {}'.format(k,self.paint(clean(v))))
def format_system(d, s):
    d.section('System overview')
    for k,v in [('CPU',s['cpu_model']),('CPU cores','{} visible / {} available'.format(s['logical_cpus'],s['available_cpus'])),('CPU frequency / cache',str(s['cpu_mhz'])+' MHz / '+s['cache']),('AES / VM extensions',('Yes' if s['aes'] else 'Not exposed')+' / '+('Yes' if s['virtualization_extensions'] else 'Not exposed')),('Memory',size(s['memory_total'])+' ('+size(s['memory_used'])+' used)'),('Swap',size(s['swap_total'])+' ('+size(s['swap_used'])+' used)'),('Test filesystem',size(s['disk_total'])+' ('+size(s['disk_used'])+' used)'),('Disk test path',s['disk_path']),('Uptime','{}d {:02d}h {:02d}m'.format(s['uptime_seconds']//86400,s['uptime_seconds']//3600%24,s['uptime_seconds']//60%60)),('Load average',' / '.join('{:.2f}'.format(x) for x in s['load_average'])),('OS',s['os']),('Architecture / kernel',s['architecture']+' / '+s['kernel']),('Virtualization',s['virtualization']),('TCP congestion',s['tcp_congestion'])]:d.field(k,v)
    if s['cpu_quota'] is not None:d.field('Cgroup CPU quota',s['cpu_quota'])
    if s['memory_quota']:d.field('Cgroup RAM limit',size(s['memory_quota']))
def format_disk_end(d, disk):
    if 'write_mib_s' not in disk:
        d.field('Status', disk.get('reason', disk['status']))
        return
    if 'read_mib_s' in disk:
        d.line()
    d.line('  Write tests include fsync. The median is the middle of 3 writes.')
    d.line('  Direct read bypasses Linux page cache. Controller/storage caches can still affect these tests.')
    if 'random_read' in disk or 'random_write' in disk:
        d.line('  Higher speed/IOPS is better. Lower p95 latency is better.')
        d.line('  p95: 95% of I/O calls completed within this time; 1000 µs = 1 ms.')
    if disk.get('direct_io_error'):
        d.field('Direct I/O unavailable', disk['direct_io_error'])

def format_network_header(d, is_ookla):
    d.section('Network · Ookla Speedtest' if is_ookla else 'Network · HTTPS transfer test')
    if is_ookla:
        d.line('  {:<26} {:>10} {:>10} {:>10} {:>10}'.format('Server','Up Mbps','Down Mbps','Latency ms','Jitter ms'))
    else:
        d.line('  {:<22} {:>11} {:>11} {:>10} {:>10}'.format('Node','Up Mbps','Down Mbps','TCP ms','Jitter ms'))

def format_network_node(d, n, is_ookla):
    if is_ookla:
        if n['status'] != 'ok':
            d.field(n['name'], n.get('reason', 'Failed'))
            return
        destination = n['selection'] if n['selection'] in dict(OOKLA_CITIES) else n['location']
        columns = [('upload_mbps','32'), ('download_mbps','36'), ('latency_ms','33'), ('jitter_ms','33')]
        values = ' '.join(d.paint(format(n[key], '>10.2f'), color) for key, color in columns)
        d.line('  ' + format(destination, '<26.26') + ' ' + values)
        return

    columns = [('upload_mbps',11,'32'), ('download_mbps',11,'36'),
               ('tcp_median_ms',10,'33'), ('tcp_jitter_ms',10,'33')]
    values = []
    for key, width, color in columns:
        value = n.get(key)
        failed = ((key == 'download_mbps' and n['status'] == 'error') or
                  (key == 'upload_mbps' and n.get('upload_status') == 'error'))
        label = '{:.2f}'.format(value) if value is not None else 'FAILED' if failed else 'N/A'
        values.append(d.paint(format(label, '>' + str(width)), '31' if label == 'FAILED' else color))
    label = n['name'] + ('*' if n.get('time_limited') else '')
    d.line('  ' + format(label, '<22') + ' ' + ' '.join(values))
    if n.get('reason'):
        d.line('    ' + d.paint('Download: ' + n['reason'], '90'))
    if n.get('upload_error'):
        d.line('    ' + d.paint('Upload: ' + n['upload_error'], '90'))

def format_network_end(d, net, is_ookla):
    if net.get('engine_note'):
        d.line('  ' + d.paint(net['engine_note'], '33'))
    if net.get('reason'):
        d.field('Status', net['reason'])
    if not is_ookla and net.get('nodes'):
        d.field('IPv4 / IPv6', net.get('ipv4','?') + ' / ' + net.get('ipv6','?'))
        d.line('  N/A upload: endpoint only provides download files (not zero speed).')
        d.line('  * = rate until the 12s deadline. Cloudflare uses anycast routing.')
        d.line('  HTTPS throughput / TCP timing are not Ookla measurements or ICMP ping.')

def format_summary(d, r):
    d.section('Run summary');d.field('Elapsed','{:.1f} seconds'.format(r['elapsed_seconds']));d.field('Timestamp',r['timestamp']);d.field('Result',r['status'].upper());d.field('Python',platform.python_version());d.line('  Local results only · herza.sh · By Herza, Ltd.\n')
def main():
    p=argparse.ArgumentParser(description='HERZA Bench — CPU, disk and network in one command')
    p.add_argument('--version',action='version',version=VERSION)
    p.add_argument('--mode',choices=['info','quick','full'],default='full')
    p.add_argument('--json',action='store_true');p.add_argument('--output');p.add_argument('--disk-dir',default='.')
    p.add_argument('--no-network',action='store_true');p.add_argument('--no-disk',action='store_true');p.add_argument('--no-color',action='store_true');p.add_argument('--color',action='store_true');p.add_argument('--workers',type=int,default=4)
    p.add_argument('--network-engine',choices=['auto','ookla','https'],default='auto')
    p.add_argument('--work-dir',default='.',help='Parent directory for temporary Ookla files; must allow execution (default: current directory)')
    p.add_argument('--speedtest-bin',help='Path to your official licensed Ookla CLI')
    p.add_argument('--local-only',action='store_true',help='Only test the automatically selected Ookla server')
    p.add_argument('--server-id',action='append',type=int,default=[],help='Additional Ookla server ID; automatic local test always runs first')
    a=p.parse_args()
    if len(a.server_id)>5 or any(i<=0 for i in a.server_id):p.error('Use up to five positive server IDs')
    if a.local_only and a.server_id:p.error('--local-only cannot be combined with --server-id')
    if a.network_engine=='https' and (a.speedtest_bin or a.server_id or a.local_only):p.error('Ookla options cannot be used with HTTPS mode')
    if platform.system()!='Linux':p.error('Linux is required')
    if not 1<=a.workers<=16:p.error('--workers must be between 1 and 16')
    if not os.path.isdir(a.disk_dir):p.error('--disk-dir must be an existing directory')
    if a.output and os.path.lexists(a.output):p.error('Output path already exists; refusing to overwrite')
    d=Display(not a.json,not a.no_color and (a.color or (sys.stdout.isatty() and not os.environ.get('NO_COLOR'))))
    d.line(d.paint('\n  HERZA BENCH  /  '+VERSION,'1;36'));d.line('  Know Your Server. Measure What Matters.');d.line("  curl -q -fsS --proto '=https' https://herza.sh/bench.sh | bash")
    start=time.monotonic();s=system_info(a.disk_dir)
    r={'schema_version':2,'tool':'HERZA Bench','version':VERSION,'timestamp':datetime.datetime.now(datetime.timezone.utc).astimezone().isoformat(),'system':s,'mode':a.mode,'tests':{},'methodology':{'cpu':'Python integer trial-division primes 10000..10999, 2s per phase, startup included','cpu_python':platform.python_version(),'disk':'64 MiB temporary file; 3 sequential writes plus fsync, direct reads, QD1 random I/O at 4 KiB and 1 MiB; random payload repeats each MiB; cache/controller/compression affect results','network':'Auto: official Ookla CLI, reused or temporarily downloaded with pinned SHA-256; automatic best-server first, then nine pinned destinations by default; HTTPS only when explicitly selected','privacy':'HERZA does not upload result JSON; providers see source IP and may store their own measurements. Client IP/MAC and provider result URLs are excluded from HERZA JSON'}}
    t = r['tests']
    t['cpu'] = skipped('info mode')
    t['disk'] = skipped('disabled')
    t['network'] = skipped('disabled')
    format_system(d, s)
    d.section('CPU · integer prime workload')
    if a.mode != 'info':
        workers = min(a.workers, s['available_cpus'], max(1, math.ceil(s['cpu_quota'])) if s['cpu_quota'] else a.workers)
        try:
            single = cpu_test(1)
            d.field('Single worker', '{:,.1f} primes/s'.format(single['events_per_second']))
            parallel = cpu_test(workers)
            d.field('Parallel ({} workers)'.format(parallel['workers']), '{:,.1f} primes/s'.format(parallel['events_per_second']))
            d.line('  primes/s = prime numbers identified per second in this test.')
            d.line('  Parallel totals all workers. Higher is faster for this workload.')
            t['cpu'] = {'status': 'ok', 'single': single, 'parallel': parallel}
        except Exception as exc:
            t['cpu'] = {'status': 'error', 'reason': type(exc).__name__}
    if t['cpu']['status'] != 'ok':
        d.field('Status', t['cpu']['reason'])

    def quiet_progress(_message):
        pass

    d.section('Storage · read and write performance')
    if a.mode == 'full' and not a.no_disk:
        try:
            t['disk'] = disk_test(a.disk_dir, quiet_progress, d if d.enabled else None)
        except Exception as exc:
            t['disk'] = {'status': 'error', 'reason': str(exc)}
    format_disk_end(d, t['disk'])

    is_ookla = a.network_engine != 'https'
    if a.mode != 'info' and not a.no_network:
        format_network_header(d, is_ookla)
        on_node = (lambda node: format_network_node(d, node, is_ookla)) if d.enabled else None
        try:
            t['network'] = selected_network(a, quiet_progress, on_node)
        except Exception as exc:
            t['network'] = {'status': 'error', 'reason': type(exc).__name__}
        format_network_end(d, t['network'], is_ookla)
    else:
        d.section('Network')
        d.field('Status', t['network']['reason'])
    r['elapsed_seconds']=round(time.monotonic()-start,3);r['status']='partial' if any(v['status'] in ('error','partial') for v in t.values()) else 'ok'
    encoded=json.dumps(r,indent=2,ensure_ascii=True)+'\n'
    if a.output:
        fd=os.open(a.output,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600)
        with os.fdopen(fd,'w') as f:f.write(encoded)
    if a.json:print(encoded,end='')
    else:format_summary(d,r)
    return 2 if r['status']=='partial' else 0
if __name__=='__main__':
    # Isolated Python ignores PYTHONIOENCODING. Keep UTF-8 on older C locales.
    for stream_name in ('stdout', 'stderr'):
        stream = getattr(sys, stream_name)
        if stream.encoding != 'utf-8':
            utf8_stream = io.TextIOWrapper(stream.detach(), encoding='utf-8',
                                           errors='backslashreplace', line_buffering=True)
            setattr(sys, stream_name, utf8_stream)
    signal.signal(signal.SIGTERM,lambda *_:sys.exit(143))
    try:sys.exit(main())
    except KeyboardInterrupt:print('\nInterrupted; temporary files cleaned.',file=sys.stderr);sys.exit(130)
HERZA_ENGINE