LAB 05 · 像素 patch 词典：拟合与重建

# 当前项目中的独立材料：已存在则复用，缺失才从本讲原站准备。
from pathlib import Path
import io, zipfile, urllib.request
_base = Path.cwd()
_folder = 'lab10-patch'
_sentinel = 'assets/target.npz'
_choices = [_base/_folder, _base/'lecture03-labs'/_folder, _base]
_ready = next((p for p in _choices if (p/_sentinel).is_file()), None)
if _ready is None:
    _url = 'https://codingai-lec03.pages.dev/course/materials/' + _folder + '.zip'
    _payload = urllib.request.urlopen(_url, timeout=60).read()
    with zipfile.ZipFile(io.BytesIO(_payload)) as _archive:
        _names = [i.filename for i in _archive.infolist() if not i.is_dir()]
        _prefix = _folder + '/'
        _prefixed = all(n.startswith(_prefix) for n in _names)
        _destination = _base if _prefixed else _base/_folder
        for _item in _archive.infolist():
            _target = (_destination/_item.filename).resolve()
            if not _target.is_relative_to(_destination.resolve()):
                raise ValueError('材料包中有越界路径，停止解压。')
            if _item.is_dir():
                _target.mkdir(parents=True, exist_ok=True)
            elif not _target.exists():
                _target.parent.mkdir(parents=True, exist_ok=True)
                _target.write_bytes(_archive.read(_item))
    _ready = next((p for p in _choices if (p/_sentinel).is_file()), None)
assert _ready is not None, '请将本讲完整材料包解压到当前项目的 ' + _folder + '/；保留文件结构。'
LAB_ROOT = _ready
print('材料目录已就绪：', LAB_ROOT)

import json, hashlib, platform, sys
from pathlib import Path
import numpy as np
import matplotlib.pyplot as plt
ROOT=LAB_ROOT
if not (ROOT/'assets').is_dir() and (ROOT/'lab10-patch'/'assets').is_dir():ROOT=ROOT/'lab10-patch'
assert (ROOT/'assets').is_dir(), '请解压本实验完整包，并在该目录或其父目录运行。'
OUT=ROOT/'runs/current'; OUT.mkdir(parents=True,exist_ok=True)
FIG=OUT/'figures'; FIG.mkdir(exist_ok=True)
plt.rcParams.update({'figure.dpi':120,'font.size':10,'axes.spines.top':False,'axes.spines.right':False})
BLUE,GOLD,GREEN,PURPLE='#002676','#E6A500','#00856A','#69458D'
def sha_file(path):return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def savefig(name):
    if not plt.gcf().get_constrained_layout():plt.tight_layout()
    plt.savefig(FIG/name,dpi=140,bbox_inches='tight');plt.show()

provenance=json.loads((ROOT/'assets/provenance.json').read_text())
assert sha_file(ROOT/'assets/target.npz')==provenance['target_asset_sha256']
with np.load(ROOT/'assets/target.npz',allow_pickle=False) as z:
    images=z['images'];sample_ids=z['official_train_ids'];split=z['split']
# digit_labels are in the source archive but deliberately never read here.
assert len(set(sample_ids))==640
train=np.flatnonzero(split==0);heldout=np.flatnonzero(split==2)
assert not set(sample_ids[train])&set(sample_ids[heldout])
partition=np.random.default_rng(20260908).permutation(60000)
assert np.array_equal(sample_ids,partition[12000:12640])
x=images.astype(np.float32)/255
shifted=np.zeros_like(x);shifted[:,:,:,2:]=x[:,:,:,:-2];x=shifted
assert x.shape==(640,1,28,28)
print('无标签 fit 输入：',len(train),'张；保留：',len(heldout),'张。没有导入 torch、加载 CNN 或读取类别标签。')

