LAB 08 · 从真实响应到遮挡对照

import copy, gzip, hashlib, io, json, os, time
from pathlib import Path
from collections import OrderedDict
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle
import torch
from torch import nn
from torch.nn import functional as F
torch.set_num_threads(2)
torch.manual_seed(20260917)

MODEL_PATH=Path('results/cnn-plain-state.pt')
TEST_FILES={
 't10k-images-idx3-ubyte.gz':'9fb629c4189551a2d022fa330f9573f3',
 't10k-labels-idx1-ubyte.gz':'ec29112dd5afa0611ce80d1b7f02629c'}
missing=[str(p) for p in [MODEL_PATH,*[Path('data')/n for n in TEST_FILES]] if not p.is_file()]
if missing:
    raise RuntimeError('请先按 LAB 02 默认流程产生这些项目文件，再返回本页：'+', '.join(missing))
asset_bytes={'cnn-plain-state.pt':MODEL_PATH.read_bytes()}
for name,md5 in TEST_FILES.items():
    raw=(Path('data')/name).read_bytes()
    assert hashlib.md5(raw).hexdigest()==md5, 'MNIST 文件不匹配：'+name
    asset_bytes[name]=raw
model_sha256=hashlib.sha256(asset_bytes['cnn-plain-state.pt']).hexdigest()
COURSE_MODEL_SHA256='ba8b738cc67d4ec3cd83d75a5dbe8a70ec548377335ba0abaaa7c23cd14eeba3'
print('课件同一检查点:',model_sha256==COURSE_MODEL_SHA256)

source_state=torch.load(io.BytesIO(asset_bytes['cnn-plain-state.pt']),map_location='cpu',weights_only=True)
def make_cnn():
    return nn.Sequential(OrderedDict([
      ('conv1',nn.Conv2d(1,8,3,padding=1)),('relu1',nn.ReLU()),('pool1',nn.MaxPool2d(2)),
      ('conv2',nn.Conv2d(8,16,3,padding=1)),('relu2',nn.ReLU()),('pool2',nn.MaxPool2d(2)),
      ('flatten',nn.Flatten()),('hidden',nn.Linear(784,32)),('relu3',nn.ReLU()),('logits',nn.Linear(32,10))]))
source_model=make_cnn();source_model.load_state_dict(source_state);source_model.eval()
assert sum(p.numel() for p in source_model.parameters())==26698
print('已读取项目内的 26698 参数 CNN；PyTorch',torch.__version__)
print('当前模型 SHA-256:',model_sha256)

raw_images=gzip.decompress(asset_bytes['t10k-images-idx3-ubyte.gz'])
raw_labels=gzip.decompress(asset_bytes['t10k-labels-idx1-ubyte.gz'])
images=np.frombuffer(raw_images,dtype=np.uint8,offset=16).reshape(10000,1,28,28)[:3].copy()
labels=np.frombuffer(raw_labels,dtype=np.uint8,offset=8)[:3].copy()
assert labels.tolist()==[7,2,1]
inputs=torch.tensor(images,dtype=torch.float32)/255
with torch.no_grad():
    relu1=source_model.relu1(source_model.conv1(inputs))
    relu2=source_model.relu2(source_model.conv2(source_model.pool1(relu1)))
    logits=source_model(inputs)
measured={'inputs':inputs.numpy(),'labels':labels,'test_indices':np.arange(3),
          'relu1':relu1.numpy(),'relu2':relu2.numpy(),'logits':logits.numpy()}
response_manifest={'color':{'layers':{
 'relu1':{'max':max(float(relu1.max()),1e-12)},
 'relu2':{'max':max(float(relu2.max()),1e-12)}}}}
print('固定 test 序号:',measured['test_indices'].tolist(),'真实标签:',labels.tolist())
print('模型预测:',logits.argmax(1).tolist())
print('本次同层统一色标:',response_manifest['color']['layers'])
print('模型 SHA-256:',model_sha256)

LAYER='relu2'   # relu1 / relu2
CHANNEL=0      # L1: 0..7; L2: 0..15
assert LAYER in ('relu1','relu2')
activation={'relu1':relu1.numpy(),'relu2':relu2.numpy()}[LAYER]
assert 0<=CHANNEL<activation.shape[1]
vmax=response_manifest['color']['layers'][LAYER]['max']
def support(row,col,layer):
    size,jump,offset=(3,1,-1) if layer=='relu1' else (8,2,-3)
    top,left=row*jump+offset,col*jump+offset
    return max(0,top),max(0,left),min(28,top+size),min(28,left+size)
