| """Provider-defined physical features, fixed splits, and source-only discretization. |
| |
| CPU work is restricted to archive I/O/parsing and provenance. All feature |
| aggregation, preprocessing, probability estimation and inference use CUDA. |
| """ |
| from __future__ import annotations |
|
|
| import hashlib |
| import io |
| import itertools |
| import json |
| from pathlib import Path |
| import time |
| import urllib.request |
| import zipfile |
|
|
| import numpy as np |
| import torch |
|
|
| ROOT = Path('artifacts/external_decision_sufficiency_01') |
| PROTOCOL = Path('configs/external_sufficiency_protocol_01.json') |
|
|
|
|
| def sha(path): |
| h = hashlib.sha256() |
| with Path(path).open('rb') as f: |
| while chunk := f.read(8 << 20): |
| h.update(chunk) |
| return h.hexdigest() |
|
|
|
|
| def write(path, value): |
| Path(path).parent.mkdir(parents=True, exist_ok=True) |
| Path(path).write_text(json.dumps(value, indent=2)+'\n') |
|
|
|
|
| def split_group(name): |
| return int(hashlib.sha256(('20101:'+name).encode()).hexdigest(), 16) % 2 |
|
|
|
|
| def acquire(protocol): |
| entries = [] |
| for name, spec in protocol['sources'].items(): |
| path = ROOT/'sources'/f'{name}.zip' |
| assert not path.exists(), 'A source archive was already opened/downloaded.' |
| path.parent.mkdir(parents=True, exist_ok=True) |
| start = time.perf_counter() |
| urllib.request.urlretrieve(spec['url'], path) |
| elapsed = time.perf_counter()-start |
| with zipfile.ZipFile(path) as z: |
| inventory = [{'name':i.filename,'uncompressed_bytes':i.file_size, |
| 'compressed_bytes':i.compress_size,'crc32':i.CRC} |
| for i in z.infolist()] |
| entries.append(dict(name=name, url=spec['url'],doi=spec['doi'],license=spec['license'], |
| sha256=sha(path),bytes=path.stat().st_size, |
| download_seconds=elapsed,inventory=inventory)) |
| print(json.dumps({'downloaded':name,'bytes':path.stat().st_size,'seconds':elapsed}), flush=True) |
| write(ROOT/'sources/manifest.json', {'archives':entries,'downloaded_after_lock':True}) |
|
|
|
|
| def matrix(z, name): |
| candidates = [n for n in z.namelist() if n.endswith('/'+name) or n == name] |
| assert len(candidates) == 1, (name, candidates) |
| |
| return np.loadtxt(io.BytesIO(z.read(candidates[0])), dtype=np.float64) |
|
|
|
|
| def hydraulic(protocol, device): |
| spec = protocol['sources']['hydraulic'] |
| with zipfile.ZipFile(ROOT/'sources/hydraulic.zip') as z: |
| profile_cpu = matrix(z, 'profile.txt').astype(np.int64) |
| profile = torch.as_tensor(profile_cpu, device=device) |
| columns = [] |
| for name in spec['features']: |
| raw = torch.as_tensor(matrix(z, name+'.txt'), device=device) |
| assert raw.shape[0] == len(profile) |
| columns.append(raw.mean(1)) |
| del raw |
| x = torch.stack(columns, 1) |
| groups = [','.join(map(str, p[:4])) for p in profile_cpu] |
| side = torch.tensor([split_group(g) for g in groups], device=device) |
| tasks = [] |
| for t, task in enumerate(spec['tasks']): |
| context = spec['context_column'][t] |
| source = profile[:, context] == spec['source_context'][t] |
| h = (profile[:, t] != spec['normal'][t]).long() |
| for value in spec['target_contexts'][t]: |
| target = profile[:, context] == value |
| tasks.append(dict(dataset='hydraulic',environment=f'context{context}={value}', |
| task=task,x=x,h=h,source=source, |
| adaptation=target & (side == 0),evaluation=target & (side == 1), |
| groups=groups,record_ids=list(range(len(x))), |
| context_description=f'profile column {context}; source optimal, target {value}')) |
| return tasks |
|
|
|
|
| def gas(protocol, device): |
| spec = protocol['sources']['gas'] |
| arrays, labels, batches, groups, ids = [], [], [], [], [] |
| with zipfile.ZipFile(ROOT/'sources/gas.zip') as z: |
| for batch in spec['source_batches']+spec['target_batches']: |
| names = [n for n in z.namelist() if n.endswith(f'batch{batch}.dat')] |
| assert len(names) == 1 |
| rows, hs = [], [] |
| for i, line in enumerate(z.read(names[0]).decode().splitlines()): |
| tokens = line.split() |
| if not tokens: |
| continue |
| hs.append(int(tokens[0])) |
| values = dict((int(v.split(':')[0])-1,float(v.split(':')[1])) for v in tokens[1:]) |
| rows.append([values[j] for j in spec['features_zero_based']]) |
| groups.append(f'batch{batch}:block{i//16}') |
| ids.append(f'batch{batch}:row{i}') |
| arrays.extend(rows); labels.extend(hs); batches.extend([batch]*len(rows)) |
| x = torch.tensor(arrays, device=device, dtype=torch.float64) |
| labels = torch.tensor(labels, device=device) |
| batches = torch.tensor(batches, device=device) |
| side = torch.tensor([split_group(g) for g in groups], device=device) |
| source_batch = torch.zeros_like(side, dtype=torch.bool) |
| for batch in spec['source_batches']: |
| source_batch |= batches == batch |
| tasks = [] |
| for a, b in itertools.combinations(range(1, 7), 2): |
| relevant = (labels == a) | (labels == b) |
| for batch in spec['target_batches']: |
| target = relevant & (batches == batch) |
| tasks.append(dict(dataset='gas',environment=f'batch{batch}',task=f'gas{a}_vs_{b}', |
| x=x,h=(labels == b).long(),source=source_batch & relevant, |
| adaptation=target & (side == 0),evaluation=target & (side == 1), |
| groups=groups,record_ids=ids, |
| context_description='provider temporal batch; source batches 1,2,3')) |
| return tasks |
|
|
|
|
| def encode_records(x, h, threshold): |
| assert x.device.type == 'cuda' and threshold.device.type == 'cuda' |
| binary = (x > threshold).long() |
| |
| return h*256+(binary*x.new_tensor([128,64,32,16,8,4,2,1],dtype=torch.long)).sum(1) |
|
|
|
|
| def prepare(protocol): |
| device = protocol['device'] |
| start = time.perf_counter() |
| tasks = hydraulic(protocol, device)+gas(protocol, device) |
| torch.cuda.synchronize(device) |
| parse_seconds = time.perf_counter()-start |
| destination = ROOT/'prepared' |
| destination.mkdir(parents=True, exist_ok=True) |
| summaries, excluded = [], [] |
| for index, task in enumerate(tasks): |
| start = time.perf_counter() |
| masks = [task[k] for k in ('source','adaptation','evaluation')] |
| ns = [int(m.sum()) for m in masks] |
| evaluation_h = task['h'][masks[2]] |
| counts = [int((evaluation_h == h).sum()) for h in (0,1)] |
| group_sets = [set(g for g, keep in zip(task['groups'],m.tolist()) if keep) for m in masks] |
| assert not (group_sets[1] & group_sets[2]), 'Target configuration/block leakage' |
| reasons = [] |
| eligibility = protocol['eligibility'] |
| if ns[0] < eligibility['source_minimum']: reasons.append('source_support') |
| if ns[1] < 64: reasons.append('adaptation_support') |
| if ns[2] < eligibility['evaluation_minimum']: reasons.append('evaluation_support') |
| if min(counts, default=0) < eligibility['evaluation_minimum_per_hypothesis']: reasons.append('evaluation_hypothesis_support') |
| if len(group_sets[2]) < eligibility['eligible_groups_minimum']: reasons.append('evaluation_group_support') |
| summary = dict(index=index,dataset=task['dataset'],environment=task['environment'],task=task['task'], |
| counts=dict(zip(('source','adaptation','evaluation'),ns)), |
| evaluation_hypothesis_counts=counts,group_counts=[len(g) for g in group_sets], |
| exclusions=reasons,context_description=task['context_description']) |
| if reasons: |
| excluded.append(summary) |
| continue |
| threshold = task['x'][masks[0]].median(0).values |
| events = encode_records(task['x'], task['h'], threshold) |
| arrays = {k:events[m] for k,m in zip(('source','adaptation','evaluation'),masks)} |
| |
| provenance = {k:[dict(record_id=i,group=g) for i,g,keep in |
| zip(task['record_ids'],task['groups'],m.tolist()) if keep] |
| for k,m in zip(('source','adaptation','evaluation'),masks)} |
| assert torch.isfinite(task['x']).all() |
| torch.cuda.synchronize(device) |
| summary['preprocessing_seconds'] = time.perf_counter()-start |
| summary['thresholds'] = threshold.tolist() |
| summary['file'] = f'task_{index:03}.pt' |
| torch.save({**arrays,'thresholds':threshold},destination/summary['file']) |
| write(destination/f'task_{index:03}_provenance.json',provenance) |
| summaries.append(summary) |
| write(destination/'manifest.json',dict(tasks=summaries,excluded=excluded, |
| total_declared_tasks=len(tasks),raw_parse_and_aggregation_seconds=parse_seconds, |
| fixed_binary_adapter=True,provider_defined_sequential_task=False, |
| target_configuration_or_contiguous_block_leakage=False, |
| pipeline_note='All recorded future bits are evaluator/archival payloads. Policy construction sees only adaptation/source samples.')) |
| print(json.dumps({'prepared':len(summaries),'excluded':len(excluded),'parse_seconds':parse_seconds}),flush=True) |
|
|
