diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..88f76cf --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +.venv/ +weights/ +*.partial +*.log diff --git a/README.md b/README.md new file mode 100644 index 0000000..779b5ec --- /dev/null +++ b/README.md @@ -0,0 +1,21 @@ +# Model parameter similarity experiments + +Bounded, inference-free analysis of official unquantized model weights. The first experiment +is **Qwen3.5-2B**. Model weights are downloaded and analyzed on the authorized remote host, +then deleted after archived results pass validation. No model weights are committed here. + +- [Procedure and limits](docs/SOP.md) +- `pilot/download.py`: revision-pinned official checkpoint download and verification. +- `pilot/tensors.py`: memory-mapped safetensors/BF16 reader and metrics. +- `pilot/analyze.py`: bounded tensor/block search, exact verification, representative groups. +- `pilot/plot.py`: scientific PNG/PDF/SVG figures and report from saved data. +- `pilot/validate.py`: result/group validation, checksums and guarded weight cleanup. +- `pilot/run.py`: one-model supervisor with bounded download wait and analysis timeouts. +- `tests/test_pilot.py`: synthetic numerical and 3D correctness tests. + +The primary result is **verified net byte savings versus direct relative weight error**. +This does not demonstrate preserved task capability or faster inference. Different block +scales are alternative partitions and their savings must not be summed. + +Dependencies are pinned in `requirements.txt`; install only prebuilt wheels on the remote host. +Full resolved versions and run provenance accompany each result. See the SOP for commands. diff --git a/docs/SOP.md b/docs/SOP.md new file mode 100644 index 0000000..d2ab99e --- /dev/null +++ b/docs/SOP.md @@ -0,0 +1,90 @@ +# Parameter similarity SOP — 2B pilot draft + +Status: implementation draft; findings and revisions are added after the first real run. + +## Scope and decision + +Analyze official, unquantized safetensors without instantiating a model, running inference, +training, or executing repository-supplied code. Start with Qwen/Qwen3.5-2B only. +The question is the number of representative-centered sharing groups under direct weight +replacement error limits, and the net byte savings of a theoretical shared representation. +Weight error is not task-accuracy loss, and parameter sharing does not imply fewer FLOPs. + +## Prerequisites + +1. Confirm authorized remote host, free disk, memory, CPU, existing work, and access. +2. Install prebuilt wheels only. Never compile locally or silently use another runner. +3. Freeze the official repository revision and download file manifest including SHA256. +4. Allocate checkpoint size plus 3 GB disk margin. Use an experiment-owned directory, + not a shared Hugging Face cache. Never delete pre-existing weights. + +## Bounded procedure + +1. Run synthetic correctness tests on the remote host: BF16, 3D indexing, directional + replacement error, and non-transitive groups. +2. Download one checkpoint sequentially and verify upstream sizes and LFS SHA256. +3. Scan each tensor in 262,144-element chunks. Record shape, dtype, axis role, + layer, bytes, moments, max magnitude, and finite-value checks. Memory-map weights; + evict scanned pages. Do not search for exactly equal weight tensors. +4. Whole-tensor screening: same role, same shape, same dtype. Sample 2,048 aligned, + evenly spaced elements. Select up to 10 neighbors by each of cosine and relative + error. Fairly distribute the 2,000-pair exact-verification cap across role groups. +5. Block screening: non-overlapping 64x64 and 128x128 2D partitions. For 3D tensors, + use axis-0 slices and bounded volumes adapted to actual dimensions. Keep up to + 256 uniformly reservoir-sampled blocks per role/shape/scale for the 1.7 GiB pilot host + (below the planned maximum of 512). Record edge bytes + and sampling coverage. Verify at most 20,000 block pairs. Never extrapolate sampled + discoveries to unseen blocks or add savings across alternative scales. +6. Exact metrics: full-element cosine, Pearson, directional replacement errors, + symmetric error, optimal-scalar residual. Tail diagnostics use samples and say so. + Controls include random actual comparable pairs, shuffled sampled elements, + and norm-matched Gaussian vectors. Save sampled versus exact error discrepancies. + Bounded mode-unfolding SVD probes are diagnostic only, not a replacement test. +7. Form deterministic star groups at 1%, 2%, 5%, 10%, 20% relative weight error. + Every member requires an exact verified edge to its representative. Unknown + pairs remain independent; never use graph connected components as sharing groups. +8. Compute original bytes minus representatives, 16-byte mapping per analyzed object, + and 64-byte descriptor per shared group. Fall back to unchanged storage if larger. + Count savings against the entire model, preserving unsampled blocks and edge regions. +9. Save raw data, groups, thresholds, coverage, timings, resource usage, figures + (PNG/PDF/SVG), model metadata and report. Validate every group against exact edges. +10. Copy results to the deliverable workspace and verify checksums before deleting + the experiment-owned checkpoint. Save a cleanup record. Do not delete the venv, + result data or any unrelated cache. +11. Review full diff, commit and push a work branch with Codex Bot identity, then + create a PR for manual merge. Never silently initialize/push main or auto-merge. + +## Budget and stop conditions + +- Real 2B analysis computation: hard wall-time limit 7,200 seconds. Reserve final time + for summaries; download and report-generation time are recorded separately. +- Use one BLAS thread to avoid interference with existing services. +- Save exact-pair tables every 100 pairs; stage deadlines prevent whole-tensor work + from consuming the complete block budget. +- On timeout or failure, retain source weights until partial results are recoverable + and report coverage/failure truthfully. Never call an incomplete run complete. +- Empty or negligible low-error sharing is a valid negative result. Do not add scales, + increase error thresholds, align permutations, train or expand to other models automatically. +- No runtime capability/performance claim without separately authorized evaluation. + +## Commands (remote, from project root) + +```sh +OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 venv/bin/python -B -m unittest discover -s tests -v +venv/bin/python -B -m pilot.download --root runs/qwen35-2b +OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 timeout 7200 venv/bin/python -B -m pilot.analyze --root runs/qwen35-2b --seconds 6600 +OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 venv/bin/python -B -m pilot.plot --root runs/qwen35-2b +venv/bin/python -B -m pilot.validate --root runs/qwen35-2b +# Copy archived results, verify checksums at the destination, then: +venv/bin/python -B -m pilot.validate --root runs/qwen35-2b --cleanup-weights +``` + +## Follow-up decisions after the pilot + +- Separate low-byte normalization groups from matrix/expert compression opportunities. +- Decide whether approximate-neighbor recall and block coverage justify another bounded pass. +- If meaningful candidates exist, request a small calibration-forward experiment to measure + output distortion, then held-out loss/tasks, then implementation-specific memory and latency. +- For MoE, analyze router behavior and joint gate/up/down expert consistency before treating + independently similar matrix blocks as interchangeable experts. +- Use these observations to revise this SOP before downloading a second model. diff --git a/pilot/__init__.py b/pilot/__init__.py new file mode 100644 index 0000000..d86dfa6 --- /dev/null +++ b/pilot/__init__.py @@ -0,0 +1 @@ +"""Bounded, inference-free weight similarity pilot.""" diff --git a/pilot/analyze.py b/pilot/analyze.py new file mode 100644 index 0000000..0955e30 --- /dev/null +++ b/pilot/analyze.py @@ -0,0 +1,388 @@ +"""Bounded CPU-only screening, exact verification and conservative sharing groups.""" +import argparse +import collections +import csv +import hashlib +import itertools +import json +import math +import pathlib +import platform +import resource +import time + +import numpy as np + +from .tensors import decode, finish, load_tensors, metrics + +THRESHOLDS = [.01, .02, .05, .10, .20] + + +def write_json(path, obj): + path.write_text(json.dumps(obj, indent=2, allow_nan=False)) + + +def write_csv(path, rows): + if not rows: + path.write_text('') + return + with path.open('w', newline='') as f: + writer = csv.DictWriter(f, fieldnames=list(rows[0])) + writer.writeheader() + writer.writerows(rows) + + +def stable_seed(key, seed): + return int.from_bytes(hashlib.sha256((str(seed) + str(key)).encode()).digest()[:8], 'little') + + +def read_object(obj, tensors): + t = tensors[obj['tensor']] + if obj['kind'] == 'tensor': + return t.read() + slices = tuple(slice(s, s + b) for s, b in zip(obj['start'], obj['shape'])) + return decode(t.raw.reshape(t.shape)[slices], t.dtype).ravel() + + +def sample_object(obj, tensors, sample_size=2048): + t = tensors[obj['tensor']] + n = math.prod(obj['shape']) + indices = np.linspace(0, n - 1, min(n, sample_size), dtype=np.int64) + if obj['kind'] == 'tensor': + return t.read(indices) + coords = np.unravel_index(indices, obj['shape']) + global_coords = tuple(c + s for c, s in zip(coords, obj['start'])) + flat = np.ravel_multi_index(global_coords, t.shape) + return t.read(flat) + + +def exact_pair(a, b, tensors, deadline): + if a['kind'] != 'tensor': + return metrics(read_object(a, tensors), read_object(b, tensors)) + ta, tb = tensors[a['tensor']], tensors[b['tensor']] + xx = yy = xy = delta = sx = sy = 0. + for start in range(0, ta.numel, 262144): + if time.monotonic() > deadline: + raise TimeoutError('Analysis deadline reached') + sl = slice(start, min(start + 262144, ta.numel)) + x, y = ta.read(sl).astype(np.float64), tb.read(sl).astype(np.float64) + xx += float(x @ x) + yy += float(y @ y) + xy += float(x @ y) + d = x - y + delta += float(d @ d) + sx += float(x.sum()) + sy += float(y.sum()) + ta.evict() + tb.evict() + ta.evict() + tb.evict() + return finish(xx, yy, xy, delta, sx, sy, ta.numel) + + +def sharing_groups(objects, pairs, threshold): + """Star groups; every member must have an exact verified edge to its representative.""" + adjacent = collections.defaultdict(dict) + for pair in pairs: + a, b = pair['a'], pair['b'] + if pair['error_member_to_rep'] <= threshold: + adjacent[b][a] = pair['error_member_to_rep'] + if pair['error_rep_to_member'] <= threshold: + adjacent[a][b] = pair['error_rep_to_member'] + unassigned = set(range(len(objects))) + order = sorted(unassigned, key=lambda i: (-len(adjacent[i]), objects[i]['id'])) + groups = [] + for rep in order: + if rep not in unassigned: + continue + members = sorted(set(adjacent[rep]) & unassigned) + members.append(rep) + unassigned.difference_update(members) + groups.append({'representative': rep, 'members': members, + 'max_error': max([adjacent[rep][i] for i in members if i != rep] or [0.])}) + return groups + + +def nearest_candidates(objects, features, cap, neighbors=10): + buckets = collections.defaultdict(list) + for i, obj in enumerate(objects): + buckets[(obj['role'], tuple(obj['shape']), obj['scale'], obj['dtype'])].append(i) + candidates = [] + eligible_pairs = 0 + for key, ids in sorted(buckets.items(), key=lambda item: str(item[0])): + eligible_pairs += len(ids) * (len(ids) - 1) // 2 + if len(ids) < 2: + continue + x = np.stack([features[i] for i in ids]).astype(np.float64) + norm = np.linalg.norm(x, axis=1) + u = x / np.maximum(norm[:, None], 1e-30) + # Cap each bucket's reservoir to bound the screening matrix. + cos = u @ u.T + sq = np.maximum(0., norm[:, None] ** 2 + norm[None, :] ** 2 - 2 * (x @ x.T)) + err = np.sqrt(sq / np.maximum(norm[:, None] ** 2, 1e-30)) + np.fill_diagonal(cos, -np.inf) + np.fill_diagonal(err, np.inf) + seen = set() + for row, i in enumerate(ids): + near = set(np.argsort(err[row])[:neighbors]) | set(np.argsort(-cos[row])[:neighbors]) + for col in near: + j = ids[col] + edge = (min(i, j), max(i, j)) + if i == j or edge in seen: + continue + seen.add(edge) + candidates.append({'a': edge[0], 'b': edge[1], 'screen_error': float(min(err[row, col], err[col, row])), + 'screen_cosine': float(cos[row, col])}) + # Low-error candidates first, then guarantee role/scale diversity round-robin. + queues = collections.defaultdict(list) + for c in sorted(candidates, key=lambda c: (c['screen_error'], c['a'], c['b'])): + o = objects[c['a']] + queues[(o['role'], tuple(o['shape']), o['scale'])].append(c) + queues = [collections.deque(q) for _, q in sorted(queues.items(), key=lambda x: str(x[0]))] + chosen = [] + while queues and len(chosen) < cap: + for q in queues: + if q and len(chosen) < cap: + chosen.append(q.popleft()) + queues = [q for q in queues if q] + return chosen, {'eligible_pairs_in_screened_objects': eligible_pairs, + 'candidate_pairs_before_cap': len(candidates), 'selected_pairs': len(chosen)} + + +def object_record(t, kind='tensor', shape=None, start=None, scale='whole'): + shape = tuple(shape or t.shape) + return {'id': t.name if kind == 'tensor' else f'{t.name}|{scale}|{start}', + 'tensor': t.name, 'role': t.role, 'layer': t.layer, 'kind': kind, 'dtype': t.dtype, + 'shape': shape, 'start': start, 'scale': scale, + 'bytes': math.prod(shape) * (t.bytes // t.numel)} + + +def build_blocks(tensors, seed, deadline, reservoir_limit=256): + reservoirs, counts, rngs = {}, collections.Counter(), {} + totals = collections.defaultdict(lambda: {'blocks': 0, 'bytes': 0, 'edge_bytes': 0}) + for t in tensors.values(): + if len(t.shape) == 2: + specs = [(f'2d-{s}', (s, s)) for s in (64, 128) if min(t.shape) >= s] + elif len(t.shape) == 3: + specs = [('3d-axis0-slice', (1, min(64, t.shape[1]), min(64, t.shape[2]))), + ('3d-volume', (min(4, t.shape[0]), min(32, t.shape[1]), min(32, t.shape[2])))] + else: + continue + for scale, shape in specs: + nblocks = math.prod(d // b for d, b in zip(t.shape, shape)) + totals[scale]['blocks'] += nblocks + block_bytes = math.prod(shape) * (t.bytes // t.numel) + totals[scale]['bytes'] += nblocks * block_bytes + totals[scale]['edge_bytes'] += t.bytes - nblocks * block_bytes + for grid in itertools.product(*(range(d // b) for d, b in zip(t.shape, shape))): + if time.monotonic() > deadline: + raise TimeoutError('Deadline during block enumeration') + start = tuple(i * b for i, b in zip(grid, shape)) + obj = object_record(t, 'block', shape, start, scale) + key = (t.role, shape, scale) + if key not in reservoirs: + reservoirs[key], rngs[key] = [], np.random.default_rng(stable_seed(key, seed)) + counts[key] += 1 + reservoir = reservoirs[key] + if len(reservoir) < reservoir_limit: + reservoir.append(obj) + else: + j = int(rngs[key].integers(counts[key])) + if j < reservoir_limit: + reservoir[j] = obj + objects = [obj for key in sorted(reservoirs, key=str) for obj in reservoirs[key]] + coverage = [{'role': key[0], 'shape': str(key[1]), 'scale': key[2], + 'total_blocks': counts[key], 'sampled_blocks': len(reservoirs[key])} + for key in sorted(reservoirs, key=str)] + return objects, coverage, dict(totals) + + +def structural_probe(a, b, tensors): + """Bounded mode unfoldings of aligned subvolumes; exploratory, never a grouping criterion.""" + def small(obj): + t = tensors[obj['tensor']] + shape = obj['shape'] + starts = obj['start'] or [0] * len(shape) + slices = tuple(slice(s, s + min(d, 32)) for s, d in zip(starts, shape)) + return decode(t.raw.reshape(t.shape)[slices], t.dtype) + x, y = small(a), small(b) + result = [] + if x.ndim < 2: + return result + for axis in range(x.ndim): + mx = np.moveaxis(x, axis, 0).reshape(x.shape[axis], -1) + my = np.moveaxis(y, axis, 0).reshape(y.shape[axis], -1) + if min(mx.shape) < 2: + result.append({'axis': axis, 'probe_shape': list(x.shape), 'trivial_mode_skipped': True}) + continue + ux, sx, _ = np.linalg.svd(mx, full_matrices=False) + uy, sy, _ = np.linalg.svd(my, full_matrices=False) + k = min(4, max(1, min(len(sx), len(sy)) // 2)) + result.append({'axis': axis, 'probe_shape': list(x.shape), + 'k': k, + 'normalized_spectrum_distance': float(np.linalg.norm(sx / max(np.linalg.norm(sx), 1e-30) - sy / max(np.linalg.norm(sy), 1e-30))), + 'left_subspace_overlap': float(np.linalg.norm(ux[:, :k].T @ uy[:, :k]) ** 2 / k)}) + return result + + +def analyze_mode(mode, objects, tensors, root, deadline, cap, seed, baseline_cap=100): + print(f'{mode}: sampling {len(objects)} objects', flush=True) + features = [sample_object(o, tensors) for o in objects] + candidates, coverage = nearest_candidates(objects, features, cap) + write_json(root / f'{mode}_objects.json', objects) + pairs, nulls, structures = [], [], [] + rng = np.random.default_rng(seed) + random_pairs = [] + comparable = collections.defaultdict(list) + for j, obj in enumerate(objects): + comparable[(obj['role'], tuple(obj['shape']), obj['scale'], obj['dtype'])].append(j) + for key, ids in sorted(comparable.items(), key=lambda item: str(item[0])): + if len(ids) < 2: + continue + for _ in range(3): + a, b = rng.choice(ids, size=2, replace=False).tolist() + random_pairs.append({'a': a, 'b': b, 'role': key[0], 'sampled': True, + **metrics(features[a], features[b])}) + write_json(root / f'{mode}_random_pairs.json', random_pairs) + timeout = False + structural_counts = collections.Counter() + for i, candidate in enumerate(candidates): + if time.monotonic() > deadline: + timeout = True + break + a, b = candidate['a'], candidate['b'] + try: + measured = exact_pair(objects[a], objects[b], tensors, deadline) + except TimeoutError: + timeout = True + break + row = {**candidate, **measured, + 'a_id': objects[a]['id'], 'b_id': objects[b]['id'], + 'role': objects[a]['role'], 'scale_name': objects[a]['scale'], + 'layer_distance': abs(objects[a]['layer'] - objects[b]['layer'])} + row['screen_vs_exact_min_error_absolute_difference'] = abs(candidate['screen_error'] - min(measured['error_member_to_rep'], measured['error_rep_to_member'])) + # Tail diagnostics are sampled, explicitly labelled, not claimed exact. + fx, fy = features[a].astype(np.float64), features[b].astype(np.float64) + row['sampled_error_q99_over_member_rms'] = float(np.quantile(np.abs(fx - fy), .99) / max(np.sqrt(np.mean(fx * fx)), 1e-30)) + pairs.append(row) + if i < baseline_cap: + for method, y in [('shuffled', rng.permutation(fy)), ('gaussian_norm_matched', rng.normal(size=fy.size))]: + if method == 'gaussian_norm_matched': + y *= np.linalg.norm(fy) / max(np.linalg.norm(y), 1e-30) + nulls.append({'a': a, 'b': b, 'method': method, 'sampled': True, **metrics(fx, y)}) + ndim = len(objects[a]['shape']) + if ndim >= 2 and structural_counts[ndim] < 20: + structures.append({'a': a, 'b': b, 'modes': structural_probe(objects[a], objects[b], tensors)}) + structural_counts[ndim] += 1 + if (i + 1) % 100 == 0: + write_csv(root / f'{mode}_pairs.csv', pairs) + print(f'{mode}: verified {i + 1}/{len(candidates)}', flush=True) + write_csv(root / f'{mode}_pairs.csv', pairs) + write_json(root / f'{mode}_nulls.json', nulls) + write_json(root / f'{mode}_structure.json', structures) + coverage.update({'objects_screened': len(objects), 'pairs_verified': len(pairs), 'timed_out': timeout, + 'screen_sample_elements': 2048, 'candidate_neighbors_per_metric': 10}) + write_json(root / f'{mode}_coverage.json', coverage) + return pairs + + +def summarize(mode, objects, pairs, model_bytes, root, partition_totals=None): + rows = [] + scales = sorted({o['scale'] for o in objects}) + for scale in scales: + ids = [i for i, o in enumerate(objects) if o['scale'] == scale] + selected = set(ids) + for threshold in THRESHOLDS: + groups = sharing_groups(objects, [p for p in pairs if p['a'] in selected], threshold) + groups = [g for g in groups if g['representative'] in selected] + original = sum(objects[i]['bytes'] for i in ids) + rep_bytes = sum(objects[g['representative']]['bytes'] for g in groups) + # Concrete theoretical layout: 16 bytes per object mapping, plus 64 bytes per shared group. + # Identity case falls back to original checkpoint at zero additional cost. + mapping_bytes = 16 * len(ids) + 64 * sum(len(g['members']) > 1 for g in groups) + proposed = rep_bytes + mapping_bytes + saved = max(0, original - proposed) + partition_count = partition_totals[scale]['blocks'] if partition_totals else len(ids) + rows.append({'mode': mode, 'scale': scale, 'threshold': threshold, + 'objects': len(ids), 'groups': len(groups), + 'total_partition_objects': partition_count, + 'unexamined_partition_objects': partition_count - len(ids), + 'conservative_partition_groups': partition_count - len(ids) + len(groups), + 'shared_groups': sum(len(g['members']) > 1 for g in groups), + 'max_group_size': max([len(g['members']) for g in groups] or [0]), + 'verified_shared_bytes': sum(sum(objects[i]['bytes'] for i in g['members']) for g in groups if len(g['members']) > 1), + 'examined_object_bytes': original, 'representative_bytes': rep_bytes, + 'mapping_bytes_if_encoded': mapping_bytes, 'net_saved_bytes': saved, + 'net_saved_model_fraction': saved / model_bytes, + 'max_replacement_error': max([g['max_error'] for g in groups] or [0.])}) + write_json(root / f'groups_{mode}_{scale}_{threshold:.2f}.json', groups) + return rows + + +def main(): + p = argparse.ArgumentParser() + p.add_argument('--root', type=pathlib.Path, required=True) + p.add_argument('--seconds', type=int, default=7200) + p.add_argument('--seed', type=int, default=20261109) + args = p.parse_args() + started = time.monotonic() + deadline = started + args.seconds + root = args.root + output = root / 'analysis' + output.mkdir(exist_ok=True) + tensors = load_tensors(root / 'weights') + inventory = [] + for t in tensors.values(): + if time.monotonic() > deadline: + raise TimeoutError('Deadline during inventory') + n = 0 + total = squared = 0. + max_abs = 0. + for x in t.chunked(): + if not np.isfinite(x).all(): + raise ValueError(f'Nonfinite weights: {t.name}') + z = x.astype(np.float64) + n += z.size + total += float(z.sum()) + squared += float(z @ z) + max_abs = max(max_abs, float(np.abs(z).max())) + t.evict() + inventory.append({'name': t.name, 'shape': str(t.shape), 'ndim': len(t.shape), 'dtype': t.dtype, + 'numel': t.numel, 'bytes': t.bytes, 'role': t.role, 'layer': t.layer, + 'mean': total / n, 'rms': math.sqrt(squared / n), 'max_abs': max_abs}) + t.evict() + write_csv(output / 'inventory.csv', inventory) + print(f'Inventory: {len(tensors)} tensors, {sum(t.bytes for t in tensors.values())} bytes', flush=True) + whole = [object_record(t) for t in tensors.values()] + # Stage budgets prevent the whole-tensor stage from consuming block-analysis time. + whole_deadline = min(deadline, started + args.seconds * .50) + whole_pairs = analyze_mode('tensor', whole, tensors, output, whole_deadline, 2000, args.seed) + blocks, block_coverage, block_totals = build_blocks(tensors, args.seed, deadline) + write_csv(output / 'block_sampling.csv', block_coverage) + write_json(output / 'block_partitions.json', block_totals) + block_pairs = analyze_mode('block', blocks, tensors, output, deadline, 20000, args.seed) + model_bytes = sum(t.bytes for t in tensors.values()) + rows = summarize('tensor', whole, whole_pairs, model_bytes, output) + rows += summarize('block', blocks, block_pairs, model_bytes, output, block_totals) + write_csv(output / 'compression_curve.csv', rows) + summary = {'model': json.loads((root / 'manifest.json').read_text())['repo'], + 'revision': json.loads((root / 'manifest.json').read_text())['revision'], + 'tensor_count': len(tensors), 'parameter_count': sum(t.numel for t in tensors.values()), + 'model_weight_bytes': model_bytes, 'ndim_counts': dict(collections.Counter(len(t.shape) for t in tensors.values())), + 'elapsed_seconds': time.monotonic() - started, 'analysis_budget_seconds': args.seconds, + 'peak_rss_kib': resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, + 'python': platform.python_version(), 'numpy': np.__version__, 'seed': args.seed, + 'limitations': ['No inference or functional equivalence measured', 'No exhaustive pair search', + 'Block search uses per-role/shape/scale reservoirs of 256', + 'Different block scales overlap; savings must not be added', + 'Block savings count only verified disjoint sampled blocks, no extrapolation', + 'Spectral probes use aligned bounded subvolumes, not whole tensors', + 'Shared encoding is theoretical, not a runnable compressed checkpoint', + 'No cross-role grouping, neuron permutation alignment, or training'], + 'completed': True} + write_json(output / 'summary.json', summary) + print(json.dumps(summary), flush=True) + + +if __name__ == '__main__': + main() diff --git a/pilot/download.py b/pilot/download.py new file mode 100644 index 0000000..2e903b0 --- /dev/null +++ b/pilot/download.py @@ -0,0 +1,140 @@ +"""Download pinned official safetensors, no framework or shared cache.""" +import argparse +import concurrent.futures +import hashlib +import json +import pathlib +import os +import shutil +import time +import threading +import urllib.parse +import urllib.request + + +def fetch_json(url): + with urllib.request.urlopen(url, timeout=90) as response: + return json.load(response) + + +def ranged_download(url, part, total, workers=8): + """Resume a sequential prefix; journal completed byte ranges before final hashing.""" + journal = part.with_suffix(part.suffix + '.ranges.json') + if journal.exists(): + state = json.loads(journal.read_text()) + if state['total'] != total: + raise RuntimeError('Range journal size mismatch') + else: + state = {'total': total, 'prefix': part.stat().st_size if part.exists() else 0, 'done': []} + journal.write_text(json.dumps(state)) + lock = threading.Lock() + fd = os.open(part, os.O_CREAT | os.O_RDWR, 0o600) + chunk_size = 64 * 1024 * 1024 + ranges = [(start, min(start + chunk_size, total) - 1) + for start in range(state['prefix'], total, chunk_size)] + + def one(bounds): + start, end = bounds + if start in state['done']: + return + for attempt in range(3): + try: + req = urllib.request.Request(url, headers={'Range': f'bytes={start}-{end}'}) + with urllib.request.urlopen(req, timeout=180) as response: + expected = f'bytes {start}-{end}/{total}' + if response.status != 206 or response.headers.get('Content-Range') != expected: + raise RuntimeError('Server did not honor exact requested range') + offset = start + while chunk := response.read(min(4 * 1024 * 1024, end - offset + 1)): + written = os.pwrite(fd, chunk, offset) + if written != len(chunk): + raise RuntimeError('Short positional write') + offset += written + if offset > end: + break + if offset != end + 1: + raise RuntimeError('Incomplete range') + with lock: + os.fsync(fd) + state['done'].append(start) + tmp = journal.with_suffix('.tmp') + tmp.write_text(json.dumps(state)) + tmp.replace(journal) + done = state['prefix'] + sum(b - a + 1 for a, b in ranges if a in state['done']) + print(f'Weight download {done}/{total} bytes ({100 * done / total:.1f}%)', flush=True) + return + except Exception: + if attempt == 2: + raise + time.sleep(2 ** attempt) + try: + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool: + list(pool.map(one, ranges)) + finally: + os.close(fd) + journal.unlink() + + +def main(): + p = argparse.ArgumentParser() + p.add_argument('--root', type=pathlib.Path, required=True) + p.add_argument('--repo', default='Qwen/Qwen3.5-2B') + args = p.parse_args() + root = args.root.resolve() + root.mkdir(parents=True, exist_ok=True) + weights = root / 'weights' + weights.mkdir(exist_ok=True) + marker = weights / '.pilot-owned' + marker.write_text(args.repo) + old_manifest = root / 'manifest.json' + pinned = json.loads(old_manifest.read_text())['revision'] if old_manifest.exists() else None + endpoint = f'https://huggingface.co/api/models/{args.repo}' + if pinned: + endpoint += f'/revision/{pinned}' + info = fetch_json(endpoint + '?blobs=true') + revision = info['sha'] + files = [f for f in info['siblings'] if f['rfilename'].endswith('.safetensors') + or f['rfilename'] in ('config.json', 'model.safetensors.index.json', 'README.md')] + total = sum(f.get('size', 0) for f in files) + if shutil.disk_usage(root).free < total + 3_000_000_000: + raise RuntimeError('Insufficient space for checkpoint plus 3 GB margin') + manifest = {'repo': args.repo, 'revision': revision, 'expected_bytes': total, + 'started_unix': time.time(), 'files': [], 'license': info.get('cardData', {}).get('license')} + (root / 'source_metadata.json').write_text(json.dumps(info, indent=2)) + for entry in files: + name = entry['rfilename'] + if pathlib.PurePosixPath(name).name != name: + raise RuntimeError(f'Unexpected nested path: {name}') + dest = weights / name + part = dest.with_suffix(dest.suffix + '.partial') + digest = hashlib.sha256() + url = f'https://huggingface.co/{args.repo}/resolve/{revision}/{urllib.parse.quote(name)}?download=true' + print(f'Downloading {name}, expected {entry.get("size")} bytes', flush=True) + if entry.get('size', 0) > 128 * 1024 * 1024: + ranged_download(url, part, entry['size']) + with part.open('rb') as stream: + while chunk := stream.read(4 * 1024 * 1024): + digest.update(chunk) + else: + with urllib.request.urlopen(url, timeout=180) as response, part.open('wb') as out: + while chunk := response.read(4 * 1024 * 1024): + out.write(chunk) + digest.update(chunk) + sha = digest.hexdigest() + if entry.get('size') and part.stat().st_size != entry['size']: + raise RuntimeError(f'Size mismatch: {name}') + expected_sha = entry.get('lfs', {}).get('sha256') + if expected_sha and sha != expected_sha: + raise RuntimeError(f'SHA256 mismatch: {name}') + part.rename(dest) + manifest['files'].append({'name': name, 'bytes': dest.stat().st_size, + 'sha256': sha, 'upstream_sha256': expected_sha}) + (root / 'manifest.json').write_text(json.dumps(manifest, indent=2)) + manifest['completed_unix'] = time.time() + (root / 'manifest.json').write_text(json.dumps(manifest, indent=2)) + shutil.copy2(weights / 'config.json', root / 'model_config.json') + print(json.dumps({'revision': revision, 'bytes': sum(f['bytes'] for f in manifest['files'])}), flush=True) + + +if __name__ == '__main__': + main() diff --git a/pilot/plot.py b/pilot/plot.py new file mode 100644 index 0000000..bcb85b8 --- /dev/null +++ b/pilot/plot.py @@ -0,0 +1,156 @@ +"""Scientific figures and a compact result report from saved data only.""" +import argparse +import csv +import json +import pathlib + +import matplotlib +matplotlib.use('Agg') +import matplotlib.pyplot as plt +import numpy as np + + +def read_csv(path): + with path.open() as f: + return list(csv.DictReader(f)) + + +def main(): + p = argparse.ArgumentParser() + p.add_argument('--root', type=pathlib.Path, required=True) + args = p.parse_args() + root = args.root + out = root / 'analysis' + figures = root / 'figures' + figures.mkdir(exist_ok=True) + plt.rcParams.update({'font.size': 10, 'axes.spines.top': False, 'axes.spines.right': False, + 'pdf.fonttype': 42, 'svg.fonttype': 'none', 'figure.dpi': 120}) + def save(fig, name): + fig.tight_layout() + for ext in ('png', 'pdf', 'svg'): + fig.savefig(figures / f'{name}.{ext}', dpi=200, bbox_inches='tight') + plt.close(fig) + + curves = read_csv(out / 'compression_curve.csv') + summary = json.loads((out / 'summary.json').read_text()) + fig, axes = plt.subplots(1, 2, figsize=(11, 4)) + for scale in sorted({r['scale'] for r in curves}): + rows = [r for r in curves if r['scale'] == scale] + x = [100 * float(r['threshold']) for r in rows] + axes[0].plot(x, [100 * float(r['net_saved_model_fraction']) for r in rows], 'o-', label=scale) + axes[1].plot(x, [int(r['shared_groups']) for r in rows], 'o-', label=scale) + axes[0].set(xlabel='Maximum relative weight error (%)', ylabel='Verified net savings / model bytes (%)') + axes[1].set(xlabel='Maximum relative weight error (%)', ylabel='Non-singleton sharing groups') + axes[0].legend(fontsize=8) + fig.suptitle('Conservative verified sharing; block scales are separate alternatives') + save(fig, 'compression_frontier') + + pair_sets = {mode: read_csv(out / f'{mode}_pairs.csv') for mode in ('tensor', 'block')} + fig, axes = plt.subplots(2, 2, figsize=(11, 7)) + for row, (mode, pairs) in enumerate(pair_sets.items()): + for scale in sorted({r['scale_name'] for r in pairs}): + group = [r for r in pairs if r['scale_name'] == scale] + axes[row, 0].hist([float(r['cosine']) for r in group if r['cosine']], bins=np.linspace(-1, 1, 46), histtype='step', label=scale) + errors = [float(r['error_member_to_rep']) for r in group] + axes[row, 1].hist(errors, bins=np.linspace(0, max(1e-6, max(errors, default=0.) * 1.01), 46), histtype='step', label=scale) + axes[row, 0].set(title=f'{mode}: exact cosine of screened candidates', xlabel='Cosine', ylabel='Candidate pairs') + axes[row, 1].set(title=f'{mode}: exact direct replacement error', xlabel='Relative Frobenius error', ylabel='Candidate pairs') + axes[row, 0].legend(fontsize=7) + save(fig, 'candidate_distributions') + + pairs = pair_sets['tensor'] + roles = sorted({r['role'] for r in pairs}) + # Use a large 2D repeated projection, not tiny norms, for the primary heatmap. + objects = json.loads((out / 'tensor_objects.json').read_text()) + candidates = [role for role in roles if any(o['role'] == role and len(o['shape']) == 2 for o in objects)] + role = max(candidates or roles, key=lambda k: sum(r['role'] == k for r in pairs)) + chosen = [(i, o) for i, o in enumerate(objects) if o['role'] == role] + chosen.sort(key=lambda item: item[1]['layer']) + indices = {i: j for j, (i, _) in enumerate(chosen)} + matrix = np.full((len(chosen), len(chosen)), np.nan) + np.fill_diagonal(matrix, 1.) + for r in pairs: + a, b = int(r['a']), int(r['b']) + if a in indices and b in indices and r['cosine']: + matrix[indices[a], indices[b]] = matrix[indices[b], indices[a]] = float(r['cosine']) + fig, ax = plt.subplots(figsize=(7, 6)) + cmap = plt.get_cmap('coolwarm').copy() + cmap.set_bad('#dddddd') + im = ax.imshow(matrix, cmap=cmap, vmin=-1, vmax=1) + labels = [o['layer'] for _, o in chosen] + ticks = np.arange(0, len(labels), max(1, len(labels) // 10)) + ax.set_xticks(ticks, [labels[i] for i in ticks]) + ax.set_yticks(ticks, [labels[i] for i in ticks]) + ax.set(xlabel='Layer', ylabel='Layer', title=role.replace('model.', '') + '\nGray = unverified; diagonal is identity') + fig.colorbar(im, ax=ax, label='Exact cosine') + save(fig, 'layer_heatmap') + + fig, ax = plt.subplots(figsize=(8, 4)) + for mode in ('tensor', 'block'): + rows = pair_sets[mode] + distances = sorted({int(r['layer_distance']) for r in rows}) + median = [np.median([float(r['cosine']) for r in rows if int(r['layer_distance']) == d and r['cosine']]) for d in distances] + ax.plot(distances, median, 'o-', label=mode) + ax.set(xlabel='Absolute layer distance', ylabel='Median exact cosine', title='Screened candidates; role mixture and selection bias apply') + ax.legend() + save(fig, 'layer_distance') + + fig, axes = plt.subplots(1, 2, figsize=(11, 4)) + for ax, mode in zip(axes, ('tensor', 'block')): + nulls = json.loads((out / f'{mode}_nulls.json').read_text()) + for method in ('shuffled', 'gaussian_norm_matched'): + values = [r['cosine'] for r in nulls if r['method'] == method and r['cosine'] is not None] + ax.hist(values, bins=np.linspace(-1, 1, 26), histtype='step', label=method) + real_random = json.loads((out / f'{mode}_random_pairs.json').read_text()) + ax.hist([r['cosine'] for r in real_random if r['cosine'] is not None], bins=np.linspace(-1, 1, 26), histtype='step', label='random actual same-role pairs') + ax.set(title=f'{mode}: sampled null controls', xlabel='Sampled cosine', ylabel='Pairs') + ax.legend(fontsize=8) + save(fig, 'null_controls') + + fig, axes = plt.subplots(1, 2, figsize=(11, 4)) + structure = json.loads((out / 'block_structure.json').read_text()) + for ax, ndim in zip(axes, (2, 3)): + probes = [m for pair in structure for m in pair['modes'] if len(m['probe_shape']) == ndim and not m.get('trivial_mode_skipped')] + for axis in sorted({m['axis'] for m in probes}): + selected = [m for m in probes if m['axis'] == axis] + ax.scatter([m['normalized_spectrum_distance'] for m in selected], + [m['left_subspace_overlap'] for m in selected], s=18, alpha=.65, label=f'mode {axis}') + ax.set(title=f'{ndim}D blocks: bounded mode-unfolding probes', xlabel='Normalized spectrum distance', ylabel='Leading left subspace overlap (k <= 4)') + if probes: + ax.legend(fontsize=8) + else: + ax.text(.5, .5, 'No eligible probes', transform=ax.transAxes, ha='center') + save(fig, 'block_structure') + + parts = json.loads((out / 'block_partitions.json').read_text()) + lines = ['# Qwen3.5-2B parameter similarity pilot', '', + f"- Revision: `{summary['revision']}`", f"- Stored parameters: {summary['parameter_count']:,}", + f"- Weight bytes: {summary['model_weight_bytes']:,}", + f"- Tensor dimensionalities: {summary['ndim_counts']}", + f"- Analysis seconds: {summary['elapsed_seconds']:.1f} / {summary['analysis_budget_seconds']}", + f"- Peak analysis RSS: {summary['peak_rss_kib'] / 1024:.1f} MiB", '', + '## Verified sharing estimates', '', + '| Representation | Weight-error limit | Verified sharing groups | Conservative total groups* | Largest group | Net model savings |', + '|---|---:|---:|---:|---:|---:|---:|'] + for r in curves: + lines.append(f"| {r['scale']} | {float(r['threshold']):.0%} | {r['shared_groups']} | {r['conservative_partition_groups']} | {r['max_group_size']} | {float(r['net_saved_model_fraction']):.6%} |") + lines += ['', '*Conservative total groups counts unexamined full blocks as singletons; boundary regions and ineligible tensors stay unchanged and are outside that block count.', + 'Savings count only verified replacement edges. Unsampled blocks remain unchanged. Different block scales must not be summed.', + 'Theoretical encoding: one stored representative per group, 16 bytes per object mapping and 64 bytes per shared-group descriptor; identity fallback if larger.', + 'Exact grouping uses direct replacement only. Scaled and spectral results are diagnostic.', '', + '## Coverage', ''] + for mode in ('tensor', 'block'): + coverage = json.loads((out / f'{mode}_coverage.json').read_text()) + lines.append(f'- {mode}: `{json.dumps(coverage)}`') + lines += [f'- Complete-block partition totals (per alternative): `{json.dumps(parts)}`', '', + '## Interpretation limits', '', *['- ' + s for s in summary['limitations']], + '- Candidate histograms are selection-biased and do not represent the distribution of all tensor pairs.', + '- 1D normalization weights may cluster strongly but carry few bytes; do not infer material compression from group counts.', + '- No task accuracy, perplexity, inference speed or runtime-memory improvement has been measured.', + '- Stored parameter counts may omit tied aliases; no duplicate-weight equality search is performed.', '', + '## Figures', '', *[f'![{name}](figures/{name}.png)' for name in ['compression_frontier', 'candidate_distributions', 'layer_heatmap', 'layer_distance', 'null_controls', 'block_structure']]] + (root / 'REPORT.md').write_text('\n'.join(lines) + '\n') + + +if __name__ == '__main__': + main() diff --git a/pilot/run.py b/pilot/run.py new file mode 100644 index 0000000..b3c0879 --- /dev/null +++ b/pilot/run.py @@ -0,0 +1,58 @@ +"""Supervise the one-model job; no automatic download of subsequent models.""" +import argparse +import json +import os +import pathlib +import subprocess +import sys +import time + + +def main(): + p = argparse.ArgumentParser() + p.add_argument('--root', type=pathlib.Path, required=True) + args = p.parse_args() + root = args.root.resolve() + root.mkdir(exist_ok=True, parents=True) + status_file = root / 'status.json' + start = time.time() + env = dict(os.environ, OPENBLAS_NUM_THREADS='1', OMP_NUM_THREADS='1') + + def status(phase, **extra): + status_file.write_text(json.dumps({'phase': phase, 'updated_unix': time.time(), + 'supervisor_started_unix': start, **extra}, indent=2)) + print(phase, flush=True) + + def command(module, log, timeout, extra=()): + with (root / log).open('w') as stream: + subprocess.run([sys.executable, '-B', '-m', module, '--root', str(root), *extra], + check=True, timeout=timeout, env=env, stdout=stream, stderr=subprocess.STDOUT) + + try: + status('waiting_for_download') + download_wait_start = time.monotonic() + while True: + manifest_path = root / 'manifest.json' + if manifest_path.exists(): + manifest = json.loads(manifest_path.read_text()) + if manifest.get('completed_unix') and all((root / 'weights' / f['name']).exists() for f in manifest['files']): + break + if time.monotonic() - download_wait_start > 7200: + raise TimeoutError('Download wait exceeded 2 hours; no analysis started') + time.sleep(10) + status('analyzing', analysis_hard_limit_seconds=7200) + command('pilot.analyze', 'analysis.log', 7200, ('--seconds', '6600')) + status('plotting') + command('pilot.plot', 'plot.log', 300) + status('validating') + command('pilot.validate', 'validate.log', 120) + with (root / 'environment.txt').open('w') as f: + subprocess.run([sys.executable, '-B', '-m', 'pip', 'freeze'], check=True, stdout=f, env=env) + status('results_ready_weights_retained_until_archive_copy') + except Exception as e: + status('failed', error_type=type(e).__name__, error=str(e)) + raise + + +if __name__ == '__main__': + main() diff --git a/pilot/tensors.py b/pilot/tensors.py new file mode 100644 index 0000000..5cfd38a --- /dev/null +++ b/pilot/tensors.py @@ -0,0 +1,82 @@ +"""Read safetensors BF16 directly; never instantiate a model or use pickle.""" +import json +import math +import mmap +import re +import struct + +import numpy as np + +DTYPES = {'BF16': (' 100_000_000: + raise ValueError('Oversized safetensors header') + header = json.loads(f.read(size)) + for name, meta in header.items(): + if name == '__metadata__': + continue + if name in tensors: + raise ValueError(f'Duplicate tensor name: {name}') + tensors[name] = Tensor(file, 8 + size, name, meta) + if not tensors: + raise ValueError('No safetensors found') + return tensors + + +def metrics(x, y): + """x is member; y is representative. Float64 accumulation.""" + x, y = x.astype(np.float64), y.astype(np.float64) + xx, yy, xy = float(x @ x), float(y @ y), float(x @ y) + delta = float((x - y) @ (x - y)) + return finish(xx, yy, xy, delta, float(x.sum()), float(y.sum()), x.size) + + +def finish(xx, yy, xy, delta, sx, sy, n): + tiny = 1e-30 + vx, vy = max(0., xx - sx * sx / n), max(0., yy - sy * sy / n) + return {'cosine': xy / np.sqrt(xx * yy) if xx * yy > tiny else None, + 'pearson': (xy - sx * sy / n) / np.sqrt(vx * vy) if vx * vy > tiny else None, + 'error_member_to_rep': np.sqrt(delta / max(xx, tiny)), + 'error_rep_to_member': np.sqrt(delta / max(yy, tiny)), + 'symmetric_error': np.sqrt(2 * delta / max(xx + yy, tiny)), + 'scaled_error': np.sqrt(max(0., xx - xy * xy / max(yy, tiny)) / max(xx, tiny)), + 'scale': xy / max(yy, tiny)} diff --git a/pilot/validate.py b/pilot/validate.py new file mode 100644 index 0000000..895c953 --- /dev/null +++ b/pilot/validate.py @@ -0,0 +1,85 @@ +"""Validate archived analysis before allowing deletion of downloaded weights.""" +import argparse +import csv +import hashlib +import json +import pathlib +import shutil +import time + + +def validate(root): + analysis = root / 'analysis' + summary = json.loads((analysis / 'summary.json').read_text()) + assert summary['completed'] + manifest = json.loads((root / 'manifest.json').read_text()) + assert summary['revision'] == manifest['revision'] + for mode in ('tensor', 'block'): + objects = json.loads((analysis / f'{mode}_objects.json').read_text()) + with (analysis / f'{mode}_pairs.csv').open() as f: + pairs = list(csv.DictReader(f)) + coverage = json.loads((analysis / f'{mode}_coverage.json').read_text()) + assert len(pairs) == coverage['pairs_verified'] + edges = {} + for p in pairs: + a, b = int(p['a']), int(p['b']) + assert objects[a]['shape'] == objects[b]['shape'] + assert objects[a]['role'] == objects[b]['role'] + edges[(b, a)] = float(p['error_member_to_rep']) + edges[(a, b)] = float(p['error_rep_to_member']) + for file in analysis.glob(f'groups_{mode}_*.json'): + threshold = float(file.stem.rsplit('_', 1)[1]) + groups = json.loads(file.read_text()) + members = [i for g in groups for i in g['members']] + assert len(set(members)) == len(members), file + for g in groups: + assert g['representative'] in g['members'] + for member in g['members']: + if member != g['representative']: + assert edges[(g['representative'], member)] <= threshold + 1e-12 + assert g['max_error'] <= threshold + 1e-12 + assert (root / 'REPORT.md').is_file() + images = list((root / 'figures').glob('*.png')) + assert len(images) >= 5 + for file in images: + assert file.read_bytes()[:8] == b'\x89PNG\r\n\x1a\n' + checksums = {} + for file in sorted(root.rglob('*')): + if (file.is_file() and 'weights' not in file.relative_to(root).parts + and file.name not in ('checksums.json', 'validation.json', 'cleanup.json', 'status.json') + and not file.name.endswith('.log')): + checksums[str(file.relative_to(root))] = hashlib.sha256(file.read_bytes()).hexdigest() + (root / 'checksums.json').write_text(json.dumps(checksums, indent=2)) + result = {'validated_unix': time.time(), 'files_checksummed': len(checksums), + 'group_edges_checked': True, 'passed': True} + (root / 'validation.json').write_text(json.dumps(result, indent=2)) + return result + + +def main(): + p = argparse.ArgumentParser() + p.add_argument('--root', required=True, type=pathlib.Path) + p.add_argument('--cleanup-weights', action='store_true') + args = p.parse_args() + root = args.root.resolve() + result = validate(root) + if args.cleanup_weights: + weights = root / 'weights' + manifest = json.loads((root / 'manifest.json').read_text()) + assert not weights.is_symlink() + assert (weights / '.pilot-owned').read_text() == manifest['repo'] + expected_names = {entry['name'] for entry in manifest['files']} | {'.pilot-owned'} + assert {file.name for file in weights.iterdir()} == expected_names + assert all(file.is_file() and not file.is_symlink() for file in weights.iterdir()) + deleted_bytes = sum(file.stat().st_size for file in weights.iterdir()) + shutil.rmtree(weights) + (root / 'cleanup.json').write_text(json.dumps({'deleted_unix': time.time(), + 'deleted_bytes': deleted_bytes, + 'deleted_directory': 'weights', + 'reason': 'User-authorized delete-after-analysis; archived results validated'}, indent=2)) + result['weights_removed'] = not weights.exists() + print(json.dumps(result)) + + +if __name__ == '__main__': + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b2b9779 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +numpy==2.5.4 +matplotlib==3.11.2 diff --git a/tests/test_pilot.py b/tests/test_pilot.py new file mode 100644 index 0000000..95ddf3d --- /dev/null +++ b/tests/test_pilot.py @@ -0,0 +1,113 @@ +import json +import io +import pathlib +import struct +import tempfile +import unittest +from unittest.mock import patch + +import numpy as np + +from pilot.analyze import (build_blocks, exact_pair, object_record, read_object, + sample_object, sharing_groups) +from pilot.tensors import decode, load_tensors, metrics + + +class PilotTests(unittest.TestCase): + def test_directional_replacement_error_and_scaling(self): + x = np.array([1., 2., 3.]) + m = metrics(x, 2 * x) + self.assertAlmostEqual(m['cosine'], 1.) + self.assertAlmostEqual(m['error_member_to_rep'], 1.) + self.assertAlmostEqual(m['error_rep_to_member'], .5) + self.assertAlmostEqual(m['scaled_error'], 0.) + self.assertAlmostEqual(m['scale'], .5) + + def test_bfloat16(self): + raw = np.array([0x3f80, 0xc000, 0x0000], dtype='