def peak_summary(a):
    a=np.asarray(a)
    assert a.ndim==2 and np.isfinite(a).all()
    peak=float(a.max())
    ties=np.argwhere(a==peak)
    # These are post-ReLU responses. Zero has no meaningful positive hotspot.
    if peak<=0:
        return {'peak':peak,'peak_row_col':None,'tied_maxima':len(ties),'has_positive_response':False}
    row,col=map(int,ties[0])
    return {'peak':peak,'peak_row_col':[row,col],'tied_maxima':len(ties),'has_positive_response':True}
fig,axes=plt.subplots(2,3,figsize=(10,6),layout='constrained')
observations=[]
for j in range(3):
    a=activation[j,CHANNEL];summary=peak_summary(a)
    axes[0,j].imshow(measured['inputs'][j,0],cmap='gray',vmin=0,vmax=1)
    axes[0,j].set_title(f"test {j}: label {measured['labels'][j]}")
    im=axes[1,j].imshow(a,cmap='Blues',vmin=0,vmax=vmax,interpolation='nearest')
    if summary['has_positive_response']:
        row,col=summary['peak_row_col']
        top,left,bottom,right=support(row,col,LAYER)
        axes[0,j].add_patch(Rectangle((left-.5,top-.5),right-left,bottom-top,fill=False,color='#FDB515',lw=2))
        axes[1,j].add_patch(Rectangle((col-.5,row-.5),1,1,fill=False,color='#FDB515',lw=2))
        axes[1,j].set_title(f'{LAYER}/c{CHANNEL}'+chr(10)+f'peak ({row},{col}) = {summary["peak"]:.3f}; ties={summary["tied_maxima"]}')
        summary['support']=[top,left,bottom,right]
    else:
        axes[1,j].set_title(f'{LAYER}/c{CHANNEL}'+chr(10)+'no positive response; no hotspot')
        summary['support']=None
    observations.append({'test_index':j,**summary})
fig.colorbar(im,ax=axes.ravel().tolist(),shrink=.7,label='response (same layer scale)')
plt.show()
print(json.dumps(observations,ensure_ascii=False,indent=2))

SAMPLE=0
TARGET=int(measured['labels'][SAMPLE])
PATCH_SIZE=5
FILL=0.0
PATCH_A=(19,10)
PATCH_B=(0,0)
assert 0<=SAMPLE<3 and 0<=TARGET<10 and 0<PATCH_SIZE<=28
def occlude(x,corner):
    row,col=corner
    assert 0<=row<=28-PATCH_SIZE and 0<=col<=28-PATCH_SIZE
    z=x.clone();z[:,:,row:row+PATCH_SIZE,col:col+PATCH_SIZE]=FILL
    return z
original=inputs[SAMPLE:SAMPLE+1]
variants=torch.cat([original,occlude(original,PATCH_A),occlude(original,PATCH_B)])
with torch.no_grad():
    scores=source_model(variants)[:,TARGET].numpy()
fig,axes=plt.subplots(1,3,figsize=(9,3))
change=(variants-original).abs()
changed_pixels=(change>0).flatten(1).sum(1).tolist()
absolute_change=change.flatten(1).sum(1).tolist()
for j,(ax,x,score,name) in enumerate(zip(axes,variants[:,0],scores,['original','patch A','patch B'])):
    ax.imshow(x,cmap='gray',vmin=0,vmax=1)
    if j:
        row,col=(PATCH_A,PATCH_B)[j-1]
        ax.add_patch(Rectangle((col-.5,row-.5),PATCH_SIZE,PATCH_SIZE,fill=False,color='#FDB515',lw=2))
    ax.set_title(f'{name}: logit={score:.3f}'+chr(10)+f'{changed_pixels[j]} changed pixels');ax.axis('off')
plt.show()
occlusion_result={'sample':SAMPLE,'target':TARGET,'score':'target logit','patch_size':PATCH_SIZE,'fill':FILL,
                  'patch_A':PATCH_A,'patch_B':PATCH_B,'scores':scores.tolist(),
                  'changed_pixels':changed_pixels,'absolute_pixel_change_sum':absolute_change,
                  'original_minus_A':float(scores[0]-scores[1]),'original_minus_B':float(scores[0]-scores[2])}
print(json.dumps(occlusion_result,ensure_ascii=False,indent=2))