Remaster twenty-nine original pictures with reproducible artwork tools
This commit is contained in:
1 parent
6e4008c62f
commit
668b3ff86d
253 files changed
+670875
-79
No files matched your search
@@ -0,0 +1,259 @@
|
||||
"""Narrow DERIVED STAG03 SourceScale4 geometry/full-generated-RGB dispatcher.
|
||||
|
||||
Dependency implementations execute from authenticated buffered bytes in fresh
|
||||
namespaces. Input files are read once through no-follow, contained descriptors.
|
||||
This tool performs no engine operation or artwork approval.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import argparse
|
||||
import builtins
|
||||
import ctypes
|
||||
from fractions import Fraction
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import stat
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import numpy as np
|
||||
from PIL import Image, __version__ as PILLOW_VERSION
|
||||
|
||||
METHOD='derived-STAG03-fullcanvas-rational-generated-RGB-SourceScale4-warning66-v1'
|
||||
FREE_FLOOR=3_000_000_000
|
||||
OUTPUT_RESERVE=16_000_000
|
||||
PINS={
|
||||
'decoder':'66184a4758027dadbadb5e20fe90b2add2b9fc218b7810304b1a1ea2de74f58e',
|
||||
'geometry':'c6123e9591f0a169846677019ddcb944bf9c2b4c9184a7d8ab02a08546de409b',
|
||||
'functional':'f8078284e300947aec4d2c9e70cbc7198fa3cb5d34de6b36c07cacc5f3820a84',
|
||||
}
|
||||
GROUPS=[sorted([[x,y]for x in range(l,r)for y in range(13,16)])for l,r in ((1,12),(19,30))]
|
||||
TOOL_BYTES=Path(__file__).read_bytes()
|
||||
|
||||
def sha(payload):return hashlib.sha256(payload).hexdigest()
|
||||
|
||||
def route(value):
|
||||
if (type(value)is not str or not value or '\\'in value or ':'in value
|
||||
or Path(value).is_absolute() or Path(value).as_posix()!=value
|
||||
or '..'in Path(value).parts):
|
||||
raise ValueError('canonical contained relative path required')
|
||||
return Path(value)
|
||||
|
||||
def safe_read(base, binding, limit):
|
||||
if (type(binding)is not dict or set(binding)!={'path','sha256'}
|
||||
or type(binding['sha256'])is not str
|
||||
or re.fullmatch('[0-9a-f]{64}',binding['sha256'])is None):
|
||||
raise ValueError('exact path/SHA256 binding required')
|
||||
relative=route(binding['path'])
|
||||
flags=os.O_RDONLY|os.O_NOFOLLOW
|
||||
directory=os.open(base,flags|os.O_DIRECTORY)
|
||||
try:
|
||||
for part in relative.parts[:-1]:
|
||||
next_fd=os.open(part,flags|os.O_DIRECTORY,dir_fd=directory)
|
||||
os.close(directory);directory=next_fd
|
||||
fd=os.open(relative.name,flags,dir_fd=directory)
|
||||
try:
|
||||
before=os.fstat(fd)
|
||||
if not stat.S_ISREG(before.st_mode) or before.st_size>limit:
|
||||
raise ValueError('bounded regular input required')
|
||||
chunks=[];size=0
|
||||
while True:
|
||||
chunk=os.read(fd,min(65536,limit+1-size))
|
||||
if not chunk:break
|
||||
chunks.append(chunk);size+=len(chunk)
|
||||
if size>limit:raise ValueError('input exceeds byte ceiling')
|
||||
payload=b''.join(chunks)
|
||||
after=os.fstat(fd)
|
||||
if (before.st_dev,before.st_ino,before.st_size,before.st_mtime_ns)!=(after.st_dev,after.st_ino,after.st_size,after.st_mtime_ns):
|
||||
raise ValueError('input changed during descriptor read')
|
||||
finally:os.close(fd)
|
||||
finally:os.close(directory)
|
||||
if sha(payload)!=binding['sha256']:raise ValueError('input SHA256 mismatch')
|
||||
return payload
|
||||
|
||||
def verified_modules(payloads):
|
||||
"""Never use importlib/SourceFileLoader or re-open verified dependency paths."""
|
||||
if set(payloads)!=set(PINS) or any(sha(payloads[k])!=v for k,v in PINS.items()):
|
||||
raise ValueError('reviewed dependency bytes differ')
|
||||
loaded={}
|
||||
real_import=builtins.__import__
|
||||
proxy_sys=types.SimpleNamespace(path=list(sys.path),version=sys.version)
|
||||
def importer(name,globals=None,locals=None,fromlist=(),level=0):
|
||||
if name in loaded:return loaded[name]
|
||||
if name=='sys':return proxy_sys
|
||||
return real_import(name,globals,locals,fromlist,level)
|
||||
for key,name in (('decoder','hd_frame_policy'),('geometry','hd_dshp20_cleanup'),('functional','hd_other_functional_cleanup')):
|
||||
module=types.ModuleType(name)
|
||||
module.__file__=f'/authenticated-buffered-dependency/{name}.py'
|
||||
module.__dict__['__builtins__']={**vars(builtins),'__import__':importer}
|
||||
exec(compile(payloads[key],module.__file__,'exec'),module.__dict__)
|
||||
loaded[name]=module
|
||||
return loaded['hd_frame_policy'],loaded['hd_dshp20_cleanup'],loaded['hd_other_functional_cleanup']
|
||||
|
||||
def rational(value):
|
||||
if type(value)is int and abs(value)<=10**9:return Fraction(value)
|
||||
if type(value)is not str or re.fullmatch('-?[0-9]+(?:/[1-9][0-9]*)?',value)is None:
|
||||
raise ValueError('finite exact integer/rational string required')
|
||||
result=Fraction(value)
|
||||
if abs(result.numerator)>10**9 or result.denominator>10**9:
|
||||
raise ValueError('rational exceeds bounded arithmetic')
|
||||
return result
|
||||
|
||||
def rectangle(value,w,h,label):
|
||||
if type(value)is not list or len(value)!=4 or any(type(x)is not int for x in value):
|
||||
raise ValueError(label+' needs four integers')
|
||||
l,t,r,b=value
|
||||
if not 0<=l<r<=w or not 0<=t<b<=h:raise ValueError(label+' is out of bounds')
|
||||
return value
|
||||
|
||||
def alpha_bounds(mask):
|
||||
yy,xx=np.nonzero(mask)
|
||||
if not len(xx):raise ValueError('empty occupied support')
|
||||
return [int(xx.min()),int(yy.min()),int(xx.max()+1),int(yy.max()+1)]
|
||||
|
||||
def render(recipe,inputs,modules):
|
||||
decoder,geometry,functional=modules
|
||||
if (type(recipe)is not dict or set(recipe)!={'format_version','method','asset','source','palette','parent_raw',
|
||||
'logical_canvas','origin','output_scale','alignment','max_fill_distance_pixels','functional_groups',
|
||||
'dependencies','expected_tool_sha256'} or type(recipe['format_version'])is not int
|
||||
or recipe['format_version']!=1 or recipe['method']!=METHOD or recipe['asset']!='300ae'):
|
||||
raise ValueError('exact STAG03 recipe fields/version/asset required')
|
||||
if recipe['logical_canvas']!=[32,32] or recipe['origin']!=[0,0] or type(recipe['output_scale'])is not int or recipe['output_scale']!=4:
|
||||
raise ValueError('native32 canvas/origin0/scale4 required')
|
||||
if any(type(x)is not int for x in recipe['logical_canvas']+recipe['origin']):
|
||||
raise ValueError('literal integer canvas/origin required')
|
||||
if type(recipe['max_fill_distance_pixels'])is not int or recipe['max_fill_distance_pixels']!=12:
|
||||
raise ValueError('exact twelve-output-pixel noncascading donor bound required')
|
||||
if recipe['functional_groups']!=GROUPS or any(type(x)is not int for g in recipe['functional_groups']for xy in g for x in xy):
|
||||
raise ValueError('exact two disjoint sorted33-cell functional groups required')
|
||||
w,h,cells=decoder._indices(inputs['source'])
|
||||
if (w,h)!=(32,32):raise ValueError('authentic Source canvas differs')
|
||||
source=np.asarray(cells,np.uint16).reshape(h,w)
|
||||
if (source<256).sum()<264 or any(source[y,x]>=256 for g in GROUPS for x,y in g):
|
||||
raise ValueError('all66 cells must be painted Source and within quarter bound')
|
||||
pb=inputs['palette']
|
||||
if len(pb)!=768 or max(pb)>63:raise ValueError('exact six-bit Original palette required')
|
||||
p6=np.frombuffer(pb,np.uint8).reshape(256,3).astype(np.uint16)
|
||||
palette=((p6<<2)|(p6>>4)).astype(np.uint8)
|
||||
with Image.open(io.BytesIO(inputs['parent_raw']))as image:
|
||||
if image.format!='PNG' or image.width*image.height>4_194_304:
|
||||
raise ValueError('bounded PNG parent required')
|
||||
parent=image.convert('RGBA')
|
||||
alignment=recipe['alignment']
|
||||
if type(alignment)is not dict or set(alignment)!={'raw_painted_bounds','source_painted_bounds','inverse_affine'}:
|
||||
raise ValueError('exact alignment declarations required')
|
||||
rl,rt,rr,rb=rectangle(alignment['raw_painted_bounds'],*parent.size,'raw bounds')
|
||||
sl,st,sr,sb=rectangle(alignment['source_painted_bounds'],w,h,'Source bounds')
|
||||
if alpha_bounds(np.asarray(parent)[:,:,3]>=128)!=[rl,rt,rr,rb] or alpha_bounds(source<256)!=[sl,st,sr,sb]:
|
||||
raise ValueError('declared bounds differ from actual complete Source/raw support')
|
||||
values=alignment['inverse_affine']
|
||||
if type(values)is not list or len(values)!=6:raise ValueError('six inverse affine rationals required')
|
||||
affine=list(map(rational,values))
|
||||
ax,ay=Fraction(rr-rl,(sr-sl)*4),Fraction(rb-rt,(sb-st)*4)
|
||||
expected=[ax,Fraction(0),rl-ax*sl*4,Fraction(0),ay,rt-ay*st*4]
|
||||
if affine!=expected:raise ValueError('affine differs from complete Source/raw support alignment')
|
||||
sampled=np.asarray(parent.transform((128,128),Image.Transform.AFFINE,
|
||||
tuple(map(float,affine)),Image.Resampling.BICUBIC)).copy()
|
||||
s4=geometry.scale2(geometry.scale2(source));mask=s4<256
|
||||
trusted=sampled[:,:,3]>=128
|
||||
rgb,fill=geometry.fill_generated_rgb(sampled[:,:,:3],trusted,mask,s4,12)
|
||||
body=np.zeros((128,128,4),np.uint8)
|
||||
body[:,:,:3][mask]=rgb[mask];body[:,:,3][mask]=255
|
||||
out=body;retentions=[];selected=np.zeros(mask.shape,bool)
|
||||
for group in GROUPS:
|
||||
out,receipt=functional.retain_functional_original(out,source,palette,group)
|
||||
retentions.append(receipt)
|
||||
for x,y in group:selected[y*4:(y+1)*4,x*4:(x+1)*4]=True
|
||||
if not np.array_equal(out[:,:,3],body[:,:,3]) or not np.array_equal(out[~selected],body[~selected]):
|
||||
raise ValueError('functional retention changed alpha or undeclared body samples')
|
||||
outside=mask&~selected
|
||||
direct=outside&trusted
|
||||
if int(direct.sum())*2<int(mask.sum()):raise ValueError('less than half complete body retains direct generated RGB')
|
||||
if not np.array_equal(out[:,:,:3][direct],sampled[:,:,:3][direct]):
|
||||
raise ValueError('direct generated RGB was changed')
|
||||
# Boundary interpolation is not evidence of preserved generated detail.
|
||||
# Check fully direct, unpatched interior cells, excluding a one-cell rim.
|
||||
interior=np.zeros((32,32),bool)
|
||||
for y in range(1,31):
|
||||
for x in range(1,31):interior[y,x]=bool(np.all(source[y-1:y+2,x-1:x+2]<256))
|
||||
for group in GROUPS:
|
||||
for x,y in group:interior[y,x]=False
|
||||
detail_cells=0
|
||||
for y,x in np.argwhere(interior):
|
||||
ys,xs=slice(y*4,(y+1)*4),slice(x*4,(x+1)*4)
|
||||
if np.all(direct[ys,xs]) and np.any(out[ys,xs,:3]!=out[y*4,x*4,:3]):detail_cells+=1
|
||||
if not detail_cells:
|
||||
raise ValueError('no generated interior detail remains')
|
||||
return out,dict(output_class='DERIVED_NOT_UNTOUCHED_GENERATION',
|
||||
native=[32,32],origin=[0,0],output_scale=4,fullcanvas_inverse_affine=[str(x)for x in affine],
|
||||
alpha_exact_SourceScale4=True,inputs_not_written_by_dispatcher=True,
|
||||
Source_painted_cells=int((source<256).sum()),functional_Original_cells=66,
|
||||
actual_Original_Scale4_RGBA_samples=int(selected.sum()),functional_Original_not_HD_credit=True,
|
||||
functional_groups=retentions,outside_functional_RGBA_exact=True,
|
||||
direct_generated_RGB_samples=int(direct.sum()),total_painted_samples=int(mask.sum()),
|
||||
direct_generated_nonconstant_interior_cells=detail_cells,
|
||||
noncascading_generated_RGB_fill=fill,actual_functional_calls=2,
|
||||
runtime_versions=dict(python=sys.version,numpy=np.__version__,Pillow=PILLOW_VERSION))
|
||||
|
||||
def publish_no_replace(staging,output):
|
||||
"""Atomic directory install, never replace even an empty existing directory."""
|
||||
libc=ctypes.CDLL(None,use_errno=True)
|
||||
rename=getattr(libc,'renameat2',None)
|
||||
if rename is None:raise RuntimeError('Linux renameat2 required for no-replace publication')
|
||||
rename.argtypes=[ctypes.c_int,ctypes.c_char_p,ctypes.c_int,ctypes.c_char_p,ctypes.c_uint]
|
||||
rename.restype=ctypes.c_int
|
||||
fd=os.open(output.parent,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW)
|
||||
try:
|
||||
if rename(fd,os.fsencode(staging.name),fd,os.fsencode(output.name),1)!=0:
|
||||
number=ctypes.get_errno();raise OSError(number,os.strerror(number),str(output))
|
||||
finally:os.close(fd)
|
||||
|
||||
def run_recipe(recipe_path,output):
|
||||
recipe_path=Path(recipe_path)
|
||||
if recipe_path.is_symlink():raise ValueError('recipe must not be a symlink')
|
||||
base=recipe_path.parent.resolve(strict=True)
|
||||
output=Path(output)
|
||||
output=output if output.is_absolute()else base/output
|
||||
if output.parent.resolve(strict=True)!=base or output.name in ('','.','..'):
|
||||
raise ValueError('fresh output must be a direct child of recipe root')
|
||||
if output.exists() or output.is_symlink():raise ValueError('fresh absent output directory required')
|
||||
if shutil.disk_usage(base).free-OUTPUT_RESERVE<FREE_FLOOR:
|
||||
raise ValueError('three-gigabyte free floor plus bounded output reserve required')
|
||||
rb=recipe_path.read_bytes()
|
||||
if len(rb)>128*1024:raise ValueError('recipe exceeds metadata ceiling')
|
||||
recipe=json.loads(rb)
|
||||
if type(recipe)is not dict or recipe.get('expected_tool_sha256')!=sha(TOOL_BYTES):
|
||||
raise ValueError('actual tool bytes differ from declared reviewed dispatcher')
|
||||
deps=recipe.get('dependencies')
|
||||
if type(deps)is not dict or set(deps)!=set(PINS):raise ValueError('exact three dependencies required')
|
||||
payloads={k:safe_read(base,v,128*1024)for k,v in deps.items()}
|
||||
modules=verified_modules(payloads)
|
||||
inputs={k:safe_read(base,recipe[k],32*1024*1024 if k=='parent_raw' else 4*1024*1024)
|
||||
for k in ('source','palette','parent_raw')}
|
||||
rgba,report=render(recipe,inputs,modules)
|
||||
buffer=io.BytesIO();Image.fromarray(rgba).save(buffer,format='PNG',compress_level=9)
|
||||
png=buffer.getvalue()
|
||||
execution=dict(status='success',method=METHOD,recipe_sha256=sha(rb),tool_sha256=sha(TOOL_BYTES),
|
||||
source_sha256={'300ae':sha(inputs['source'])},parent_raw_sha256=[sha(inputs['parent_raw'])],
|
||||
output_sha256={'300ae':sha(png)})
|
||||
report.update(execution=execution,authenticated_dependency_bytes=PINS,
|
||||
dependency_execution='Fresh isolated namespaces executed from verified buffered bytes; no dependency/input path reopening for processing.',
|
||||
Source_decoder_sha256=PINS['decoder'],RGBA_sha256=sha(rgba.tobytes()))
|
||||
staging=Path(tempfile.mkdtemp(prefix='.stag03-derived-',dir=base))
|
||||
try:
|
||||
for name,payload in [('300ae-derived.png',png),('derived-provenance.json',(json.dumps(report,indent=2,sort_keys=True)+'\n').encode())]:
|
||||
with(staging/name).open('xb')as stream:stream.write(payload);stream.flush();os.fsync(stream.fileno())
|
||||
publish_no_replace(staging,output)
|
||||
finally:
|
||||
if staging.exists():shutil.rmtree(staging)
|
||||
return report
|
||||
|
||||
if __name__=='__main__':
|
||||
parser=argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('recipe',type=Path);parser.add_argument('fresh_output',type=Path)
|
||||
args=parser.parse_args()
|
||||
print(json.dumps(run_recipe(args.recipe,args.fresh_output)['execution'],sort_keys=True))
|
||||
@@ -0,0 +1,261 @@
|
||||
"""Reproduce the reviewed TARGT14 soil stage with portable checked inputs.
|
||||
|
||||
Pixel math for frames1..13 is copied verbatim from the fixed V4 diagnostic.
|
||||
Phase0 is the untouched genuine10375 PNG, not a generated substitute.
|
||||
A successful reproduction never grants artistic approval or registration.
|
||||
"""
|
||||
from pathlib import Path, PurePosixPath
|
||||
import argparse, hashlib, io, json, math, os, platform, re, shutil, stat, sys, tempfile, types
|
||||
import numpy as np
|
||||
import PIL
|
||||
from PIL import Image
|
||||
METHOD='derived-TARGT-shared-half-soil13-and-genuine10375-phase0-portable-v1'
|
||||
KERNEL_SHA='0e4d7d5a4f7a7ddf6e396e757691156dfc74d231d9925744c342078532b89a98'
|
||||
DECODER_SHA='66184a4758027dadbadb5e20fe90b2add2b9fc218b7810304b1a1ea2de74f58e'
|
||||
DONOR_SHA='a3bfdc1b398539e133d97400940256400e295dbbf1ecbc978f202799a6916204'
|
||||
PALETTE_SHA='1cf6adb926d1dee1888b3e3d33b51a7049b3e5eb5f90a3067ef71c8970547204'
|
||||
ANCESTOR_RAW_SHA='207d3e613a66390a0025b3f824533d90a668c25a0808358137f7268401add9eb'
|
||||
PHASE0_SHA='87a7476a06ebb210304c54e65d132cf2003ba43e4670459ca2155ed9bece945e'
|
||||
PARAMETERS=dict(output_scale=8,Source_macro_sigma_logical=.5,Source_macro_sigma_output_pixels=4.,
|
||||
donor_fine_sigma_logical=.5,donor_fine_sigma_output_pixels=4.,absolute_luma_fine_gain=1.,
|
||||
Source_RGB_mean_control='additive-per-phase-group-only-no-variance-boost',
|
||||
donor_transfer='raw-fine-only-linear-Source-group-hue-no-mid-normalization-or-exp')
|
||||
LUMA=np.array([.2126,.7152,.0722])
|
||||
def sha(body):return hashlib.sha256(body).hexdigest()
|
||||
|
||||
def groups(source,soil):
|
||||
dark=soil&(source>=204)&(source<=206)
|
||||
return {'brown_field':soil&~dark,'dark_ordinary_brown':dark}
|
||||
|
||||
def expand(mask,scale):return np.repeat(np.repeat(mask,scale,0),scale,1)
|
||||
|
||||
def logical_average(rgb,scale=8):return rgb.reshape(32,scale,32,scale,3).mean((1,3))
|
||||
|
||||
def logical_masks(soil):
|
||||
padded=np.pad(soil,1,constant_values=False);eroded=soil.copy()
|
||||
for dy in range(3):
|
||||
for dx in range(3):eroded&=padded[dy:dy+32,dx:dx+32]
|
||||
boundary=soil&~eroded;top=np.zeros_like(soil)
|
||||
if soil.any():top[np.nonzero(soil)[0].min()]=soil[np.nonzero(soil)[0].min()]
|
||||
return boundary,eroded,top
|
||||
|
||||
def diagnostics(rgb,source,palette,soil):
|
||||
if not soil.any():return dict(mutable_cells=0,failures=[])
|
||||
actual=logical_average(rgb)@LUMA;original=palette[np.minimum(source,255)]@LUMA
|
||||
out=dict(mutable_cells=int(soil.sum()),Source_RGB_mean=palette[source[soil]].mean(0).tolist(),
|
||||
actual_RGB_mean=logical_average(rgb)[soil].mean(0).tolist(),
|
||||
Source_luma_std=float(original[soil].std()),actual_luma_std=float(actual[soil].std()),
|
||||
block_contrast={},edges={},failures=[])
|
||||
error=np.abs(np.array(out['actual_RGB_mean'])-out['Source_RGB_mean'])
|
||||
if np.any(error>1.):out['failures'].append('Source own-phase RGB mean exceeds1level')
|
||||
if original[soil].std()>.05:
|
||||
corr=float(np.corrcoef(original[soil],actual[soil])[0,1]) if actual[soil].std()>1e-12 else 0.;out['Source_pattern_correlation']=corr
|
||||
if not math.isfinite(corr) or corr<.90:out['failures'].append('Source own-phase pattern correlation below0.90')
|
||||
for side in [2,4,8]:
|
||||
a=[];b=[]
|
||||
for y in range(0,32,side):
|
||||
for x in range(0,32,side):
|
||||
if soil[y:y+side,x:x+side].all():
|
||||
a.append(original[y:y+side,x:x+side].mean());b.append(actual[y:y+side,x:x+side].mean())
|
||||
target=float(np.std(a)) if a else 0.;measured=float(np.std(b)) if b else 0.;out['block_contrast'][str(side)]=dict(full_blocks=len(a),Source_std=target,actual_std=measured)
|
||||
if len(a)>=4 and target>.05 and measured<.90*target:out['failures'].append('Source coarse block contrast below90percent at'+str(side))
|
||||
boundary,interior,top=logical_masks(soil)
|
||||
for name,edge in [('all_inner_boundary',boundary),('top_inner_boundary',top)]:
|
||||
if not interior.any() or not edge.any():
|
||||
out['edges'][name]=dict(status='NOT_APPLICABLE_NO_ERODED_INTERIOR_OR_EDGE');continue
|
||||
target=float(original[edge].mean()-original[interior].mean());measured=float(actual[edge].mean()-actual[interior].mean())
|
||||
out['edges'][name]=dict(Source_minus_interior=target,actual_minus_interior=measured)
|
||||
bad=abs(measured-target)>2.
|
||||
if abs(target)>=1.:bad=bad or measured*target<=0 or abs(measured)<.90*abs(target)
|
||||
if bad:out['failures'].append('Source '+name+' tone/contrast not preserved')
|
||||
return out
|
||||
|
||||
def donor_bands(donor_rgb,kernel):
|
||||
donor=np.asarray(donor_rgb)
|
||||
if donor.ndim!=3 or donor.shape[2]!=3 or donor.dtype!=np.uint8:raise ValueError('donor RGB uint8 required')
|
||||
d=np.asarray(Image.fromarray(donor).resize((256,256),Image.Resampling.LANCZOS),float)@LUMA
|
||||
fine=d-kernel.gaussian(d,.5*8)
|
||||
if float(fine.std())<.25:raise ValueError('donor has no coherent generated material')
|
||||
# Absolute measured luma, not percentile-normalized/log lighting. No MID.
|
||||
return [fine],dict(fine=dict(std=float(fine.std()),mean=float(fine.mean()),absolute_luma_gain=1.,mid_used=False))
|
||||
|
||||
def apply_frame(accepted,source,palette,static,bands,kernel):
|
||||
if not isinstance(source,np.ndarray) or source.shape!=(32,32) or source.dtype.kind not in 'iu' or np.any(source<0) or np.any(source>256):raise ValueError('Source32 integer labels required')
|
||||
if not isinstance(palette,np.ndarray) or palette.shape!=(256,3) or palette.dtype!=np.uint8:raise ValueError('RGB uint8 palette required')
|
||||
if not isinstance(static,np.ndarray) or static.shape!=(32,32) or static.dtype!=np.bool_:raise ValueError('boolean static plane required')
|
||||
if not isinstance(accepted,np.ndarray) or accepted.shape!=(256,256,4) or accepted.dtype!=np.uint8:raise ValueError('accepted256RGBA required')
|
||||
if len(bands)!=1 or any(not isinstance(b,np.ndarray) or b.shape!=(256,256) or b.dtype.kind!='f' or not np.isfinite(b).all() or np.any(np.abs(b)>255.0000001) for b in bands):raise ValueError('one raw finite luma fine field required')
|
||||
if not np.array_equal(accepted[:,:,3],expand(source<256,8)*255):raise ValueError('accepted Source alpha mismatch')
|
||||
soil=(source>=194)&(source<=239)&~static;out=accepted.copy()
|
||||
if not soil.any():return out,dict(mutable_cells=0,protected_RGBA_exact=True,alpha_exact=True,failures=[])
|
||||
Source_rgb=palette[np.minimum(source,255)].astype(float)
|
||||
actual_groups=groups(source,soil)
|
||||
base=np.zeros((256,256,3),float);modulated=base.copy();group_metrics={}
|
||||
for name,logical in actual_groups.items():
|
||||
if not logical.any():continue
|
||||
area=expand(logical,8);gm=logical.astype(float)
|
||||
# Direct bilinear Source/group interpolation, then same-group0.5logical
|
||||
# smoothing. No ScaleNx contour lattice and no per-cell variance boost.
|
||||
weights=kernel.gaussian(kernel.resize_float(gm,(256,256)),.5*8)
|
||||
numerator=kernel.gaussian(kernel.resize_float(Source_rgb*gm[:,:,None],(256,256)),.5*8)
|
||||
field=numerator/np.maximum(weights[:,:,None],1e-12)
|
||||
target=palette[source[logical]].astype(float);means=target.mean(0)
|
||||
field+=means-field[area].mean(0)
|
||||
base[area]=field[area]
|
||||
mean_luma=float(means@LUMA)
|
||||
if mean_luma<=1e-9:raise ValueError('raw fine transfer requires positive Source group luma')
|
||||
fine=bands[0]-bands[0][area].mean()
|
||||
# Fixed group hue is Source-owned and has unit Rec709 luma. Adding the
|
||||
# genuine raw fine field therefore changes luma by EXACTLY fine, with
|
||||
# no relative-brightness amplification and zero group-mean RGB drift.
|
||||
direction=means/mean_luma
|
||||
candidate=field+fine[:,:,None]*direction
|
||||
modulated[area]=candidate[area]
|
||||
group_metrics[name]=dict(logical_cells=int(logical.sum()),Source_RGB_mean=means.tolist(),
|
||||
Source_macro_sigma_logical=.5,Source_variance_boost=False,absolute_fine_luma_gain=1.,fine_luma_std=float(fine[area].std()),
|
||||
fine_luma_mean=float(fine[area].mean()),Source_owned_RGB_direction=direction.tolist(),direction_luma=float(direction@LUMA),mid_used=False,
|
||||
linear_unit_luma_before_rounding_and_clipping=True,preclip_out_of_RGB_range_samples=int(np.sum((candidate[area]<0)|(candidate[area]>255))))
|
||||
mask=expand(soil,8);out[mask,:3]=np.clip(np.rint(modulated[mask]),0,255).astype(np.uint8)
|
||||
if not np.array_equal(out[~mask],accepted[~mask]) or not np.array_equal(out[:,:,3],accepted[:,:,3]):raise AssertionError('protected RGBA/alpha changed')
|
||||
d=diagnostics(out[:,:,:3],source,palette,soil)
|
||||
d.update(groups=group_metrics,protected_RGBA_exact=True,alpha_exact=True,
|
||||
protected_RGBA_pixels=int((~mask).sum()),changed_soil_pixels=int(np.any(out!=accepted,2).sum()),
|
||||
source_macro_base_RGB_sha256=sha(np.clip(np.rint(base),0,255).astype(np.uint8).tobytes()))
|
||||
return out,d
|
||||
|
||||
def relative(value):
|
||||
if type(value) is not str or not value or '\\' in value or ':' in value:
|
||||
raise ValueError('portable nonempty POSIX relative path required')
|
||||
p=PurePosixPath(value)
|
||||
if p.is_absolute() or any(x in ('','..','.') for x in value.split('/')) or str(p)!=value:
|
||||
raise ValueError('normalized contained relative path required')
|
||||
return p
|
||||
|
||||
def reference(ref):
|
||||
if type(ref) is not dict or set(ref)!={'path','sha256'} or type(ref['sha256']) is not str or not re.fullmatch('[0-9a-f]{64}',ref['sha256']):
|
||||
raise ValueError('exact portable path/lowercase64hex SHA required')
|
||||
relative(ref['path']);return ref
|
||||
|
||||
def contained(root,name,existing=True):
|
||||
root=Path(root).resolve(strict=True);p=root
|
||||
for piece in relative(name).parts:
|
||||
p=p/piece
|
||||
if p.is_symlink():raise ValueError('symlink component forbidden')
|
||||
if not p.resolve(strict=False).is_relative_to(root):raise ValueError('outside project root')
|
||||
if existing and not p.is_file():raise ValueError('regular file required')
|
||||
return p
|
||||
|
||||
def read_regular(path,expected=None):
|
||||
path=Path(path);flags=os.O_RDONLY|getattr(os,'O_NOFOLLOW',0);fd=os.open(path,flags)
|
||||
try:
|
||||
a=os.fstat(fd)
|
||||
if not stat.S_ISREG(a.st_mode) or a.st_size>32*1024*1024:raise ValueError('bounded regular file required')
|
||||
with os.fdopen(fd,'rb',closefd=False) as f:body=f.read()
|
||||
b=os.fstat(fd)
|
||||
finally:os.close(fd)
|
||||
state=lambda s:(s.st_dev,s.st_ino,s.st_size,s.st_mtime_ns,s.st_mode)
|
||||
if state(a)!=state(b):raise ValueError('file changed during read')
|
||||
if expected is not None and sha(body)!=expected:raise ValueError('input SHA mismatch')
|
||||
return body
|
||||
|
||||
def load_verified(body,path,name):
|
||||
"""Execute exactly bytes already authenticated, without reopening a path."""
|
||||
module=types.ModuleType(name);module.__file__=str(path)
|
||||
exec(compile(body,str(path),'exec'),module.__dict__);return module
|
||||
|
||||
def selectors(frame):
|
||||
if frame==0:return [[0,0],[0,31],[31,0],[31,30],[31,31]]
|
||||
if 5<=frame<=9:
|
||||
row=19-3*(frame-5);return [[x,y]for x in range(4,27)for y in range(row,row+3)]
|
||||
return []
|
||||
|
||||
def validate_recipe(r):
|
||||
keys={'format_version','method','parameters','palette','donor','phase0','ancestor_raw','ancestor_recipe','kernel','decoder','frames'}
|
||||
if type(r) is not dict or set(r)!=keys or type(r['format_version']) is not int or r['format_version']!=1 or r['method']!=METHOD:
|
||||
raise ValueError('exact versioned portable complete14 recipe required')
|
||||
if type(r['parameters']) is not dict or set(r['parameters'])!=set(PARAMETERS) or any(type(r['parameters'][k]) is not type(v) or r['parameters'][k]!=v for k,v in PARAMETERS.items()):
|
||||
raise ValueError('fixed V4 pixel parameters cannot be tuned')
|
||||
for key in keys-{'format_version','method','parameters','frames'}:reference(r[key])
|
||||
for key,digest in [('palette',PALETTE_SHA),('donor',DONOR_SHA),('phase0',PHASE0_SHA),('ancestor_raw',ANCESTOR_RAW_SHA),('kernel',KERNEL_SHA),('decoder',DECODER_SHA)]:
|
||||
if r[key]['sha256']!=digest:raise ValueError('fixed genuine input or code changed: '+key)
|
||||
frames=r['frames']
|
||||
if type(frames) is not list or len(frames)!=14:raise ValueError('complete14 frames required')
|
||||
for i,f in enumerate(frames):
|
||||
if type(f) is not dict or set(f)!={'asset','source','accepted803_RGBA','origin','original_detail_cells'} or f['asset']!=f'{0x10152+i:x}':
|
||||
raise ValueError('complete canonical numeric Source order required')
|
||||
reference(f['source']);reference(f['accepted803_RGBA'])
|
||||
if type(f['origin']) is not list or len(f['origin'])!=2 or any(type(x) is not int or x!=0 for x in f['origin']):raise ValueError('literal zero origin required')
|
||||
if f['original_detail_cells']!=selectors(i) or any(type(x) is not int for p in f['original_detail_cells']for x in p):raise ValueError('exact350 authentic runtime retention selectors required')
|
||||
source=frames[0]['source']['path']
|
||||
if [f['source']['path']for f in frames]!=[source+('.'+str(i)if i else '')for i in range(14)]:raise ValueError('Source paths must preserve complete family numbering')
|
||||
return r
|
||||
|
||||
def validate_ancestor(ancestor,frames):
|
||||
if type(ancestor) is not dict or type(ancestor.get('format_version')) is not int or ancestor['format_version']!=1 or ancestor.get('method')!='derived-continuous-generated-hatch-Source-broad-tone-phase-rebuild-v1':raise ValueError('accepted803 ancestor recipe required')
|
||||
if ancestor.get('palette',{}).get('sha256')!=PALETTE_SHA or ancestor.get('raw_png',{}).get('sha256')!=ANCESTOR_RAW_SHA:raise ValueError('ancestor genuine palette/raw identity changed')
|
||||
prior=ancestor.get('frames')
|
||||
if type(prior) is not list or len(prior)!=14:raise ValueError('complete14 ancestor required')
|
||||
for a,b in zip(prior,frames):
|
||||
if a.get('asset')!=b['asset'] or a.get('source',{}).get('sha256')!=b['source']['sha256']:raise ValueError('ancestor own Source identities changed')
|
||||
if type(a.get('origin')) is not list or len(a['origin'])!=2 or any(type(x)is not int or x!=0 for x in a['origin']):raise ValueError('ancestor origin changed')
|
||||
|
||||
def decode_source(decoder,payload):
|
||||
w,h,c=decoder._indices(payload)
|
||||
if (w,h)!=(32,32):raise ValueError("authentic Source32 canvas required")
|
||||
src=np.frombuffer(c,np.uint8).astype(np.uint16) if isinstance(c,bytes) else np.asarray(c,np.uint16)
|
||||
return src.reshape(h,w)
|
||||
|
||||
def run(project,recipe_path,output_path):
|
||||
"""Read only contained inputs and publish a fresh reproduction directory."""
|
||||
project=Path(project).resolve(strict=True);recipe_file=contained(project,recipe_path)
|
||||
output=contained(project,output_path,existing=False)
|
||||
if output.exists() or output.is_symlink():raise ValueError('output exists; no overwrite')
|
||||
if not output.parent.is_dir():raise ValueError('output parent must exist')
|
||||
body=read_regular(recipe_file);r=validate_recipe(json.loads(body));inputs={}
|
||||
def checked(ref):
|
||||
reference(ref);p=contained(project,ref['path']);b=read_regular(p,ref['sha256']);inputs[ref['path']]=ref['sha256'];return b
|
||||
ancestor=json.loads(checked(r['ancestor_recipe']));validate_ancestor(ancestor,r['frames']);checked(r['ancestor_raw'])
|
||||
kb=checked(r['kernel']);db=checked(r['decoder'])
|
||||
kernel=load_verified(kb,contained(project,r['kernel']['path']),'_verified_targt_kernel')
|
||||
decoder=load_verified(db,contained(project,r['decoder']['path']),'_verified_targt_decoder')
|
||||
palette=kernel.palette_rgb(checked(r['palette']))
|
||||
donor=Image.open(io.BytesIO(checked(r['donor'])))
|
||||
phase0body=checked(r['phase0']);phase0=Image.open(io.BytesIO(phase0body))
|
||||
for im in [donor,phase0]:
|
||||
if im.format!='PNG' or im.size!=(1254,1254) or im.convert('RGBA').getchannel('A').getextrema()!=(255,255):raise ValueError('genuine whole opaque1254 PNG required')
|
||||
bands,band_metrics=donor_bands(np.asarray(donor.convert('RGB')),kernel)
|
||||
sources=[];parents=[]
|
||||
for frame in r['frames']:
|
||||
src=decode_source(decoder,checked(frame['source']));sources.append(src)
|
||||
image=Image.open(io.BytesIO(checked(frame['accepted803_RGBA'])))
|
||||
if image.format!='PNG' or image.size!=(256,256):raise ValueError('accepted803 parent256PNG required')
|
||||
parent=np.asarray(image.convert('RGBA'))
|
||||
if not np.array_equal(parent[:,:,3],expand(src<256,8)*255):raise ValueError('accepted803 Source alpha mismatch')
|
||||
parents.append(parent)
|
||||
if not np.all(sources[0]<256):raise ValueError('phase0 must be full opaque own Source')
|
||||
static=np.all(np.stack(sources)[1:]==sources[1],axis=0)&(sources[1]<256)
|
||||
if int(static.sum())!=96:raise ValueError('common96 Source static cells changed')
|
||||
payloads={r['frames'][0]['asset']:phase0body};metrics=[]
|
||||
for frame,src,parent in zip(r['frames'][1:],sources[1:],parents[1:]):
|
||||
out,metric=apply_frame(parent,src,palette,static,bands,kernel);metric['asset']=frame['asset'];metrics.append(metric)
|
||||
stream=io.BytesIO();Image.fromarray(out).save(stream,format='PNG',compress_level=9);payloads[frame['asset']]=stream.getvalue()
|
||||
execution=dict(status='success',method=METHOD,recipe_sha256=sha(body),tool_sha256=sha(read_regular(__file__)),source_sha256={f['asset']:f['source']['sha256']for f in r['frames']},parent_raw_sha256=sorted({ANCESTOR_RAW_SHA,DONOR_SHA,PHASE0_SHA}),output_sha256={a:sha(b)for a,b in payloads.items()})
|
||||
result=dict(execution=execution,parameters=PARAMETERS,phase0=dict(whole_genuine_raw_copied_exact=True,own_Source_used=True,retained_Original_cells=5),frames=metrics,warning_retained_Original_cells=345,retained_Original_cells_not_HD_credit=True,diagnostic_quality_failures_preserved=[dict(asset=m['asset'],failures=m['failures'])for m in metrics if m['failures']],band_metrics=band_metrics,inputs=inputs,verified_dependencies_executed_from_authenticated_bytes=True,runtime=dict(python=platform.python_version(),numpy=np.__version__,pillow=PIL.__version__),artistic_approval=False,registration=False,World=0,SourceF_helper=0,imagegen=0)
|
||||
stage=Path(tempfile.mkdtemp(prefix='.'+output.name+'.stage-',dir=output.parent))
|
||||
try:
|
||||
for asset,b in payloads.items():(stage/(asset+'-derived.png')).write_bytes(b)
|
||||
(stage/'execution-and-derived-provenance.json').write_text(json.dumps(result,indent=2)+'\n')
|
||||
for path,digest in inputs.items():read_regular(contained(project,path),digest)
|
||||
# The user owns this authoring process; mkdir gives portable no-overwrite
|
||||
# publication and a concurrent destination is never replaced.
|
||||
output.mkdir()
|
||||
try:
|
||||
for p in sorted(stage.iterdir()):shutil.move(str(p),str(output/p.name))
|
||||
except BaseException:
|
||||
shutil.rmtree(output);raise
|
||||
finally:shutil.rmtree(stage,ignore_errors=True)
|
||||
return result
|
||||
|
||||
if __name__=='__main__':
|
||||
parser=argparse.ArgumentParser();parser.add_argument('project',type=Path);parser.add_argument('recipe_relative');parser.add_argument('new_output_relative')
|
||||
args=parser.parse_args();print(json.dumps(run(args.project,args.recipe_relative,args.new_output_relative),indent=2))
|
||||
@@ -0,0 +1,327 @@
|
||||
"""Private SHIP15G1 complete3 component registration; not artistic approval.
|
||||
|
||||
Genuine generated RGB is sampled only from the same phase/material. Literal
|
||||
Source support supplies derived alpha; exactly29 declared logical cells supply
|
||||
baked Original RGB. Parent PNG/Source files are never changed. This file is
|
||||
prepared code only until Root releases an actual recipe execution.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import argparse
|
||||
from collections import deque
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import math
|
||||
from pathlib import Path
|
||||
import platform
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image, __version__ as PILLOW_VERSION
|
||||
|
||||
METHOD = 'derived-generated-ship15-component-mast-pods-wake-source-support-v1'
|
||||
DECODER_SHA = '66184a4758027dadbadb5e20fe90b2add2b9fc218b7810304b1a1ea2de74f58e'
|
||||
CAPS = dict(metal_max_logical_distance=1.0, metal_max_missing_fraction=0.13,
|
||||
wake_max_logical_distance=2.75, wake_max_missing_fraction=0.18)
|
||||
AXES = {
|
||||
'main_machinery_mast_hull': ([36, 39, 42, 53], [4, 7, 15, 18]),
|
||||
'left_pod': ([21, 29], [7, 15]),
|
||||
'right_pod': ([55, 63], [7, 15]),
|
||||
'teal_wake': ([0, 21, 29, 36, 53, 55, 63, 72], [2, 4, 7, 15, 18, 20]),
|
||||
}
|
||||
|
||||
|
||||
def sha(data):
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def exact_keys(value, keys, label):
|
||||
if type(value) is not dict or set(value) != set(keys):
|
||||
raise ValueError(f'{label}: unknown/missing fields')
|
||||
|
||||
|
||||
def checked(binding, base):
|
||||
exact_keys(binding, {'path', 'sha256'}, 'input binding')
|
||||
if type(binding['path']) is not str or not binding['path'] or type(binding['sha256']) is not str:
|
||||
raise ValueError('input path/hash types')
|
||||
path = Path(binding['path'])
|
||||
if not path.is_absolute():
|
||||
path = Path(base) / path
|
||||
data = path.read_bytes()
|
||||
if sha(data) != binding['sha256']:
|
||||
raise ValueError(f'input hash mismatch: {path}')
|
||||
return path.resolve(), data
|
||||
|
||||
|
||||
def finite_number(v):
|
||||
return type(v) in (int, float) and math.isfinite(v)
|
||||
|
||||
|
||||
def validate_model(model, name, width, height):
|
||||
exact_keys(model, {'source_x', 'raw_x', 'source_y', 'raw_y', 'donor_bounds'}, name)
|
||||
expected_x, expected_y = AXES[name]
|
||||
for key, expected in [('source_x', expected_x), ('source_y', expected_y)]:
|
||||
if type(model[key]) is not list or model[key] != expected or any(type(v) is not int for v in model[key]):
|
||||
raise ValueError(f'{name}: Source knots differ')
|
||||
for source_key, raw_key, maximum in [('source_x', 'raw_x', width), ('source_y', 'raw_y', height)]:
|
||||
values = model[raw_key]
|
||||
if (type(values) is not list or len(values) != len(model[source_key]) or
|
||||
any(not finite_number(v) for v in values) or
|
||||
not all(0 <= v <= maximum for v in values) or
|
||||
not all(b > a for a, b in zip(values, values[1:]))):
|
||||
raise ValueError(f'{name}: nonfinite/nonmonotone/outside raw knots')
|
||||
bounds = model['donor_bounds']
|
||||
if (type(bounds) is not list or len(bounds) != 4 or any(type(v) is not int for v in bounds) or
|
||||
not (0 <= bounds[0] < bounds[2] <= width and 0 <= bounds[1] < bounds[3] <= height)):
|
||||
raise ValueError(f'{name}: invalid donor bounds')
|
||||
if not (bounds[0] <= model['raw_x'][0] <= model['raw_x'][-1] <= bounds[2] and
|
||||
bounds[1] <= model['raw_y'][0] <= model['raw_y'][-1] <= bounds[3]):
|
||||
raise ValueError(f'{name}: knots escape same-component donor bounds')
|
||||
|
||||
|
||||
def nearest_owners(trusted):
|
||||
"""Raster seeds, N/W/E/S BFS; owners always refer to initial trusted pixels."""
|
||||
if not trusted.any():
|
||||
raise ValueError('no genuine parent donors for component/material')
|
||||
h, w = trusted.shape
|
||||
distance = np.full((h, w), np.iinfo(np.int32).max, np.int32)
|
||||
owner_y = np.full((h, w), -1, np.int32)
|
||||
owner_x = np.full((h, w), -1, np.int32)
|
||||
seeds = np.argwhere(trusted)
|
||||
todo = deque()
|
||||
for y, x in seeds:
|
||||
y, x = int(y), int(x)
|
||||
distance[y, x] = 0
|
||||
owner_y[y, x], owner_x[y, x] = y, x
|
||||
todo.append((x, y))
|
||||
while todo:
|
||||
x, y = todo.popleft()
|
||||
n = int(distance[y, x]) + 1
|
||||
for nx, ny in ((x, y-1), (x-1, y), (x+1, y), (x, y+1)):
|
||||
if 0 <= nx < w and 0 <= ny < h and distance[ny, nx] > n:
|
||||
distance[ny, nx] = n
|
||||
owner_y[ny, nx], owner_x[ny, nx] = owner_y[y, x], owner_x[y, x]
|
||||
todo.append((nx, ny))
|
||||
return distance, owner_y, owner_x
|
||||
|
||||
|
||||
def sample_component(parent, source_mask, model, name, scale, caps):
|
||||
"""Nearest genuine raw RGB at declared inverse-map coordinates, no gains."""
|
||||
h, w = source_mask.shape
|
||||
output_mask = np.repeat(np.repeat(source_mask, scale, axis=0), scale, axis=1)
|
||||
yy, xx = np.nonzero(output_mask)
|
||||
if len(xx) == 0:
|
||||
raise ValueError(f'{name}: empty Source component')
|
||||
u = np.interp((xx+0.5)/scale, model['source_x'], model['raw_x'])
|
||||
v = np.interp((yy+0.5)/scale, model['source_y'], model['raw_y'])
|
||||
px, py = np.floor(u).astype(int), np.floor(v).astype(int)
|
||||
l, t, r, b = model['donor_bounds']
|
||||
if np.any(px < l) or np.any(px >= r) or np.any(py < t) or np.any(py >= b):
|
||||
raise ValueError(f'{name}: mapped sample outside donor bounds')
|
||||
# Donor tests are performed on an untouched crop of this exact phase only.
|
||||
patch = parent[t:b, l:r]
|
||||
rgb = patch[:, :, :3].astype(np.int16)
|
||||
trusted = patch[:, :, 3] >= 128
|
||||
if name == 'teal_wake':
|
||||
trusted &= (rgb[:, :, 1] > rgb[:, :, 0]+10) & (rgb[:, :, 2] > rgb[:, :, 0]+10)
|
||||
kind = 'wake'
|
||||
else:
|
||||
trusted &= rgb.max(axis=2)-rgb.min(axis=2) <= 18
|
||||
kind = 'metal'
|
||||
distance, oy, ox = nearest_owners(trusted)
|
||||
qx, qy = px-l, py-t
|
||||
d = distance[qy, qx].astype(float)
|
||||
direct = trusted[qy, qx]
|
||||
minimum_raw_per_logical = min(
|
||||
min(np.diff(model['raw_x'])/np.diff(model['source_x'])),
|
||||
min(np.diff(model['raw_y'])/np.diff(model['source_y'])))
|
||||
logical_distance = d/minimum_raw_per_logical
|
||||
missing_fraction = float((~direct).mean())
|
||||
if missing_fraction > caps[f'{kind}_max_missing_fraction'] + 1e-12:
|
||||
raise ValueError(f'{name}: genuine donor missing fraction exceeds cap')
|
||||
if float(logical_distance.max()) > caps[f'{kind}_max_logical_distance'] + 1e-12:
|
||||
raise ValueError(f'{name}: genuine donor distance exceeds cap')
|
||||
sy, sx = oy[qy, qx], ox[qy, qx]
|
||||
if not np.all(trusted[sy, sx]):
|
||||
raise AssertionError('owner is not an original genuine donor')
|
||||
# No interpolated/new pixel can become a donor; all reads use the parent.
|
||||
values = patch[sy, sx, :3].copy()
|
||||
return output_mask, values, dict(component=name, Source_cells=int(source_mask.sum()),
|
||||
output_samples=len(xx), direct_genuine_samples=int(direct.sum()),
|
||||
nearest_parent_extension_samples=int((~direct).sum()),
|
||||
missing_fraction=missing_fraction,
|
||||
max_raw_Manhattan_distance=float(d.max()),
|
||||
max_equivalent_logical_distance=float(logical_distance.max()),
|
||||
same_phase_material_and_bounds=True, cascading_fill=False)
|
||||
|
||||
|
||||
def validate_recipe(recipe):
|
||||
exact_keys(recipe, {'format_version', 'label', 'decoder_module', 'palette', 'parent_raw',
|
||||
'output_scale', 'caps', 'frames'}, 'recipe')
|
||||
if type(recipe['format_version']) is not int or recipe['format_version'] != 1:
|
||||
raise ValueError('recipe format version')
|
||||
if type(recipe['label']) is not str or not recipe['label'].strip():
|
||||
raise ValueError('recipe label')
|
||||
if type(recipe['output_scale']) is not int or recipe['output_scale'] != 8:
|
||||
raise ValueError('this complete3 recipe requires declared8x master')
|
||||
exact_keys(recipe['caps'], CAPS, 'caps')
|
||||
if any(not finite_number(recipe['caps'][k]) or recipe['caps'][k] != v for k, v in CAPS.items()):
|
||||
raise ValueError('caps must equal reviewed proposed hard limits')
|
||||
frames = recipe['frames']
|
||||
if type(frames) is not list or len(frames) != 3:
|
||||
raise ValueError('complete3 required')
|
||||
for i, frame in enumerate(frames):
|
||||
exact_keys(frame, {'asset', 'frame', 'source', 'logical_canvas', 'origin', 'raw_crop',
|
||||
'components', 'functional_original_cells'}, 'frame')
|
||||
if (frame['asset'] != f'{0x1030b+i:05x}' or type(frame['frame']) is not int or frame['frame'] != i or
|
||||
frame['logical_canvas'] != [72, 24] or any(type(v) is not int for v in frame['logical_canvas']) or
|
||||
frame['origin'] != [0, 0] or any(type(v) is not int for v in frame['origin'])):
|
||||
raise ValueError('Source family/order/native/origin changed')
|
||||
exact_keys(frame['components'], AXES, 'components')
|
||||
return frames
|
||||
|
||||
|
||||
def render_family(recipe, base):
|
||||
frames = validate_recipe(recipe)
|
||||
decoder_path, decoder_bytes = checked(recipe['decoder_module'], base)
|
||||
if sha(decoder_bytes) != DECODER_SHA:
|
||||
raise ValueError('unreviewed decoder module')
|
||||
decoder_namespace = {'__name__': '_verified_ship15_decoder', '__file__': str(decoder_path)}
|
||||
exec(compile(decoder_bytes, str(decoder_path), 'exec'), decoder_namespace)
|
||||
decode = decoder_namespace['_indices']
|
||||
palette_path, palette_bytes = checked(recipe['palette'], base)
|
||||
if len(palette_bytes) != 768 or max(palette_bytes) > 63:
|
||||
raise ValueError('literal256 six-bit palette required')
|
||||
palette = np.frombuffer(palette_bytes, np.uint8).reshape(256, 3).astype(np.uint16)
|
||||
palette = ((palette << 2) | (palette >> 4)).astype(np.uint8)
|
||||
parent_path, parent_bytes = checked(recipe['parent_raw'], base)
|
||||
if len(parent_bytes) > 32*1024*1024:
|
||||
raise ValueError('parent exceeds32MiB')
|
||||
with Image.open(io.BytesIO(parent_bytes)) as opened:
|
||||
if (opened.format != 'PNG' or opened.width != opened.height or opened.height % 3 or
|
||||
opened.width*opened.height > 4_194_304 or opened.width < 72):
|
||||
raise ValueError('parent must be a bounded square three-full-strip PNG')
|
||||
parent = np.array(opened.convert('RGBA'))
|
||||
if parent[:, :, 3].min() != 0:
|
||||
raise ValueError('parent must have genuine transparent alpha')
|
||||
height, width = parent.shape[:2]
|
||||
strip_height = height//3
|
||||
sources, input_hashes = [], {str(parent_path): sha(parent_bytes), str(palette_path): sha(palette_bytes)}
|
||||
for i, frame in enumerate(frames):
|
||||
crop = frame['raw_crop']
|
||||
if (type(crop) is not list or crop != [0, i*strip_height, width, strip_height] or
|
||||
any(type(v) is not int for v in crop)):
|
||||
raise ValueError('only three literal full strips permitted, no crop fit')
|
||||
path, payload = checked(frame['source'], base)
|
||||
sw, sh, indices = decode(payload)
|
||||
if [sw, sh] != [72, 24]:
|
||||
raise ValueError('Original canvas differs')
|
||||
source = np.asarray(indices, np.uint16).reshape(sh, sw)
|
||||
if not np.all((source < 32) | ((source >= 144) & (source <= 157)) | (source == 255) | (source == 256)):
|
||||
raise ValueError('Source class outside exact declared family partition')
|
||||
if np.any((source[:2] < 256)) or np.any((source[20:] < 256)):
|
||||
raise ValueError('Source painted support outside declared full wake region')
|
||||
if np.argwhere(source == 255).tolist() != [[10, 37], [11, 37]]:
|
||||
raise ValueError('Source255 indicator changed')
|
||||
for name, model in frame['components'].items():
|
||||
validate_model(model, name, width, strip_height)
|
||||
sources.append(source)
|
||||
input_hashes[str(path)] = sha(payload)
|
||||
gray = sources[0] < 32
|
||||
if any(not np.array_equal(s < 32, gray) for s in sources[1:]):
|
||||
raise ValueError('stationary Source gray footprint changed')
|
||||
phase_cells = np.any(np.stack(sources) != sources[0], axis=0) & gray
|
||||
if int(phase_cells.sum()) != 12:
|
||||
raise ValueError('Source phase12 tone coordinates changed')
|
||||
arrays, reports = [], []
|
||||
yy, xx = np.indices((24, 72))
|
||||
gray_regions = dict(main_machinery_mast_hull=(xx >= 36)&(xx < 53)&(yy >= 4)&(yy < 18),
|
||||
left_pod=(xx >= 21)&(xx < 29)&(yy >= 7)&(yy < 15),
|
||||
right_pod=(xx >= 55)&(xx < 63)&(yy >= 7)&(yy < 15))
|
||||
gray_region_union = np.logical_or.reduce(list(gray_regions.values()))
|
||||
if np.any(gray & ~gray_region_union):
|
||||
raise ValueError('gray Source support outside three declared components')
|
||||
for i, (frame, source) in enumerate(zip(frames, sources)):
|
||||
occupied = source < 256
|
||||
functional = (source == 31) | phase_cells | (source == 255)
|
||||
expected = [[int(x), int(y)] for y, x in np.argwhere(functional)]
|
||||
cells = frame['functional_original_cells']
|
||||
if (type(cells) is not list or cells != expected or len(cells) != 29 or
|
||||
any(type(cell) is not list or len(cell) != 2 or any(type(v) is not int for v in cell) for cell in cells)):
|
||||
raise ValueError('exact row-major29 explicit Original cells required')
|
||||
masks = {name: (source < 32)®ion for name, region in gray_regions.items()}
|
||||
masks['teal_wake'] = (source >= 144)&(source <= 157)
|
||||
assigned = np.logical_or.reduce(list(masks.values())) | (source == 255)
|
||||
if not np.array_equal(assigned, occupied):
|
||||
raise ValueError('Source support partition mismatch')
|
||||
rgba = np.zeros((192, 576, 4), np.uint8)
|
||||
parent_phase = parent[i*strip_height:(i+1)*strip_height]
|
||||
counts = []
|
||||
for name, mask in masks.items():
|
||||
outmask, values, record = sample_component(parent_phase, mask, frame['components'][name], name, 8, recipe['caps'])
|
||||
rgba[:, :, :3][outmask] = values
|
||||
counts.append(record)
|
||||
alpha = np.repeat(np.repeat(occupied, 8, axis=0), 8, axis=1)
|
||||
rgba[:, :, 3][alpha] = 255
|
||||
before_functional = rgba.copy()
|
||||
functional8 = np.repeat(np.repeat(functional, 8, axis=0), 8, axis=1)
|
||||
source_rgb8 = np.repeat(np.repeat(palette[np.minimum(source, 255)], 8, axis=0), 8, axis=1)
|
||||
rgba[:, :, :3][functional8] = source_rgb8[functional8]
|
||||
if (not np.array_equal(rgba[:, :, 3] == 255, alpha) or np.any(rgba[~alpha]) or
|
||||
not np.array_equal(rgba[:, :, :3][functional8], source_rgb8[functional8]) or
|
||||
not np.array_equal(rgba[~functional8], before_functional[~functional8])):
|
||||
raise AssertionError('Source alpha/functional/outside-original support invariant failed')
|
||||
arrays.append(rgba)
|
||||
reports.append(dict(asset=frame['asset'], frame=i, logical_canvas=[72, 24], origin=[0, 0],
|
||||
component_donor_metrics=counts, Source_occupied_cells=int(occupied.sum()),
|
||||
baked_Original_cells=cells, baked_Original_RGBA_pixels=29*64,
|
||||
generated_detail_not_claimed_for_baked_Original=True,
|
||||
alpha_contract='Literal SourceNN8 occupied support, not OriginalScaleNx alpha',
|
||||
Source_alpha_and_outside_support_exact=True,
|
||||
functional_Original_RGB_exact=True, outside_29_functional_RGBA_unchanged=True))
|
||||
report = dict(method=METHOD, input_hashes=input_hashes, frames=reports,
|
||||
decoder_module=dict(path=str(decoder_path), sha256=sha(decoder_bytes)),
|
||||
python_version=platform.python_version(), python_implementation=platform.python_implementation(),
|
||||
numpy_version=np.__version__, pillow_version=PILLOW_VERSION,
|
||||
scope='Derived geometry/RGB registration only; no native QA or artistic approval',
|
||||
generated_RGB_never_replaced_by_SourceRGB_outside_declared29=True,
|
||||
parent_and_Source_unchanged=True, output_scale=8)
|
||||
return arrays, report
|
||||
|
||||
|
||||
def run_recipe(recipe_path, output):
|
||||
recipe_path, output = Path(recipe_path), Path(output)
|
||||
if output.exists() or output.is_symlink():
|
||||
raise ValueError('output exists; never overwrite prior artwork')
|
||||
data = recipe_path.read_bytes()
|
||||
arrays, report = render_family(json.loads(data), recipe_path.resolve().parent)
|
||||
files = []
|
||||
for pixels, frame in zip(arrays, report['frames']):
|
||||
buffer = io.BytesIO()
|
||||
Image.fromarray(pixels).save(buffer, format='PNG', compress_level=9)
|
||||
payload = buffer.getvalue()
|
||||
name = frame['asset']+'-derived.png'
|
||||
frame['output'] = dict(file=name, sha256=sha(payload), bytes=len(payload), RGBA_sha256=sha(pixels.tobytes()))
|
||||
files.append((name, payload))
|
||||
report['execution'] = dict(status='success', method=METHOD, recipe_sha256=sha(data),
|
||||
tool_sha256=sha(Path(__file__).read_bytes()),
|
||||
source_sha256={f['asset']: f['source']['sha256'] for f in json.loads(data)['frames']},
|
||||
parent_raw_sha256=sorted({json.loads(data)['parent_raw']['sha256']}),
|
||||
output_sha256={f['asset']: f['output']['sha256'] for f in report['frames']})
|
||||
output.mkdir(parents=True, exist_ok=False)
|
||||
for name, payload in files:
|
||||
(output/name).write_bytes(payload)
|
||||
(output/'derived-provenance.json').write_text(json.dumps(report, indent=2, sort_keys=True)+'\n')
|
||||
return report
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('recipe', type=Path)
|
||||
parser.add_argument('output_new_directory', type=Path)
|
||||
args = parser.parse_args()
|
||||
report = run_recipe(args.recipe, args.output_new_directory)
|
||||
print(json.dumps(dict(execution=report['execution'])))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,261 @@
|
||||
"""Synthetic only: no actual SHIP15 image, Source or recipe is rendered."""
|
||||
from pathlib import Path
|
||||
import copy
|
||||
import importlib.util
|
||||
import json
|
||||
import struct
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
HERE=Path(__file__).resolve().parent
|
||||
spec=importlib.util.spec_from_file_location('private_ship15', HERE/'ship15_component_cleanup.py')
|
||||
tool=importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(tool)
|
||||
DECODER=HERE.parent.parent/'tools/hd_frame_policy.py'
|
||||
TONE=[(38,7,7,6),(42,7,3,1),(43,7,4,3),(46,13,31,30),(47,13,30,28),(48,13,29,27),
|
||||
(49,13,28,26),(50,13,29,27),(51,13,27,24),(42,14,27,25),(43,14,29,27),(44,14,27,25)]
|
||||
|
||||
|
||||
def binding(p):
|
||||
return dict(path=str(p),sha256=tool.sha(p.read_bytes()))
|
||||
|
||||
|
||||
def encode_source(arr):
|
||||
out=bytearray(struct.pack('<5i',0,2,18,72,24))
|
||||
for y in range(24):
|
||||
x=0
|
||||
while x<72:
|
||||
if arr[y,x]==256:
|
||||
x+=1
|
||||
continue
|
||||
end=x+1
|
||||
while end<72 and arr[y,end]<256:end+=1
|
||||
out.extend(struct.pack('<4i',x,y,y*72+x,end-x))
|
||||
out.extend(bytes(arr[y,x:end].tolist()))
|
||||
x=end
|
||||
out.extend(struct.pack('<4i',0,0,-1,0))
|
||||
return bytes(out)
|
||||
|
||||
|
||||
class SyntheticFamily(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp=tempfile.TemporaryDirectory()
|
||||
self.root=Path(self.temp.name)
|
||||
self.palette=np.zeros((256,3),np.uint8)
|
||||
for i in range(32):self.palette[i]=63-2*i
|
||||
self.palette[144:158]=[5,25,30]
|
||||
self.palette[255]=[54,54,54]
|
||||
pp=self.root/'PALETTE';pp.write_bytes(self.palette.tobytes())
|
||||
base=np.full((24,72),256,np.uint16)
|
||||
base[2:20]=144
|
||||
base[7:15,36:53]=10
|
||||
base[4:7,39:42]=10
|
||||
base[15:18,39:42]=10
|
||||
base[7:15,21:29]=10
|
||||
base[7:15,55:63]=10
|
||||
base[10:12,37]=255
|
||||
tone_set={(x,y)for x,y,_,_ in TONE}
|
||||
black=[]
|
||||
for y,x in np.argwhere(base<32):
|
||||
if (int(x),int(y))not in tone_set:
|
||||
black.append((int(x),int(y)))
|
||||
if len(black)==15:break
|
||||
for x,y in black:base[y,x]=31
|
||||
self.sources=[]
|
||||
self.raw=np.zeros((144,144,4),np.uint8)
|
||||
frames=[]
|
||||
for i in range(3):
|
||||
s=base.copy()
|
||||
for x,y,v0,v1 in TONE:s[y,x]=v0 if i==0 else v1
|
||||
self.sources.append(s)
|
||||
p=self.root/f'SOURCE{i}';p.write_bytes(encode_source(s))
|
||||
alpha=np.repeat(np.repeat(s<256,2,axis=0),2,axis=1)
|
||||
gray=np.repeat(np.repeat(s<32,2,axis=0),2,axis=1)
|
||||
yy,xx=np.indices(alpha.shape)
|
||||
phase=np.zeros((48,144,4),np.uint8)
|
||||
phase[:,:,:3][alpha]=np.stack([10+xx%5,90+yy%10,110+xx%10],axis=2)[alpha]
|
||||
neutral=(60+xx%70+yy%11+i*7).astype(np.uint8)
|
||||
for c in range(3):phase[:,:,c][gray]=neutral[gray]
|
||||
phase[:,:,3][alpha]=255
|
||||
self.raw[i*48:(i+1)*48]=phase
|
||||
comps={}
|
||||
for name,(sx,sy)in tool.AXES.items():
|
||||
comps[name]=dict(source_x=sx,raw_x=[v*2 for v in sx],source_y=sy,raw_y=[v*2 for v in sy],
|
||||
donor_bounds=[sx[0]*2,sy[0]*2,sx[-1]*2,sy[-1]*2])
|
||||
frames.append(dict(asset=f'{0x1030b+i:05x}',frame=i,source=binding(p),logical_canvas=[72,24],origin=[0,0],
|
||||
raw_crop=[0,i*48,144,48],components=comps,functional_original_cells=[]))
|
||||
phases=np.any(np.stack(self.sources)!=self.sources[0],axis=0)&(self.sources[0]<32)
|
||||
for f,s in zip(frames,self.sources):
|
||||
f['functional_original_cells']=[[int(x),int(y)]for y,x in np.argwhere((s==31)|phases|(s==255))]
|
||||
self.assertEqual(len(f['functional_original_cells']),29)
|
||||
self.parent=self.root/'parent.png'
|
||||
Image.fromarray(self.raw).save(self.parent)
|
||||
self.recipe=dict(format_version=1,label='SYNTHETIC ONLY',decoder_module=binding(DECODER),palette=binding(pp),
|
||||
parent_raw=binding(self.parent),output_scale=8,caps=copy.deepcopy(tool.CAPS),frames=frames)
|
||||
|
||||
def tearDown(self):
|
||||
self.temp.cleanup()
|
||||
|
||||
def save_parent(self):
|
||||
Image.fromarray(self.raw).save(self.parent)
|
||||
self.recipe['parent_raw']=binding(self.parent)
|
||||
|
||||
def save_source(self,i):
|
||||
p=self.root/f'SOURCE{i}';p.write_bytes(encode_source(self.sources[i]));self.recipe['frames'][i]['source']=binding(p)
|
||||
|
||||
def assert_fails_without_output(self,recipe,pattern):
|
||||
rp=self.root/'bad-recipe.json';rp.write_text(json.dumps(recipe))
|
||||
out=self.root/'MUST-NOT-EXIST'
|
||||
with self.assertRaisesRegex((ValueError,AssertionError),pattern):tool.run_recipe(rp,out)
|
||||
self.assertFalse(out.exists())
|
||||
|
||||
def test_identity_preserves_genuine_RGB_alpha_and29_Original_only(self):
|
||||
arrays,report=tool.render_family(self.recipe,self.root)
|
||||
pal=((self.palette.astype(np.uint16)<<2)|(self.palette.astype(np.uint16)>>4)).astype(np.uint8)
|
||||
for i,(s,out)in enumerate(zip(self.sources,arrays)):
|
||||
alpha=np.repeat(np.repeat(s<256,8,axis=0),8,axis=1)
|
||||
self.assertTrue(np.array_equal(out[:,:,3]==255,alpha))
|
||||
self.assertFalse(out[~alpha].any())
|
||||
yy,xx=np.indices(alpha.shape);expected=self.raw[i*48+(yy//4),xx//4,:3]
|
||||
f=np.zeros((24,72),bool)
|
||||
for x,y in self.recipe['frames'][i]['functional_original_cells']:f[y,x]=True
|
||||
fm=np.repeat(np.repeat(f,8,axis=0),8,axis=1)
|
||||
srgb=np.repeat(np.repeat(pal[np.minimum(s,255)],8,axis=0),8,axis=1)
|
||||
self.assertTrue(np.array_equal(out[:,:,:3][alpha&~fm],expected[alpha&~fm]))
|
||||
self.assertTrue(np.array_equal(out[:,:,:3][fm],srgb[fm]))
|
||||
self.assertEqual(sum(m['nearest_parent_extension_samples']for m in report['frames'][i]['component_donor_metrics']),0)
|
||||
self.assertFalse(np.array_equal(arrays[0],arrays[1]))
|
||||
|
||||
def test_nonuniform_left_pod_transform_keeps_own_genuine_donor(self):
|
||||
for i in range(3):
|
||||
ph=self.raw[i*48:(i+1)*48]
|
||||
original=ph[14:30,42:58].copy()
|
||||
ph[14:30,42:60,:3]=[10,95,115]
|
||||
ph[14:30,42:60,3]=255
|
||||
ph[14:30,44:60]=original
|
||||
m=self.recipe['frames'][i]['components']['left_pod']
|
||||
m['raw_x']=[44,60];m['donor_bounds']=[44,14,60,30]
|
||||
self.save_parent()
|
||||
arrays,_=tool.render_family(self.recipe,self.root)
|
||||
# An ordinary non-functional bottom-right pod point reads translated raw.
|
||||
x,y=28*8+2,14*8+2
|
||||
self.assertNotIn([28,14],self.recipe['frames'][0]['functional_original_cells'])
|
||||
self.assertEqual(arrays[0][y,x,:3].tolist(),self.raw[28,58,:3].tolist())
|
||||
self.assertEqual(int(arrays[0][y,x,3]),255)
|
||||
|
||||
def test_malformed_knots_and_relaxed_caps_fail_before_output(self):
|
||||
mutations=[('raw_x',[72,78,78,106]),('raw_y',[8,float('nan'),30,36]),('source_x',[36,39,42,54]),
|
||||
('raw_x',[True,78,84,106]),('donor_bounds',[73,8,106,36])]
|
||||
for key,value in mutations:
|
||||
with self.subTest(key=key,value=value):
|
||||
r=copy.deepcopy(self.recipe);r['frames'][0]['components']['main_machinery_mast_hull'][key]=value
|
||||
self.assert_fails_without_output(r,'knots|bounds')
|
||||
r=copy.deepcopy(self.recipe);r['caps']['metal_max_logical_distance']=2
|
||||
self.assert_fails_without_output(r,'caps')
|
||||
r=copy.deepcopy(self.recipe);r['caps']['metal_max_logical_distance']=True
|
||||
self.assert_fails_without_output(r,'caps')
|
||||
|
||||
def test_missing_alpha_no_donor_and_wrong_material_fail(self):
|
||||
original=self.raw.copy()
|
||||
self.raw[14:30,42:58,3]=0;self.save_parent()
|
||||
self.assert_fails_without_output(self.recipe,'no genuine')
|
||||
self.raw=original.copy();self.raw[14:30,42:58,:3]=[0,130,150];self.save_parent()
|
||||
self.assert_fails_without_output(self.recipe,'no genuine')
|
||||
self.raw=original.copy();self.raw[:,:,3]=255;self.save_parent()
|
||||
self.assert_fails_without_output(self.recipe,'transparent alpha')
|
||||
|
||||
def test_donor_fraction_and_distance_are_separate_hard_stops(self):
|
||||
original=self.raw.copy()
|
||||
self.raw[14:30,42:46,3]=0;self.save_parent()
|
||||
self.assert_fails_without_output(self.recipe,'missing fraction')
|
||||
self.raw=original.copy();self.raw[18:24,86:92,3]=0;self.save_parent()
|
||||
self.assert_fails_without_output(self.recipe,'distance exceeds')
|
||||
|
||||
def test_Original_selectors_and_Source_support_are_fail_closed(self):
|
||||
for value in [self.recipe['frames'][0]['functional_original_cells'][:-1],
|
||||
self.recipe['frames'][0]['functional_original_cells']+[self.recipe['frames'][0]['functional_original_cells'][0]]]:
|
||||
r=copy.deepcopy(self.recipe);r['frames'][0]['functional_original_cells']=value
|
||||
self.assert_fails_without_output(r,'Original cells')
|
||||
original=self.sources[0].copy()
|
||||
self.sources[0][1,20]=144;self.save_source(0)
|
||||
self.assert_fails_without_output(self.recipe,'support outside')
|
||||
self.sources[0]=original.copy();self.sources[0][10,37]=10;self.save_source(0)
|
||||
self.assert_fails_without_output(self.recipe,'indicator changed')
|
||||
self.sources[0]=original.copy();self.sources[0][5,5]=10;self.save_source(0)
|
||||
self.assert_fails_without_output(self.recipe,'footprint changed|support outside')
|
||||
self.sources[0]=original.copy();self.sources[0][5,5]=100;self.save_source(0)
|
||||
self.assert_fails_without_output(self.recipe,'class outside')
|
||||
|
||||
def test_source_tone12_family_order_and_full_crop_not_optional(self):
|
||||
r=copy.deepcopy(self.recipe);r['frames'][0]['frame']=False
|
||||
self.assert_fails_without_output(r,'order')
|
||||
r=copy.deepcopy(self.recipe);r['frames'][1]['raw_crop']=[0,49,144,47]
|
||||
self.assert_fails_without_output(r,'full strips')
|
||||
for i in (1,2):
|
||||
self.sources[i]=self.sources[0].copy();self.save_source(i)
|
||||
self.assert_fails_without_output(self.recipe,'phase12')
|
||||
|
||||
def test_hash_check_precedes_any_output_and_preserves_parent(self):
|
||||
before={p:tool.sha(p.read_bytes())for p in self.root.iterdir()if p.is_file()}
|
||||
r=copy.deepcopy(self.recipe);r['parent_raw']['sha256']='0'*64
|
||||
self.assert_fails_without_output(r,'hash mismatch')
|
||||
for p,sha in before.items():self.assertEqual(tool.sha(p.read_bytes()),sha)
|
||||
|
||||
def test_nearest_ties_are_raster_seeded_and_never_cascade(self):
|
||||
trust=np.zeros((5,5),bool);trust[0,0]=True;trust[4,4]=True
|
||||
distance,oy,ox=tool.nearest_owners(trust)
|
||||
self.assertTrue(np.all(trust[oy,ox]))
|
||||
self.assertEqual((int(oy[2,2]),int(ox[2,2])),(0,0))
|
||||
self.assertEqual(int(distance[2,2]),4)
|
||||
for y in range(5):
|
||||
for x in range(5):self.assertEqual(int(distance[y,x]),min(x+y,8-x-y))
|
||||
|
||||
def test_verified_decoder_bytes_execute_after_temporary_path_replacement(self):
|
||||
# Only a temporary decoder copy is replaced; no shared input is changed.
|
||||
decoder_copy=self.root/'copied_decoder.py'
|
||||
verified_bytes=DECODER.read_bytes()
|
||||
decoder_copy.write_bytes(verified_bytes)
|
||||
self.recipe['decoder_module']=binding(decoder_copy)
|
||||
expected,_=tool.render_family(self.recipe,self.root)
|
||||
checked_original=tool.checked
|
||||
replaced=[]
|
||||
malicious=b"raise RuntimeError('UNVERIFIED_TEMP_DECODER_MUST_NOT_EXECUTE')\n"
|
||||
|
||||
def verify_then_replace(binding_value,base):
|
||||
path,data=checked_original(binding_value,base)
|
||||
if path==decoder_copy.resolve():
|
||||
self.assertEqual(data,verified_bytes)
|
||||
self.assertEqual(tool.sha(data),tool.DECODER_SHA)
|
||||
path.write_bytes(malicious)
|
||||
replaced.append(path)
|
||||
return path,data
|
||||
|
||||
with patch.object(tool,'checked',side_effect=verify_then_replace):
|
||||
actual,report=tool.render_family(self.recipe,self.root)
|
||||
self.assertEqual(replaced,[decoder_copy.resolve()])
|
||||
self.assertEqual(decoder_copy.read_bytes(),malicious)
|
||||
self.assertEqual(report['decoder_module']['sha256'],tool.DECODER_SHA)
|
||||
for before,after in zip(expected,actual):
|
||||
self.assertTrue(np.array_equal(before,after))
|
||||
self.assertFalse(any(p.name.endswith('-derived.png')for p in self.root.iterdir()))
|
||||
|
||||
def test_deterministic_recipe_receipts_and_new_output_only(self):
|
||||
rp=self.root/'recipe.json';rp.write_text(json.dumps(self.recipe,sort_keys=True))
|
||||
before={p:tool.sha(p.read_bytes())for p in self.root.iterdir()if p.is_file()}
|
||||
first=tool.run_recipe(rp,self.root/'first')
|
||||
second=tool.run_recipe(rp,self.root/'second')
|
||||
self.assertEqual(first,second)
|
||||
self.assertEqual(set(first['execution']),{'status','method','recipe_sha256','tool_sha256','source_sha256','parent_raw_sha256','output_sha256'})
|
||||
self.assertEqual(first['decoder_module']['sha256'],tool.DECODER_SHA)
|
||||
self.assertIn('numpy_version',first)
|
||||
for name in ['1030b-derived.png','1030c-derived.png','1030d-derived.png','derived-provenance.json']:
|
||||
self.assertEqual((self.root/'first'/name).read_bytes(),(self.root/'second'/name).read_bytes())
|
||||
with self.assertRaisesRegex(ValueError,'output exists'):tool.run_recipe(rp,self.root/'first')
|
||||
for p,sha in before.items():self.assertEqual(tool.sha(p.read_bytes()),sha)
|
||||
|
||||
|
||||
if __name__=='__main__':unittest.main()
|
||||
@@ -0,0 +1,203 @@
|
||||
"""Focused synthetic-only tests. Never load production input artwork or QA."""
|
||||
import copy
|
||||
from fractions import Fraction
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import struct
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
Q=Path(__file__).resolve().parent
|
||||
spec=importlib.util.spec_from_file_location('stag03_private_test',Q/'hd_stag03_cleanup.py')
|
||||
c=importlib.util.module_from_spec(spec);spec.loader.exec_module(c)
|
||||
|
||||
class AffineCleanup(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
|
||||
self.root=Path(self.tmp.name)
|
||||
self.source=np.full((32,32),256,np.uint16);self.source[5:30,1:30]=10
|
||||
for x in list(range(1,11))+list(range(20,30)):self.source[14,x]=[85,28,82,28,80][(x-1)%5]
|
||||
self.write_source()
|
||||
self.palette_bytes=bytes((np.arange(768)%64).astype(np.uint8));(self.root/'palette').write_bytes(self.palette_bytes)
|
||||
y,x=np.indices((160,160));rgba=np.zeros((160,160,4),np.uint8)
|
||||
rgba[:,:,:3]=np.stack([(x*3+y)%255,(x+y*2)%255,(x*7+y*5)%255],axis=-1)
|
||||
rgba[5:150,2:150,3]=255;rgba[rgba[:,:,3]==0,:3]=0
|
||||
Image.fromarray(rgba).save(self.root/'parent.png')
|
||||
self.dep_names={'decoder':'hd_frame_policy.py','geometry':'hd_dshp20_cleanup.py','functional':'hd_other_functional_cleanup.py'}
|
||||
(self.root/'deps').mkdir()
|
||||
for name in self.dep_names.values():shutil.copyfile(Q/name,self.root/'deps'/name)
|
||||
self.recipe=dict(format_version=1,method=c.METHOD,asset='300ae',source=self.bind('Source.pic'),
|
||||
parent_raw=self.bind('parent.png'),palette=self.bind('palette'),logical_canvas=[32,32],origin=[0,0],output_scale=4,
|
||||
alignment=dict(raw_painted_bounds=[2,5,150,150],source_painted_bounds=[1,5,30,30],
|
||||
inverse_affine=['37/29','0','-90/29','0','29/20','-24']),
|
||||
max_fill_distance_pixels=12,functional_groups=copy.deepcopy(c.GROUPS),
|
||||
dependencies={k:self.bind('deps/'+name)for k,name in self.dep_names.items()},expected_tool_sha256=c.sha(c.TOOL_BYTES))
|
||||
|
||||
def write_source(self):
|
||||
b=bytearray(struct.pack('<5i',0,1,31,32,32))
|
||||
for y in range(32):
|
||||
x=0
|
||||
while x<32:
|
||||
if self.source[y,x]>=256:x+=1;continue
|
||||
start=x
|
||||
while x<32 and self.source[y,x]<256:x+=1
|
||||
b.extend(struct.pack('<4i',start,y,y*32+start,x-start));b.extend(bytes(self.source[y,start:x].tolist()))
|
||||
b.extend(struct.pack('<4i',0,0,-1,0));(self.root/'Source.pic').write_bytes(b)
|
||||
|
||||
def bind(self,name):return dict(path=name,sha256=c.sha((self.root/name).read_bytes()))
|
||||
def rp(self,recipe=None):
|
||||
path=self.root/'recipe.json';path.write_text(json.dumps(self.recipe if recipe is None else recipe));return path
|
||||
def modules(self):return c.verified_modules({k:(Q/n).read_bytes()for k,n in self.dep_names.items()})
|
||||
def assert_no_output(self):
|
||||
self.assertFalse((self.root/'derived').exists())
|
||||
self.assertFalse(list(self.root.glob('.stag03-derived-*')))
|
||||
|
||||
def test_valid_fullcanvas_negative_virtual_coordinates_and_receipt(self):
|
||||
rp=self.rp();before={p:p.read_bytes()for p in self.root.rglob('*')if p.is_file()}
|
||||
report=c.run_recipe(rp,self.root/'derived')
|
||||
self.assertEqual(report['actual_functional_calls'],2)
|
||||
self.assertEqual(report['functional_Original_cells'],66)
|
||||
self.assertEqual(report['actual_Original_Scale4_RGBA_samples'],1056)
|
||||
self.assertEqual(report['execution']['source_sha256'],{'300ae':self.bind('Source.pic')['sha256']})
|
||||
self.assertEqual(report['authenticated_dependency_bytes'],c.PINS)
|
||||
self.assertTrue(report['alpha_exact_SourceScale4'])
|
||||
self.assertLess(float(Fraction(report['fullcanvas_inverse_affine'][2])),0)
|
||||
out=np.asarray(Image.open(self.root/'derived/300ae-derived.png'))
|
||||
_,g,_=self.modules();s4=g.scale2(g.scale2(self.source))
|
||||
self.assertTrue(np.array_equal(out[:,:,3]>0,s4<256))
|
||||
self.assertTrue(np.all(out[s4==256,:3]==0))
|
||||
self.assertEqual(c.sha((self.root/'derived/300ae-derived.png').read_bytes()),report['execution']['output_sha256']['300ae'])
|
||||
for p,b in before.items():self.assertEqual(p.read_bytes(),b)
|
||||
with self.assertRaisesRegex(ValueError,'fresh absent'):c.run_recipe(rp,self.root/'derived')
|
||||
|
||||
def test_bad_nonfinite_and_malformed_affines_fail_without_outputs(self):
|
||||
cases=[]
|
||||
for values in ([float('nan'),0,0,0,1,0],['nan','0','0','0','1','0'],['1/0','0','0','0','1','0'],
|
||||
['1e100','0','0','0','1','0'],[True,0,0,0,1,0],['-1','0','0','0','1','0'],
|
||||
['37/29','1','-90/29','0','29/20','-24'],['1']):
|
||||
r=copy.deepcopy(self.recipe);r['alignment']['inverse_affine']=values;cases.append(r)
|
||||
for index,r in enumerate(cases):
|
||||
with self.subTest(case=index),self.assertRaises(ValueError):c.run_recipe(self.rp(r),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
|
||||
def test_bounds_must_match_complete_source_and_parent(self):
|
||||
for key,value in [('raw_painted_bounds',[-1,5,150,150]),('raw_painted_bounds',[2,5,161,150]),
|
||||
('raw_painted_bounds',[2,5,2,150]),('raw_painted_bounds',[2,6,150,150]),
|
||||
('source_painted_bounds',[1,4,30,30]),('source_painted_bounds',[True,5,30,30])]:
|
||||
r=copy.deepcopy(self.recipe);r['alignment'][key]=value
|
||||
with self.subTest(key=key,value=value),self.assertRaises(ValueError):c.run_recipe(self.rp(r),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
|
||||
def test_exact_functional_sets_reject_extra_missing_bool_or_duplicate_cells(self):
|
||||
for kind in ('extra','missing','bool','duplicate'):
|
||||
r=copy.deepcopy(self.recipe)
|
||||
if kind=='extra':r['functional_groups'][0].append([12,14])
|
||||
if kind=='missing':r['functional_groups'][0].pop()
|
||||
if kind=='bool':r['functional_groups'][0][0]=[True,13]
|
||||
if kind=='duplicate':r['functional_groups'][0][1]=r['functional_groups'][0][0]
|
||||
with self.subTest(kind=kind),self.assertRaises(ValueError):c.run_recipe(self.rp(r),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
|
||||
def test_selected_cells_must_have_authentic_source_support(self):
|
||||
self.source[13,1]=256;self.write_source();self.recipe['source']=self.bind('Source.pic')
|
||||
with self.assertRaisesRegex(ValueError,'painted Source'):c.run_recipe(self.rp(),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
|
||||
def test_dependency_bytes_are_verified_then_isolated_not_imported_by_path(self):
|
||||
before=list(sys.path);a=self.modules();b=self.modules()
|
||||
self.assertEqual(sys.path,before)
|
||||
self.assertIsNot(a[0],b[0]);self.assertIsNot(a[1],b[1]);self.assertIsNot(a[2],b[2])
|
||||
self.assertIs(a[2].base,a[1]);self.assertIs(a[1].hd_frame_policy,a[0])
|
||||
payloads={k:(Q/n).read_bytes()for k,n in self.dep_names.items()};payloads['geometry']+=b'\n'
|
||||
with self.assertRaisesRegex(ValueError,'dependency bytes'):c.verified_modules(payloads)
|
||||
|
||||
def test_processing_never_reopens_dependency_or_input_paths(self):
|
||||
rp=self.rp();actual_render=c.render
|
||||
def render_buffered(*args):
|
||||
# Every payload already passed its declared hash. Path-reopening
|
||||
# would fail, including dependency SourceFileLoader habits.
|
||||
with patch.object(Path,'read_bytes',side_effect=AssertionError('processing reopened a path')):
|
||||
return actual_render(*args)
|
||||
with patch.object(c,'render',render_buffered):c.run_recipe(rp,self.root/'derived')
|
||||
self.assertTrue((self.root/'derived/300ae-derived.png').is_file())
|
||||
|
||||
def test_buffered_verified_dependency_survives_later_path_change(self):
|
||||
rp=self.rp();actual=c.safe_read
|
||||
def read_then_change(base,binding,limit):
|
||||
result=actual(base,binding,limit)
|
||||
if binding['path']=='deps/hd_other_functional_cleanup.py':
|
||||
(base/binding['path']).write_text('raise RuntimeError("unverified later path")')
|
||||
return result
|
||||
with patch.object(c,'safe_read',read_then_change):report=c.run_recipe(rp,self.root/'derived')
|
||||
self.assertEqual(report['authenticated_dependency_bytes']['functional'],c.PINS['functional'])
|
||||
|
||||
def test_no_parent_absolute_backslash_or_symlink_input_escape(self):
|
||||
for value in ('../Source.pic','/absolute','deps/../Source.pic','Source\\pic'):
|
||||
r=copy.deepcopy(self.recipe);r['source']['path']=value
|
||||
with self.subTest(value=value),self.assertRaises(ValueError):c.run_recipe(self.rp(r),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
(self.root/'Source-link').symlink_to(self.root/'Source.pic')
|
||||
r=copy.deepcopy(self.recipe);r['source']['path']='Source-link'
|
||||
with self.assertRaises(OSError):c.run_recipe(self.rp(r),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
(self.root/'outside-dir').symlink_to(self.root/'deps',target_is_directory=True)
|
||||
r=copy.deepcopy(self.recipe);r['dependencies']['decoder']['path']='outside-dir/hd_frame_policy.py'
|
||||
with self.assertRaises(OSError):c.run_recipe(self.rp(r),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
|
||||
def test_hash_failure_and_tool_mismatch_stop_before_output(self):
|
||||
for key in ('source','palette','parent_raw'):
|
||||
r=copy.deepcopy(self.recipe);r[key]['sha256']='0'*64
|
||||
with self.subTest(key=key),self.assertRaisesRegex(ValueError,'SHA256 mismatch'):c.run_recipe(self.rp(r),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
r=copy.deepcopy(self.recipe);r['expected_tool_sha256']='0'*64
|
||||
with self.assertRaisesRegex(ValueError,'tool bytes'):c.run_recipe(self.rp(r),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
|
||||
def test_twelve_pixel_generated_fill_never_cascades(self):
|
||||
_,g,_=self.modules();rgb=np.zeros((1,27,3),np.uint8);rgb[0,0]=[3,4,5]
|
||||
trusted=np.zeros((1,27),bool);trusted[0,0]=True;mask=np.ones_like(trusted);s=np.zeros((1,27),np.uint16)
|
||||
with self.assertRaisesRegex(ValueError,'donor'):g.fill_generated_rgb(rgb,trusted,mask,s,12)
|
||||
self.assertEqual(rgb[0,1].tolist(),[0,0,0])
|
||||
|
||||
def test_bad_support_needs_generated_donor_and_cannot_be_sourcenn_relabel(self):
|
||||
a=np.zeros((160,160,4),np.uint8);a[5:150,2:150]=[100,100,100,255]
|
||||
Image.fromarray(a).save(self.root/'parent.png');self.recipe['parent_raw']=self.bind('parent.png')
|
||||
with self.assertRaisesRegex(ValueError,'interior detail'):c.run_recipe(self.rp(),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
|
||||
def test_write_failure_leaves_no_outputs_and_no_false_receipt(self):
|
||||
rp=self.rp();before={p:p.read_bytes()for p in self.root.rglob('*')if p.is_file()}
|
||||
with patch.object(c.os,'fsync',side_effect=OSError('synthetic write failure')):
|
||||
with self.assertRaises(OSError):c.run_recipe(rp,self.root/'derived')
|
||||
self.assert_no_output()
|
||||
for p,b in before.items():self.assertEqual(p.read_bytes(),b)
|
||||
|
||||
def test_publication_race_does_not_replace_existing_even_empty_directory(self):
|
||||
rp=self.rp();actual=c.publish_no_replace
|
||||
def race(staging,output):output.mkdir();return actual(staging,output)
|
||||
with patch.object(c,'publish_no_replace',race),self.assertRaises(FileExistsError):c.run_recipe(rp,self.root/'derived')
|
||||
self.assertEqual(list((self.root/'derived').iterdir()),[])
|
||||
self.assertFalse(list(self.root.glob('.stag03-derived-*')))
|
||||
|
||||
def test_output_cannot_escape_recipe_root(self):
|
||||
with tempfile.TemporaryDirectory()as other:
|
||||
with self.assertRaisesRegex(ValueError,'direct child'):c.run_recipe(self.rp(),Path(other)/'derived')
|
||||
self.assertFalse((Path(other)/'derived').exists())
|
||||
self.assert_no_output()
|
||||
|
||||
def test_free_floor_stops_before_dependency_or_artwork_reads(self):
|
||||
with patch.object(c.shutil,'disk_usage',return_value=types.SimpleNamespace(free=3_000_000_000)):
|
||||
with patch.object(c,'safe_read',side_effect=AssertionError('input read before floor guard')):
|
||||
with self.assertRaisesRegex(ValueError,'free floor'):c.run_recipe(self.rp(),self.root/'derived')
|
||||
self.assert_no_output()
|
||||
|
||||
if __name__=='__main__':unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Synthetic input containment, unchanged material, alpha and execution tests."""
|
||||
from pathlib import Path
|
||||
import copy,hashlib,struct,tempfile,types,unittest
|
||||
import numpy as np
|
||||
import hd_targt_soil_retexture as h
|
||||
import hd_terrain_retexture as kernel
|
||||
import hd_frame_policy as decoder
|
||||
|
||||
def recipe():
|
||||
ref=lambda name,digest='0'*64:dict(path=name,sha256=digest)
|
||||
r=dict(format_version=1,method=h.METHOD,parameters=copy.deepcopy(h.PARAMETERS),palette=ref('palette',h.PALETTE_SHA),donor=ref('donor.png',h.DONOR_SHA),phase0=ref('phase0.png',h.PHASE0_SHA),ancestor_raw=ref('ancestor.png',h.ANCESTOR_RAW_SHA),ancestor_recipe=ref('ancestor.json'),kernel=ref('tools/kernel.py',h.KERNEL_SHA),decoder=ref('tools/decoder.py',h.DECODER_SHA),frames=[])
|
||||
for i in range(14):r['frames'].append(dict(asset=f'{0x10152+i:x}',source=ref('sources/TARGT_PIC'+('.'+str(i)if i else '')),accepted803_RGBA=ref(f'parents/{i}.png'),origin=[0,0],original_detail_cells=h.selectors(i)))
|
||||
return r
|
||||
|
||||
class PortableInputTests(unittest.TestCase):
|
||||
def test_recipe_is_complete_and_runtime_retention_totals350(self):
|
||||
r=recipe();self.assertIs(h.validate_recipe(r),r);self.assertEqual(sum(len(f['original_detail_cells'])for f in r['frames']),350)
|
||||
def test_reject_asset_reordering_and_short_family(self):
|
||||
for mode in ['swap','short']:
|
||||
r=recipe()
|
||||
if mode=='swap':r['frames'][0],r['frames'][1]=r['frames'][1],r['frames'][0]
|
||||
else:r['frames'].pop()
|
||||
with self.assertRaises(ValueError):h.validate_recipe(r)
|
||||
def test_literal_parameter_types_and_unchanged_gains(self):
|
||||
for value in [1,True,.51]:
|
||||
r=recipe();r['parameters']['Source_macro_sigma_logical']=value
|
||||
with self.assertRaises(ValueError):h.validate_recipe(r)
|
||||
def test_runtime_retention_cannot_shift_or_grow(self):
|
||||
for i in [0,5,6,7,8,9,10]:
|
||||
r=recipe();r['frames'][i]['original_detail_cells'].append([1,1])
|
||||
with self.assertRaises(ValueError):h.validate_recipe(r)
|
||||
def test_boolean_origin_is_not_zero_coordinate(self):
|
||||
r=recipe();r['frames'][0]['origin']=[False,0]
|
||||
with self.assertRaises(ValueError):h.validate_recipe(r)
|
||||
def test_source_numbering_is_literal(self):
|
||||
r=recipe();r['frames'][2]['source']['path']='sources/another.2'
|
||||
with self.assertRaises(ValueError):h.validate_recipe(r)
|
||||
def test_fixed_raws_and_verified_dependencies_cannot_be_rebound(self):
|
||||
for name in ['palette','donor','phase0','ancestor_raw','kernel','decoder']:
|
||||
r=recipe();r[name]['sha256']='0'*64
|
||||
with self.assertRaises(ValueError):h.validate_recipe(r)
|
||||
def test_reject_nonportable_or_escaping_path(self):
|
||||
for p in ['/tmp/a','../a','a/../b','a//b','a/./b','C:/x','a\\b','', '.']:
|
||||
with self.assertRaises(ValueError,msg=p):h.relative(p)
|
||||
def test_exact_reference_shape_and_hash(self):
|
||||
for ref in [dict(path='x',sha256='A'*64),dict(path='x',sha256='0'*64,extra=True),dict(path='../x',sha256='0'*64)]:
|
||||
with self.assertRaises(ValueError):h.reference(ref)
|
||||
def test_project_symlink_file_and_directory_rejected(self):
|
||||
with tempfile.TemporaryDirectory()as td:
|
||||
p=Path(td);(p/'real').mkdir();(p/'real/x').write_text('hi');(p/'alias').symlink_to(p/'real',target_is_directory=True);(p/'file').symlink_to(p/'real/x')
|
||||
for name in ['alias/x','file']:
|
||||
with self.assertRaises(ValueError):h.contained(p,name)
|
||||
def test_read_hash_is_checked_on_same_regular_bytes(self):
|
||||
with tempfile.TemporaryDirectory()as td:
|
||||
p=Path(td)/'x';p.write_bytes(b'checked');self.assertEqual(h.read_regular(p,hashlib.sha256(b'checked').hexdigest()),b'checked')
|
||||
with self.assertRaises(ValueError):h.read_regular(p,'0'*64)
|
||||
def test_verified_module_executes_supplied_bytes_not_reopened_path(self):
|
||||
with tempfile.TemporaryDirectory()as td:
|
||||
p=Path(td)/'dep.py';p.write_text('value=1\n');verified=h.read_regular(p);p.write_text('value=999\n')
|
||||
self.assertEqual(h.load_verified(verified,p,'test_verified').value,1)
|
||||
def test_decoder_opaque_bytes_and_sparse_transparency(self):
|
||||
raw=struct.pack('<iiiii',1,0,0,32,32)+bytes([197])*1024
|
||||
self.assertTrue(np.all(h.decode_source(decoder,raw)==197))
|
||||
sparse=struct.pack('<iiiii',0,0,0,32,32)+struct.pack('<iiii',2,3,0,2)+bytes([80,82])+struct.pack('<iiii',0,0,-1,0)
|
||||
s=h.decode_source(decoder,sparse);self.assertEqual(int((s<256).sum()),2);self.assertEqual(s[3,2],80)
|
||||
def test_existing_output_is_never_replaced(self):
|
||||
with tempfile.TemporaryDirectory()as td:
|
||||
p=Path(td);(p/'recipe').write_text('{}');(p/'output').mkdir();(p/'output/keep').write_bytes(b'exact')
|
||||
with self.assertRaises(ValueError):h.run(p,'recipe','output')
|
||||
self.assertEqual((p/'output/keep').read_bytes(),b'exact')
|
||||
|
||||
class SyntheticMaterialTests(unittest.TestCase):
|
||||
def setup_material(self):
|
||||
source=np.full((32,32),197,np.uint16);source[:2]=256;source[10:14,10:14]=10
|
||||
palette=np.zeros((256,3),np.uint8);palette[197]=[103,55,46];palette[10]=[18,18,18]
|
||||
parent=np.zeros((256,256,4),np.uint8);alpha=h.expand(source<256,8);parent[alpha,:3]=[34,28,20];parent[alpha,3]=255
|
||||
static=np.zeros((32,32),bool);static[2,:]=True
|
||||
yy,xx=np.indices((256,256));fine=np.sin(xx/3.)*.75+np.cos(yy/5.)*.5
|
||||
return source,palette,parent,static,[fine]
|
||||
def test_retexture_preserves_every_protected_RGBA_and_alpha(self):
|
||||
s,p,a,static,bands=self.setup_material();out,metrics=h.apply_frame(a,s,p,static,bands,kernel)
|
||||
mutable=h.expand((s>=194)&(s<=239)&~static,8)
|
||||
np.testing.assert_array_equal(out[~mutable],a[~mutable]);np.testing.assert_array_equal(out[:,:,3],a[:,:,3]);self.assertTrue(metrics['protected_RGBA_exact']);self.assertGreater(metrics['changed_soil_pixels'],0)
|
||||
def test_retexture_is_deterministic_and_source_group_mean_is_stable(self):
|
||||
s,p,a,static,bands=self.setup_material();one,m=h.apply_frame(a,s,p,static,bands,kernel);two,_=h.apply_frame(a,s,p,static,bands,kernel)
|
||||
np.testing.assert_array_equal(one,two);self.assertLess(max(abs(x-y)for x,y in zip(m['Source_RGB_mean'],m['actual_RGB_mean'])),1.)
|
||||
def test_nonuniform_source_pattern_runs_finite_correlation_branch(self):
|
||||
s,p,a,static,bands=self.setup_material();yy,xx=np.indices(s.shape)
|
||||
s[(s==197)&((xx//4+yy//4)%2==0)]=198;p[198]=[115,67,58]
|
||||
out,m=h.apply_frame(a,s,p,static,bands,kernel)
|
||||
self.assertGreater(m['Source_luma_std'],.05)
|
||||
self.assertTrue(np.isfinite(m['Source_pattern_correlation']))
|
||||
self.assertGreater(m['Source_pattern_correlation'],.90)
|
||||
self.assertTrue(m['protected_RGBA_exact'])
|
||||
def test_no_soil_returns_exact_parent(self):
|
||||
s,p,a,static,bands=self.setup_material();s[s==197]=10;a[:,:,3]=h.expand(s<256,8)*255
|
||||
out,m=h.apply_frame(a,s,p,static,bands,kernel);np.testing.assert_array_equal(out,a);self.assertEqual(m['mutable_cells'],0)
|
||||
def test_wrong_alpha_or_nonfinite_fine_is_rejected(self):
|
||||
s,p,a,static,bands=self.setup_material();bad=a.copy();bad[0,0,3]=255
|
||||
with self.assertRaises(ValueError):h.apply_frame(bad,s,p,static,bands,kernel)
|
||||
bands[0][0,0]=np.nan
|
||||
with self.assertRaises(ValueError):h.apply_frame(a,s,p,static,bands,kernel)
|
||||
|
||||
if __name__=='__main__':unittest.main()
|
||||
Reference in new issue
Block a user