# Actual Lloyd updates on image patches. Labels are deliberately absent from fit().
N_WORDS=16
PATCH=7
patches=x.reshape(640,1,4,PATCH,4,PATCH).transpose(0,2,4,1,3,5).reshape(640,16,PATCH*PATCH)
fit_patches=patches[train].reshape(-1,PATCH*PATCH)
fit_patches=fit_patches[fit_patches.mean(1)>.01].astype(np.float64)
def fit_lloyd_unlabeled(a,k,seed=20260918,max_iter=80):
    rng=np.random.default_rng(seed)
    centers=a[rng.choice(len(a),k,replace=False)].copy()
    trace=[]
    for iteration in range(max_iter):
        distances=((a[:,None,:]-centers[None,:,:])**2).sum(2)
        assignments=distances.argmin(1)
        new=np.stack([a[assignments==j].mean(0) if np.any(assignments==j) else centers[j] for j in range(k)])
        new_distances=((a[:,None,:]-new[None,:,:])**2).sum(2)
        sse=float(new_distances.min(1).sum())
        trace.append(sse)
        if np.allclose(new,centers,rtol=0,atol=1e-10):
            centers=new; break
        centers=new
    assert np.all(np.diff(trace)<=1e-7)
    return centers,trace
centers,trace=fit_lloyd_unlabeled(fit_patches,N_WORDS)
frozen_centers=centers.copy()
test_patches=patches[heldout].reshape(-1,49)
test_dist=((test_patches[:,None,:]-frozen_centers[None,:,:])**2).sum(2)
test_assignment=test_dist.argmin(1)
assert np.array_equal(centers,frozen_centers)
kmeans_result={'identity':'L-A true MNIST pixel-patch Lloyd fit','labels_used_in_fit':False,
 'patch_shape':[7,7],'n_words':N_WORDS,'train_images':len(train),'train_nonblank_patches':len(fit_patches),
 'fit_sample_ids':sample_ids[train].tolist(),'evaluation_sample_ids':sample_ids[heldout].tolist(),
 'train_sse_trace':trace,'heldout_mean_squared_pixel_error':float(test_dist.min(1).mean()/49),
 'warning':'train excludes nearblank patches; heldout covers all patches, so these objectives are not identically weighted'}
np.savez_compressed(OUT/'kmeans-centers.npz',centers=centers)
(OUT/'kmeans-results.json').write_text(json.dumps(kmeans_result,indent=2))
fig=plt.figure(figsize=(12,4.4),layout='constrained')
outer=fig.add_gridspec(1,2,width_ratios=[1,1.3],wspace=.12)
grid=outer[0].subgridspec(int(np.ceil(N_WORDS/4)),4,wspace=.06,hspace=.12)
for j,c in enumerate(centers):
    ax=fig.add_subplot(grid[j//4,j%4]);ax.imshow(c.reshape(7,7),cmap='gray',vmin=0,vmax=1);ax.set_title(f'word {j}',fontsize=8);ax.axis('off')
ax=fig.add_subplot(outer[1]);ax.plot(np.arange(1,len(trace)+1),trace,color=BLUE,marker='.',ms=4)
ax.set(xlabel='Lloyd iteration',ylabel='Training sum of squared distances',title='Real image patches; no digit labels')
savefig('02-unlabeled-fit.png')
print('Fitted nonblank patches:',len(fit_patches),'iterations:',len(trace))
print('Held-out pixel reconstruction MSE:',kmeans_result['heldout_mean_squared_pixel_error'])

# 当前单元无法确认宿主平台或是否使用全新进程；unknown 不等于 False。
verification={'identity':'v17 independent patch-fit','fresh_process':'unknown','model_loaded':False,'labels_read':False,'labels_used_in_fit':False,'dictionary_frozen_during_evaluation':bool(np.array_equal(centers,frozen_centers)),'monotonic_sse':bool(np.all(np.diff(trace)<=1e-7)),'local_run':'unknown','cloud_run':'unknown','environment':{'python':platform.python_version(),'numpy':np.__version__},'target_sha256':sha_file(ROOT/'assets/target.npz'),'heldout_mse':kmeans_result['heldout_mean_squared_pixel_error']}
(OUT/'verification.json').write_text(json.dumps(verification,indent=2));print(json.dumps(verification,indent=2))