Complete eleven reviewed original HD remasters
This commit is contained in:
raptorreplica committed
1 parent
83a7c8cb54
commit
fad806a237
2440 files changed
+808909
-43
No files matched your search
@@ -0,0 +1,171 @@
|
||||
"""Private Source-shaped genuine-RGB derivation proposal. Never a generator return.
|
||||
The single32x32 opaque format is decoded here honestly, including protected labels;
|
||||
the ordinary-only decoder in hd_terrain_cleanup is neither called nor bypassed.
|
||||
"""
|
||||
from pathlib import Path
|
||||
import argparse, hashlib, json, sys, struct, os, stat, io
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
ROOT=next((p for p in Path(__file__).resolve().parents if (p/'tools/hd_terrain_cleanup.py').is_file()),None)
|
||||
if ROOT is None: raise RuntimeError('place this portable helper under the game root containing tools/hd_terrain_cleanup.py')
|
||||
MAX_INPUT_BYTES=16*1024*1024
|
||||
MAX_RECIPE_BYTES=64*1024
|
||||
MAX_PNG_PIXELS=4*1024*1024
|
||||
|
||||
def read_contained(root,relative,limit=MAX_INPUT_BYTES):
|
||||
if not isinstance(relative,str) or relative.startswith('/') or any(x in ('','.','..') for x in relative.split('/')):
|
||||
raise ValueError('strict contained relative path required')
|
||||
root=Path(root).absolute()
|
||||
if '..' in root.parts:raise ValueError('noncanonical root path')
|
||||
flags=os.O_RDONLY|os.O_CLOEXEC|os.O_NOFOLLOW|os.O_NONBLOCK
|
||||
fd=os.open('/',flags|os.O_DIRECTORY)
|
||||
try:
|
||||
for part in root.parts[1:]+tuple(relative.split('/')[:-1]):
|
||||
following=os.open(part,flags|os.O_DIRECTORY,dir_fd=fd);os.close(fd);fd=following
|
||||
source=os.open(relative.split('/')[-1],flags,dir_fd=fd)
|
||||
try:
|
||||
before=os.fstat(source)
|
||||
if not stat.S_ISREG(before.st_mode) or not 0<=before.st_size<=limit:raise ValueError('regular bounded input required')
|
||||
chunks=[];size=0
|
||||
while True:
|
||||
chunk=os.read(source,min(65536,limit+1-size))
|
||||
if not chunk:break
|
||||
chunks.append(chunk);size+=len(chunk)
|
||||
if size>limit:raise ValueError('bounded EOF limit exceeded')
|
||||
after=os.fstat(source)
|
||||
identity=lambda x:(x.st_dev,x.st_ino,x.st_size,x.st_mtime_ns,x.st_mode)
|
||||
if identity(before)!=identity(after) or size!=before.st_size:raise ValueError('input changed or non-EOF read')
|
||||
payload=b''.join(chunks)
|
||||
return payload,dict(path=str(root/relative),sha256=hashlib.sha256(payload).hexdigest(),size=size,mtime_ns=before.st_mtime_ns,mode=before.st_mode)
|
||||
finally:os.close(source)
|
||||
finally:os.close(fd)
|
||||
|
||||
class BufferedInputs:
|
||||
def __init__(self,root):self.root=Path(root);self.cache={};self.physical_reads=0
|
||||
def read(self,binding):
|
||||
rel=binding['path']
|
||||
if rel not in self.cache:
|
||||
self.cache[rel]=read_contained(self.root,rel);self.physical_reads+=1
|
||||
payload,pin=self.cache[rel]
|
||||
if pin['sha256']!=binding['sha256']:raise ValueError('input SHA mismatch')
|
||||
return payload,pin
|
||||
|
||||
def decode_png(payload):
|
||||
if len(payload)>MAX_INPUT_BYTES:raise ValueError('encoded PNG too large')
|
||||
with Image.open(io.BytesIO(payload)) as im:
|
||||
if im.format!='PNG' or im.width<1 or im.height<1 or im.width*im.height>MAX_PNG_PIXELS:raise ValueError('bounded PNG required')
|
||||
if getattr(im,'n_frames',1)!=1 or im.mode not in ('RGB','RGBA'):raise ValueError('single RGB/RGBA PNG required')
|
||||
if im.mode=='RGBA' and im.getchannel('A').getextrema()!=(255,255):raise ValueError('opaque PNG donor required')
|
||||
return np.array(im.convert('RGB'))
|
||||
|
||||
EXPECTED_DEPENDENCY_SHA256='3284cf6f85ced107dfc364ea90e15aed341d478bea7e308024b5bc5218b03092'
|
||||
_dependency_bytes,_dependency_pin=read_contained(ROOT/'tools','hd_terrain_cleanup.py',1024*1024)
|
||||
if _dependency_pin['sha256']!=EXPECTED_DEPENDENCY_SHA256:raise RuntimeError('reviewed terrain math dependency changed')
|
||||
_dependency_namespace={'__name__':'reviewed_component_terrain_math','__file__':str(ROOT/'tools/hd_terrain_cleanup.py')}
|
||||
exec(compile(_dependency_bytes,_dependency_namespace['__file__'],'exec'),_dependency_namespace)
|
||||
scale_nx,gaussian,resize_float,palette_rgb=(_dependency_namespace[x]for x in ('scale_nx','gaussian','resize_float','palette_rgb'))
|
||||
|
||||
PAIR_IDS=('103d5','103d6')
|
||||
METHOD='Source-shape-component-material-RGB-band-and-finite-T-landmark-warp-v1'
|
||||
|
||||
def decode_source32(payload):
|
||||
if len(payload)!=1044 or struct.unpack_from('<5i',payload)!=(1,1,0,32,32):
|
||||
raise ValueError('requires exact opaque single32x32 Source PIC')
|
||||
return np.frombuffer(payload[20:],np.uint8).reshape(32,32).copy()
|
||||
|
||||
def material_labels(indices,approach_rows):
|
||||
a=np.asarray(indices)
|
||||
if a.ndim!=2 or a.dtype!=np.uint8 or not 0<=approach_rows<=a.shape[0]:
|
||||
raise ValueError('requires an indexed plane and bounded approach rows')
|
||||
out=np.full(a.shape,-1,np.int8)
|
||||
out[((a>=149)&(a<=159))|((a>=240)&(a<=244))]=0 # water and actual water cycling only
|
||||
out[(a<32)|(a==173)]=1 # grey rail/shadow
|
||||
out[((a>=182)&(a<=190))|((a>=192)&(a<=218))]=2 # red bank / edge earth
|
||||
out[(a>=219)&(a<=238)]=3 # deck; upper approach separately registered
|
||||
if np.any(out<0): raise ValueError('unsupported Source material or protected class')
|
||||
out[:approach_rows][out[:approach_rows]==3]=4
|
||||
return out
|
||||
|
||||
def donor_field(rgb,box,size,tile_size=None):
|
||||
if len(box)!=4 or any(type(v)is not int for v in box):raise ValueError('integer donor box required')
|
||||
x,y,w,h=box
|
||||
if min(x,y)<0 or min(w,h)<1 or x+w>rgb.shape[1]or y+h>rgb.shape[0]:raise ValueError('donor crop outside genuine image')
|
||||
im=Image.fromarray(rgb[y:y+h,x:x+w])
|
||||
if tile_size:
|
||||
sw,sh=tile_size
|
||||
if min(sw,sh)<1:raise ValueError('bad texture period')
|
||||
tile=np.array(im.resize((sw,sh),Image.Resampling.LANCZOS))
|
||||
return np.tile(tile,(int(np.ceil(size[1]/sh)),int(np.ceil(size[0]/sw)),1))[:size[1],:size[0]].astype(float)
|
||||
return np.array(im.resize(size,Image.Resampling.LANCZOS),dtype=float)
|
||||
|
||||
def derive_pair(sources,palette,fields,audit=None):
|
||||
if len(sources)!=2 or set(fields)!=set(range(5)):raise ValueError('complete pair and all five donor components required')
|
||||
guide=np.concatenate([scale_nx(x,4)for x in sources],axis=1)
|
||||
# Material ownership is classified FROM the exact SourceNx index plane, not independently scaled label edges.
|
||||
labels=material_labels(guide,48)
|
||||
owners=np.repeat(np.repeat(labels,4,0),4,1)
|
||||
source_rgb=palette[guide].astype(float);output=np.zeros((512,1024,3),float);base_output=np.zeros_like(output)
|
||||
for component in range(5):
|
||||
mask=(labels==component).astype(float)
|
||||
weight=resize_float(gaussian(mask,1.0),(1024,512))
|
||||
numerator=resize_float(gaussian(source_rgb*mask[...,None],1.0),(1024,512))
|
||||
base=numerator/np.maximum(weight[...,None],1e-12)
|
||||
donor=np.asarray(fields[component],float)
|
||||
if donor.shape!=output.shape:raise ValueError('registered donor field size mismatch')
|
||||
# Preserve genuine bounded RGB texture, never donor coarse shape/color drift.
|
||||
band=np.clip(donor-gaussian(donor,8.0),-26,26)*0.75
|
||||
output[owners==component]=(base+band)[owners==component]
|
||||
base_output[owners==component]=base[owners==component]
|
||||
output=np.clip(np.rint(output),0,255).astype(np.uint8)
|
||||
protected=np.repeat(np.repeat(guide>=240,4,0),4,1)
|
||||
exact=np.repeat(np.repeat(palette[guide],4,0),4,1)
|
||||
output[protected]=exact[protected]
|
||||
base8=np.clip(np.rint(base_output),0,255).astype(np.uint8)
|
||||
if audit is not None:
|
||||
audit.update(genuine_RGB_band_changed_nonfunctional_pixels=int(np.any(output!=base8,axis=2)[~protected].sum()),nonfunctional_pixels=int((~protected).sum()))
|
||||
return output,owners,protected,guide
|
||||
|
||||
def validate_knots(knots):
|
||||
if len(knots)<2 or knots[0]!=0 or knots[-1]!=1 or any(not np.isfinite(x)for x in knots)or any(a>=b for a,b in zip(knots,knots[1:])):raise ValueError('strict finite0..1 knot partition required')
|
||||
|
||||
def warp_landmarks(rgb,tx,ty,sx,sy):
|
||||
for knots in (tx,ty,sx,sy):validate_knots(knots)
|
||||
if len(tx)!=len(sx)or len(ty)!=len(sy):raise ValueError('paired knot counts required')
|
||||
h,w,_=rgb.shape;mesh=[]
|
||||
xs=[round(v*w)for v in tx];ys=[round(v*h)for v in ty]
|
||||
for yi in range(len(ty)-1):
|
||||
for xi in range(len(tx)-1):
|
||||
rect=(xs[xi],ys[yi],xs[xi+1],ys[yi+1])
|
||||
if rect[2]<=rect[0]or rect[3]<=rect[1]:raise ValueError('empty mesh cell')
|
||||
l,r=sx[xi]*w,sx[xi+1]*w;t,b=sy[yi]*h,sy[yi+1]*h
|
||||
mesh.append((rect,(l,t,l,b,r,b,r,t)))
|
||||
return np.array(Image.fromarray(rgb).transform((w,h),Image.Transform.MESH,mesh,Image.Resampling.BICUBIC))
|
||||
|
||||
def main():
|
||||
ap=argparse.ArgumentParser();ap.add_argument('--recipe',type=Path,required=True);ap.add_argument('--output',type=Path,required=True);args=ap.parse_args();recipe_bytes,recipe_pin=read_contained(args.recipe.parent,args.recipe.name,MAX_RECIPE_BYTES);recipe=json.loads(recipe_bytes);inputs=BufferedInputs(args.recipe.parent);out=args.output
|
||||
_,tool_pin=read_contained(Path(__file__).parent,Path(__file__).name,1024*1024)
|
||||
if out.exists():raise ValueError('new output directory required')
|
||||
if recipe['method']!=METHOD:raise ValueError('unknown method')
|
||||
pb,ppin=inputs.read(recipe['palette']);pal=palette_rgb(pb);aud=[]
|
||||
sources=[]
|
||||
if recipe['operation']=='pair':
|
||||
if [x['asset']for x in recipe['sources']]!=list(PAIR_IDS):raise ValueError('only complete selected pair supported')
|
||||
for x in recipe['sources']:
|
||||
b,pin=inputs.read(x);sources.append(decode_source32(b));aud.append(pin)
|
||||
donors={}
|
||||
for key,binding in recipe['donors'].items():
|
||||
payload,pin=inputs.read(binding);rgb=decode_png(payload);donors[int(key)]=donor_field(rgb,binding['box'],(1024,512),binding.get('texture_period'));aud.append(pin)
|
||||
band_audit={};result,owners,protected,guide=derive_pair(sources,pal,donors,band_audit);out.mkdir(parents=True)
|
||||
Image.fromarray(result).save(out/'derived-pair1024x512.png')
|
||||
for j,aid in enumerate(PAIR_IDS):Image.fromarray(result[:,j*512:(j+1)*512]).save(out/f'{aid}-derived.png')
|
||||
metrics=dict(band_audit);metrics.update({'protected_native4_samples':int((guide>=240).sum()),'protected_HDraw_pixels':int(protected.sum()),'protected_source_Nx_RGBA_exact':bool(np.array_equal(result[protected],np.repeat(np.repeat(pal[guide],4,0),4,1)[protected])),'geometry_component_owner_sha256':hashlib.sha256(owners.tobytes()).hexdigest(),'retained_generated_texture_nonprotected_pixels':int(np.any(result!=np.repeat(np.repeat(pal[guide],4,0),4,1),axis=2)[~protected].sum())})
|
||||
elif recipe['operation']=='T-landmark':
|
||||
source=recipe['source'];assert source['asset']=='104fd';b,pin=inputs.read(source);idx=decode_source32(b);aud.append(pin)
|
||||
# The measured Source has a full-width ochre top line at row9 and its lower arm at23.
|
||||
if sum(80<=i<=95 for i in idx[9])!=32 or sum(80<=i<=95 for i in idx[23])!=19:raise ValueError('Source T topology changed')
|
||||
payload,pin=inputs.read(recipe['donor']);rgb=decode_png(payload);x,y,w,h=recipe['donor']['box']
|
||||
if min(x,y)<0 or min(w,h)<1 or x+w>rgb.shape[1]or y+h>rgb.shape[0]:raise ValueError('bad T donor crop')
|
||||
result=warp_landmarks(rgb[y:y+h,x:x+w],**recipe['knots']);out.mkdir(parents=True);Image.fromarray(result).save(out/'104fd-derived.png');aud.append(pin);metrics={'source_top_yellow_row':9,'source_lower_yellow_row':23,'knots':recipe['knots'],'genuine_RGB_only_warp':True,'protected_Source_cells_to_declare_original_detail':[[int(x),int(y)]for y,x in zip(*np.where(idx>=240))]}
|
||||
else:raise ValueError('unknown operation')
|
||||
receipt={'implementation':tool_pin,'executed_dependency':_dependency_pin,'recipe_input':recipe_pin,'physical_authenticated_input_reads':inputs.physical_reads,'method':METHOD,'operation':recipe['operation'],'input_pins':aud,'palette':ppin,'recipe_sha256':hashlib.sha256(recipe_bytes).hexdigest(),'metrics':metrics,'output_is_derived_not_untouched':True,'functional_retention_no_HD_credit':'Original protected cells>=240 retain exact SourceNx RGB/labels. No global Original-nearest fallback is claimed as HD.','outputs':[{'file':p.name,'sha256':hashlib.sha256(p.read_bytes()).hexdigest()}for p in sorted(out.glob('*.png'))]};(out/'ACTUAL-DERIVATION-RECEIPT.json').write_text(json.dumps(receipt,indent=2)+'\n');print(json.dumps(receipt['metrics']))
|
||||
if __name__=='__main__':main()
|
||||
@@ -0,0 +1,160 @@
|
||||
"""One buffered global crop/aspect correction; derived artwork, never approval.
|
||||
|
||||
No Source stencil, palette remapping, contrast, per-letter fit or added pixels.
|
||||
The fixed 10016 recipe keeps all alpha>=16 ink and discloses omitted low-alpha noise.
|
||||
"""
|
||||
import argparse
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path, PurePosixPath
|
||||
import stat
|
||||
import struct
|
||||
|
||||
from PIL import Image, __version__ as PILLOW_VERSION
|
||||
|
||||
METHOD = "derived-10016-genuine-charcoal-single-global-aspect-refit-v1"
|
||||
MAX_BYTES = 32 * 1024 * 1024
|
||||
MAX_PIXELS = 4_194_304
|
||||
|
||||
|
||||
def digest(payload):
|
||||
return hashlib.sha256(payload).hexdigest()
|
||||
|
||||
|
||||
SECURE_OPENAT_AVAILABLE = (os.name == "posix" and os.open in os.supports_dir_fd and
|
||||
hasattr(os, "O_DIRECTORY") and hasattr(os, "O_NOFOLLOW"))
|
||||
|
||||
|
||||
def regular_bytes(path):
|
||||
"""Read through authenticated directory descriptors, never a reopened path.
|
||||
|
||||
The authoring helper requires secure POSIX openat; other platforms fail closed.
|
||||
This restriction does not affect the Windows game or its generated assets.
|
||||
"""
|
||||
if not SECURE_OPENAT_AVAILABLE:
|
||||
raise ValueError("secure POSIX descriptor-relative input reads are required")
|
||||
path = Path(path).absolute()
|
||||
if ".." in path.parts or not path.name:
|
||||
raise ValueError("normalized regular input path required")
|
||||
identity = lambda s: (s.st_dev, s.st_ino, s.st_mode, s.st_size, s.st_mtime_ns)
|
||||
directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
|
||||
file_flags = os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_NONBLOCK", 0)
|
||||
directories = []
|
||||
try:
|
||||
directories.append(os.open(path.anchor, directory_flags))
|
||||
for part in path.parts[1:-1]:
|
||||
directories.append(os.open(part, directory_flags, dir_fd=directories[-1]))
|
||||
parent = directories[-1]
|
||||
before = os.stat(path.name, dir_fd=parent, follow_symlinks=False)
|
||||
if not stat.S_ISREG(before.st_mode) or before.st_size > MAX_BYTES:
|
||||
raise ValueError("bounded regular nonsymlink input required")
|
||||
with os.fdopen(os.open(path.name, file_flags, dir_fd=parent), "rb") as stream:
|
||||
opened = os.fstat(stream.fileno())
|
||||
if not stat.S_ISREG(opened.st_mode) or opened.st_size > MAX_BYTES or identity(opened) != identity(before):
|
||||
raise ValueError("opened input changed or is not bounded regular data")
|
||||
payload = stream.read(MAX_BYTES + 1)
|
||||
if len(payload) > MAX_BYTES or stream.read(1):
|
||||
raise ValueError("input exceeds bounded read")
|
||||
after = os.fstat(stream.fileno())
|
||||
final = os.stat(path.name, dir_fd=parent, follow_symlinks=False)
|
||||
if identity(before) != identity(after) or identity(before) != identity(final):
|
||||
raise ValueError("input changed during buffered read")
|
||||
return payload
|
||||
except OSError as error:
|
||||
raise ValueError("secure input directory/file traversal rejected") from error
|
||||
finally:
|
||||
for descriptor in reversed(directories):
|
||||
os.close(descriptor)
|
||||
|
||||
|
||||
def checked(binding, root):
|
||||
if type(binding) is not dict or set(binding) != {"path", "sha256"}:
|
||||
raise ValueError("exact portable path/SHA256 binding required")
|
||||
text, sha = binding["path"], binding["sha256"]
|
||||
if (type(text) is not str or not text or "\\" in text or ":" in text or
|
||||
PurePosixPath(text).is_absolute() or any(p in {"", ".", ".."} for p in text.split("/")) or
|
||||
type(sha) is not str or len(sha) != 64 or any(c not in "0123456789abcdef" for c in sha)):
|
||||
raise ValueError("unsafe path or invalid SHA256")
|
||||
payload = regular_bytes(Path(root) / text)
|
||||
if digest(payload) != sha:
|
||||
raise ValueError("input hash mismatch")
|
||||
return payload
|
||||
|
||||
|
||||
def global_crop_refit(raw_payload, window, output_size, omitted_low_alpha):
|
||||
"""Pure single crop then Lanczos refit using only verified parent bytes."""
|
||||
with Image.open(io.BytesIO(raw_payload)) as opened:
|
||||
if opened.format != "PNG" or max(opened.size) > 4096 or opened.width * opened.height > MAX_PIXELS:
|
||||
raise ValueError("bounded PNG parent required")
|
||||
raw = opened.convert("RGBA")
|
||||
if (type(window) is not list or len(window) != 4 or any(type(v) is not int for v in window) or
|
||||
type(output_size) is not list or len(output_size) != 2 or any(type(v) is not int for v in output_size) or
|
||||
type(omitted_low_alpha) is not int or omitted_low_alpha < 0):
|
||||
raise ValueError("integer crop/output/noise declarations required")
|
||||
x, y, w, h = window
|
||||
if not (x >= 0 and y >= 0 and w > 0 and h > 0 and x+w <= raw.width and y+h <= raw.height):
|
||||
raise ValueError("crop exceeds parent")
|
||||
if min(output_size) <= 0 or max(output_size) > 4096 or output_size[0] * output_size[1] > MAX_PIXELS:
|
||||
raise ValueError("output canvas exceeds bounds")
|
||||
alpha = raw.getchannel("A")
|
||||
outside = alpha.copy()
|
||||
outside.paste(0, (x, y, x+w, y+h))
|
||||
histogram = outside.histogram()
|
||||
if sum(histogram[16:]):
|
||||
raise ValueError("crop would discard occupied alpha>=16 parent ink")
|
||||
if sum(histogram[1:16]) != omitted_low_alpha:
|
||||
raise ValueError("omitted low-alpha noise count differs")
|
||||
fitted = raw.crop((x, y, x+w, y+h)).resize(tuple(output_size), Image.Resampling.LANCZOS)
|
||||
buffer = io.BytesIO()
|
||||
fitted.save(buffer, format="PNG", compress_level=9)
|
||||
return buffer.getvalue(), dict(parent_canvas=list(raw.size), window=window,
|
||||
output_canvas=output_size, all_alpha16_ink_retained=True,
|
||||
omitted_alpha1_to15_samples=omitted_low_alpha,
|
||||
output_RGBA_sha256=digest(fitted.tobytes()),
|
||||
operation="one global crop then Lanczos aspect refit, no further pixel operations")
|
||||
|
||||
|
||||
def run_recipe(recipe_path, root, output):
|
||||
output = Path(output)
|
||||
if output.exists():
|
||||
raise ValueError("fresh absent output directory required")
|
||||
rb = regular_bytes(recipe_path)
|
||||
recipe = json.loads(rb)
|
||||
fields = {"format_version", "method", "asset", "source", "palette", "parent_raw",
|
||||
"window", "output_size", "omitted_low_alpha", "pillow_version"}
|
||||
if (type(recipe) is not dict or set(recipe) != fields or type(recipe["format_version"]) is not int or
|
||||
recipe["format_version"] != 1 or recipe["method"] != METHOD or recipe["asset"] != "10016" or
|
||||
recipe["window"] != [0, 289, 2167, 130] or recipe["output_size"] != [2176, 128] or
|
||||
recipe["omitted_low_alpha"] != 100 or recipe["pillow_version"] != PILLOW_VERSION):
|
||||
raise ValueError("exact fixed10016 method/recipe/Pillow version required")
|
||||
source = checked(recipe["source"], root)
|
||||
palette = checked(recipe["palette"], root)
|
||||
raw = checked(recipe["parent_raw"], root)
|
||||
if (len(source) < 20 or list(struct.unpack_from("<5i", source)) != [0, 0, 8, 136, 8] or
|
||||
len(palette) != 768):
|
||||
raise ValueError("Original header or palette differs")
|
||||
png, detail = global_crop_refit(raw, recipe["window"], recipe["output_size"], recipe["omitted_low_alpha"])
|
||||
if detail["parent_canvas"] != [2167, 725]:
|
||||
raise ValueError("fixed10016 genuine parent canvas differs")
|
||||
execution = dict(status="success", method=METHOD, recipe_sha256=digest(rb),
|
||||
tool_sha256=digest(regular_bytes(__file__)), source_sha256={"10016": digest(source)},
|
||||
parent_raw_sha256=[digest(raw)], output_sha256={"10016": digest(png)})
|
||||
report = dict(execution=execution, geometry=detail, pillow_version=PILLOW_VERSION,
|
||||
original_palette_sha256=digest(palette), source_supplies_geometry_or_RGB=False,
|
||||
no_artistic_or_runtime_approval=True)
|
||||
# Validate every input and render in memory before publishing a fresh result.
|
||||
output.mkdir(parents=True, exist_ok=False)
|
||||
(output / "10016-derived.png").write_bytes(png)
|
||||
(output / "execution-receipt.json").write_text(json.dumps(report, indent=2) + "\n")
|
||||
return report
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("recipe", type=Path)
|
||||
parser.add_argument("hd_root", type=Path)
|
||||
parser.add_argument("fresh_output", type=Path)
|
||||
args = parser.parse_args()
|
||||
print(json.dumps(run_recipe(args.recipe, args.hd_root, args.fresh_output)["execution"], sort_keys=True))
|
||||
@@ -0,0 +1,203 @@
|
||||
"""DERIVED gray-cast correction, preserving generated luminance structure.
|
||||
|
||||
Bounded to the two reviewed grating compositions. Source supplies only an exact
|
||||
semantic mask and ONE aggregate gray color per frame, never a painted SourceNx
|
||||
field. This tool does not approve art, create a runtime policy or modify inputs.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import argparse, ctypes, errno, hashlib, io, json, os, shutil, stat, struct, sys, tempfile
|
||||
from pathlib import Path, PurePosixPath
|
||||
import numpy as np
|
||||
from PIL import Image, __version__ as PILLOW_VERSION
|
||||
|
||||
METHOD = 'generated-grating-semantic-gray-cohort-mean-cast-only-v1'
|
||||
REGIONS = {'203c6': [21,3,31,31], '203c7': [2,0,31,12]}
|
||||
MAX_BYTES = 32*1024*1024
|
||||
MAX_PIXELS = 4_194_304
|
||||
MAX_SHIFT = 32.0
|
||||
LUMA = np.asarray([0.2126,0.7152,0.0722], np.float64)
|
||||
|
||||
|
||||
def sha(b): return hashlib.sha256(b).hexdigest()
|
||||
|
||||
def read_regular(root, relative, limit=MAX_BYTES):
|
||||
"""Every component, including the root, is walked with stable no-follow FDs."""
|
||||
if type(relative)is not str:raise ValueError('string relative path required')
|
||||
p=PurePosixPath(relative)
|
||||
if (p.is_absolute() or not p.parts
|
||||
or '..'in p.parts or p.as_posix()!=relative):
|
||||
raise ValueError('normalized contained relative path required')
|
||||
base=Path(root).absolute();fd=os.open('/',os.O_RDONLY|os.O_DIRECTORY|os.O_NONBLOCK)
|
||||
try:
|
||||
for name in list(base.parts[1:])+list(p.parts[:-1]):
|
||||
newer=os.open(name,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_NONBLOCK,dir_fd=fd)
|
||||
os.close(fd);fd=newer
|
||||
f=os.open(p.parts[-1],os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK,dir_fd=fd)
|
||||
try:
|
||||
before=os.fstat(f)
|
||||
if not stat.S_ISREG(before.st_mode)or before.st_size>limit:
|
||||
raise ValueError('bounded regular input required')
|
||||
with os.fdopen(f,'rb',closefd=False)as stream:body=stream.read(limit+1)
|
||||
after=os.fstat(f)
|
||||
identity=lambda q:(q.st_dev,q.st_ino,q.st_size,q.st_mtime_ns,q.st_ctime_ns)
|
||||
if identity(before)!=identity(after)or len(body)!=before.st_size or len(body)>limit:
|
||||
raise ValueError('input changed during bounded read')
|
||||
finally:os.close(f)
|
||||
finally:os.close(fd)
|
||||
return body
|
||||
|
||||
|
||||
def read_bound(root,binding,limit=MAX_BYTES):
|
||||
if (type(binding)is not dict or set(binding)!={'path','sha256'}
|
||||
or type(binding['sha256'])is not str or len(binding['sha256'])!=64
|
||||
or any(c not in '0123456789abcdef'for c in binding['sha256'])):
|
||||
raise ValueError('exact relative path/sha256 binding required')
|
||||
body=read_regular(root,binding['path'],limit)
|
||||
if sha(body)!=binding['sha256']:raise ValueError('input SHA mismatch')
|
||||
return body
|
||||
|
||||
|
||||
def source_indices(body):
|
||||
if len(body)!=1044 or struct.unpack('<5i',body[:20])!=(1,1,0,32,32):
|
||||
raise ValueError('exact complete uncompressed 32x32 Source required')
|
||||
a=np.frombuffer(body[20:],np.uint8).reshape(32,32).copy()
|
||||
if np.any(a>=240):raise ValueError('only opaque ordinary Source supported')
|
||||
return a
|
||||
|
||||
|
||||
def palette_rgb(body):
|
||||
if len(body)!=768 or max(body)>63:raise ValueError('authentic six-bit palette required')
|
||||
a=np.frombuffer(body,np.uint8).astype(np.uint16).reshape(256,3)
|
||||
return ((a<<2)|(a>>4)).astype(np.uint8)
|
||||
|
||||
|
||||
def semantic_mask(asset, source):
|
||||
if asset not in REGIONS:raise ValueError('unreviewed grating asset')
|
||||
x0,y0,x1,y1=REGIONS[asset];m=np.zeros((32,32),bool)
|
||||
m[y0:y1,x0:x1]=(source[y0:y1,x0:x1]>=11)&(source[y0:y1,x0:x1]<=30)
|
||||
if not m.any():raise ValueError('no authentic gray grating material')
|
||||
return m
|
||||
|
||||
|
||||
def selected_digest(mask):
|
||||
return sha(json.dumps(sorted([[int(x),int(y)]for y,x in np.argwhere(mask)]),
|
||||
separators=(',',':')).encode())
|
||||
|
||||
|
||||
def validate_recipe(r):
|
||||
if (type(r)is not dict or set(r)!={'format_version','method','palette','parent_raw','frames'}
|
||||
or type(r['format_version'])is not int or r['format_version']!=1 or r['method']!=METHOD
|
||||
or type(r['frames'])is not list or len(r['frames'])!=2):
|
||||
raise ValueError('exact reviewed two-frame recipe required')
|
||||
if any(type(f)is not dict for f in r['frames']):raise ValueError('frame objects required')
|
||||
if [f.get('asset')for f in r['frames']]!=['203c6','203c7']:
|
||||
raise ValueError('ordered complete pair required')
|
||||
for f in r['frames']:
|
||||
if (set(f)!={'asset','Source','selected_cells_sha256'}
|
||||
or type(f['selected_cells_sha256'])is not str
|
||||
or len(f['selected_cells_sha256'])!=64
|
||||
or any(c not in '0123456789abcdef'for c in f['selected_cells_sha256'])):
|
||||
raise ValueError('unknown frame field or selector digest')
|
||||
|
||||
|
||||
def prepare_inputs(r, root):
|
||||
validate_recipe(r)
|
||||
return (palette_rgb(read_bound(root,r['palette'],768)),
|
||||
read_bound(root,r['parent_raw']),
|
||||
[(f,source_indices(read_bound(root,f['Source'],1044)))for f in r['frames']])
|
||||
|
||||
|
||||
def derive(palette, raw_bytes, frames):
|
||||
"""Pure verified-bytes operation. Never reopen source or parent filenames."""
|
||||
with Image.open(io.BytesIO(raw_bytes)) as opened:
|
||||
if (opened.format!='PNG' or opened.mode not in ('RGB','RGBA') or opened.width!=2*opened.height
|
||||
or not 128<=opened.height<=2048 or opened.width*opened.height>MAX_PIXELS):
|
||||
raise ValueError('bounded exact 2x1 PNG parent required')
|
||||
raw=np.asarray(opened.convert('RGBA')).copy()
|
||||
if np.any(raw[:,:,3]!=255):raise ValueError('no alpha repair/hidden RGB allowed')
|
||||
size=raw.shape[0];output=[]
|
||||
for column,(f,src)in enumerate(frames):
|
||||
mask=semantic_mask(f['asset'],src)
|
||||
if selected_digest(mask)!=f['selected_cells_sha256']:raise ValueError('Source cohort digest mismatch')
|
||||
parent=raw[:,column*size:(column+1)*size,:]
|
||||
# A raw 3x3 luminance witness must lie wholly in the same gray cohort.
|
||||
coordinates=((np.arange(size)*2+1)*32)//(size*2)
|
||||
mapped=mask[coordinates[:,None],coordinates[None,:]]
|
||||
luma=parent[:,:,:3].astype(np.float64)@LUMA
|
||||
eligible=mapped[1:-1,1:-1].copy();low=luma[1:-1,1:-1].copy();high=low.copy()
|
||||
for dy,dx in [(-1,-1),(-1,0),(-1,1),(0,-1),(0,1),(1,-1),(1,0),(1,1)]:
|
||||
yy=slice(1+dy,size-1+dy);xx=slice(1+dx,size-1+dx)
|
||||
eligible &= mapped[yy,xx];low=np.minimum(low,luma[yy,xx]);high=np.maximum(high,luma[yy,xx])
|
||||
witnesses=int(np.count_nonzero(eligible&(high-low>=2.0)))
|
||||
if witnesses==0:raise ValueError('no authentic raw same-material fine luminance witness')
|
||||
baseline=np.asarray(Image.fromarray(parent).resize((128,128),Image.Resampling.LANCZOS)).copy()
|
||||
area=np.repeat(np.repeat(mask,4,0),4,1);rgb=baseline[:,:,:3].astype(np.float64)
|
||||
generated_luma=rgb@LUMA;anchor=palette[src[mask]].astype(np.float64).mean(axis=0)
|
||||
if anchor.max()-anchor.min()>8:raise ValueError('Source cohort is not near-neutral gray')
|
||||
anchor_luma=float(anchor@LUMA);shift=anchor_luma-float(generated_luma[area].mean())
|
||||
if abs(shift)>MAX_SHIFT:raise ValueError('gray mean correction exceeds fixed32 cap')
|
||||
proposed=(generated_luma[area]+shift)[:,None]+(anchor-anchor_luma)[None,:]
|
||||
if np.any(proposed<0)or np.any(proposed>255):raise ValueError('clipping forbidden')
|
||||
result=baseline.copy();result[:,:,:3][area]=np.rint(proposed).astype(np.uint8)
|
||||
if not np.array_equal(result[~area],baseline[~area]):raise AssertionError('outside changed')
|
||||
transformed_luma=result[:,:,:3].astype(np.float64)@LUMA
|
||||
error=float(np.abs(transformed_luma[area]-(generated_luma[area]+shift)).max())
|
||||
if error>0.5000001:raise AssertionError('generated luminance structure lost')
|
||||
metrics=dict(asset=f['asset'],semantic_Source_cells=int(mask.sum()),corrected_output_pixels=int(area.sum()),
|
||||
unmodified_resized_parent_pixels=int((~area).sum()),Source_aggregate_RGB=anchor.tolist(),
|
||||
generated_luma_mean=float(generated_luma[area].mean()),constant_gray_mean_offset=shift,
|
||||
raw_same_cohort_3x3_luma_witnesses=witnesses,max_luma_structure_rounding_error=error,
|
||||
generated_luma_variance=float(generated_luma[area].var()),
|
||||
output_luma_variance=float(transformed_luma[area].var()),
|
||||
exact_Original_pixel_copies=0,outside_resized_parent_exact=True,
|
||||
selected_cells_sha256=selected_digest(mask),output_scale=4)
|
||||
output.append((f['asset'],result,metrics))
|
||||
return output
|
||||
|
||||
|
||||
def publish_no_replace(stage,target):
|
||||
if os.name=='nt':os.rename(stage,target);return
|
||||
if not sys.platform.startswith('linux'):raise OSError(errno.ENOTSUP,'atomic publication unsupported')
|
||||
libc=ctypes.CDLL(None,use_errno=True)
|
||||
if not hasattr(libc,'renameat2'):raise OSError(errno.ENOTSUP,'renameat2 unavailable')
|
||||
fn=libc.renameat2;fn.argtypes=[ctypes.c_int,ctypes.c_char_p,ctypes.c_int,ctypes.c_char_p,ctypes.c_uint]
|
||||
fn.restype=ctypes.c_int
|
||||
if fn(-100,os.fsencode(stage),-100,os.fsencode(target),1)!=0:
|
||||
code=ctypes.get_errno();raise OSError(code,os.strerror(code),str(target))
|
||||
|
||||
|
||||
def run(recipe_relative,input_root,output):
|
||||
root=Path(input_root).absolute();rel=PurePosixPath(recipe_relative)
|
||||
if rel.is_absolute()or '..'in rel.parts or not rel.parts:raise ValueError('relative recipe required')
|
||||
# Recipe bytes use the same no-follow reader after a separately bound hash.
|
||||
body=read_regular(root,rel.as_posix(),1_000_000)
|
||||
r=json.loads(body);prepared=prepare_inputs(r,root);result=derive(*prepared)
|
||||
output=Path(output).absolute()
|
||||
if output.exists()or output.is_symlink()or not output.parent.is_dir():raise ValueError('fresh output directory required')
|
||||
tool_path=Path(__file__).absolute();tool_bytes=read_regular(tool_path.parent,tool_path.name);outputs={};pngs={}
|
||||
for asset,rgba,_ in result:
|
||||
stream=io.BytesIO();Image.fromarray(rgba).save(stream,format='PNG');pngs[asset]=stream.getvalue();outputs[asset]=sha(pngs[asset])
|
||||
execution=dict(status='success',method=METHOD,recipe_sha256=sha(body),tool_sha256=sha(tool_bytes),
|
||||
source_sha256={f['asset']:f['Source']['sha256']for f in r['frames']},
|
||||
parent_raw_sha256=[r['parent_raw']['sha256']],output_sha256=outputs)
|
||||
receipt=dict(status='DERIVED_NOT_ART_APPROVAL',execution=execution,frames=[m for _,_,m in result],
|
||||
runtime=dict(python=sys.version,numpy=np.__version__,Pillow=PILLOW_VERSION),
|
||||
limits='Source owns semantic geometry and one aggregate gray anchor only; generated raw owns every luminance difference. No per-cell Source RGB copying, flat fallback, new pixels outside resized-parent selection, or quality approval. Resized-parent baseline uses Pillow Lanczos3; equality to old Rust raw-bank resampling is NOT claimed.')
|
||||
stage=Path(tempfile.mkdtemp(prefix='.'+output.name+'.stage-',dir=output.parent))
|
||||
try:
|
||||
for asset in pngs:(stage/(asset+'-DERIVED-graycast-v1.png')).write_bytes(pngs[asset])
|
||||
(stage/'execution-receipt.json').write_text(json.dumps(receipt,indent=2)+'\n')
|
||||
if read_bound(root,{'path':rel.as_posix(),'sha256':sha(body)},1_000_000)!=body:raise ValueError('recipe changed')
|
||||
prepare_inputs(r,root) # Reauthenticate originals; no second render.
|
||||
if read_regular(tool_path.parent,tool_path.name)!=tool_bytes:raise ValueError('tool changed')
|
||||
publish_no_replace(stage,output)
|
||||
except BaseException:shutil.rmtree(stage,ignore_errors=True);raise
|
||||
return receipt
|
||||
|
||||
|
||||
def main():
|
||||
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--recipe',required=True)
|
||||
p.add_argument('--input-root',required=True);p.add_argument('--output',required=True);a=p.parse_args()
|
||||
run(a.recipe,a.input_root,a.output)
|
||||
|
||||
if __name__=='__main__':main()
|
||||
@@ -0,0 +1,146 @@
|
||||
import hashlib
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from PIL import Image, ImageDraw
|
||||
import hd_global_aspect_cleanup as cleanup
|
||||
|
||||
|
||||
def encode(image):
|
||||
b = io.BytesIO(); image.save(b, format="PNG"); return b.getvalue()
|
||||
|
||||
|
||||
class GlobalAspectCleanupTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name)
|
||||
|
||||
def ink(self):
|
||||
image = Image.new("RGBA", (34, 16))
|
||||
draw = ImageDraw.Draw(image)
|
||||
# Countered glyph, detached dot and comma, ring touching both crop edges.
|
||||
draw.rectangle((2, 5, 5, 10), fill=(20, 20, 20, 255))
|
||||
draw.rectangle((3, 6, 4, 9), fill=(0, 0, 0, 0))
|
||||
draw.point((8, 10), fill=(24, 24, 24, 255))
|
||||
draw.line((10, 10, 10, 11), fill=(18, 18, 18, 255))
|
||||
draw.ellipse((15, 4, 22, 11), outline=(24, 24, 24, 255))
|
||||
return image
|
||||
|
||||
def test_only_global_crop_and_refit_retains_counter_dot_comma_and_ring(self):
|
||||
parent = self.ink()
|
||||
output, report = cleanup.global_crop_refit(encode(parent), [0, 4, 34, 8], [68, 16], 0)
|
||||
actual = Image.open(io.BytesIO(output)).convert("RGBA")
|
||||
self.assertEqual(actual.size, (68, 16))
|
||||
self.assertEqual(actual.tobytes(), parent.crop((0, 4, 34, 12)).resize((68, 16), Image.Resampling.LANCZOS).tobytes())
|
||||
self.assertLess(actual.getpixel((7, 7))[3], 16) # counter remains below ink cutoff
|
||||
self.assertGreater(actual.getpixel((16, 12))[3], 128) # separate full stop
|
||||
self.assertGreater(actual.getpixel((20, 14))[3], 128) # descending comma
|
||||
self.assertTrue(report["all_alpha16_ink_retained"])
|
||||
|
||||
def test_occupied_ink_outside_window_rejected(self):
|
||||
parent = self.ink(); parent.putpixel((2, 0), (22, 22, 22, 16))
|
||||
with self.assertRaisesRegex(ValueError, "occupied alpha>=16"):
|
||||
cleanup.global_crop_refit(encode(parent), [0, 4, 34, 8], [68, 16], 0)
|
||||
|
||||
def test_low_alpha_noise_is_exactly_declared_not_silently_removed(self):
|
||||
parent = self.ink(); parent.putpixel((2, 0), (22, 22, 22, 15))
|
||||
with self.assertRaisesRegex(ValueError, "noise count differs"):
|
||||
cleanup.global_crop_refit(encode(parent), [0, 4, 34, 8], [68, 16], 0)
|
||||
_, r = cleanup.global_crop_refit(encode(parent), [0, 4, 34, 8], [68, 16], 1)
|
||||
self.assertEqual(r["omitted_alpha1_to15_samples"], 1)
|
||||
|
||||
def test_no_genuine_RGB_tone_gain_or_Source_reinsertion_when_size_unchanged(self):
|
||||
parent = self.ink(); parent.putpixel((2, 5), (14, 19, 23, 255))
|
||||
out, _ = cleanup.global_crop_refit(encode(parent), [0, 4, 34, 8], [34, 8], 0)
|
||||
self.assertEqual(Image.open(io.BytesIO(out)).getpixel((2, 1)), (14, 19, 23, 255))
|
||||
|
||||
def test_invalid_bounds_boolean_dimensions_and_output_limit_rejected(self):
|
||||
raw = encode(self.ink())
|
||||
for window, size in [([0, 4, 35, 8], [68, 16]), ([True, 4, 34, 8], [68, 16]),
|
||||
([0, 4, 34, 8], [4097, 16]), ([0, 4, 34, 8], [68, 0])]:
|
||||
with self.assertRaises(ValueError): cleanup.global_crop_refit(raw, window, size, 0)
|
||||
|
||||
@unittest.skipUnless(cleanup.SECURE_OPENAT_AVAILABLE, "requires secure POSIX openat")
|
||||
def test_wrong_hash_or_path_escape_rejected_before_decoding(self):
|
||||
(self.root / "input").write_bytes(b"hello")
|
||||
with self.assertRaisesRegex(ValueError, "hash mismatch"):
|
||||
cleanup.checked(dict(path="input", sha256="0" * 64), self.root)
|
||||
for path in ["../input", "/input", "a//input", "a/./input", "C:input", "a\\input"]:
|
||||
with self.assertRaisesRegex(ValueError, "unsafe"):
|
||||
cleanup.checked(dict(path=path, sha256="0" * 64), self.root)
|
||||
|
||||
@unittest.skipUnless(cleanup.SECURE_OPENAT_AVAILABLE, "requires secure POSIX openat")
|
||||
def test_regular_file_byte_limit_and_symlink_rejected(self):
|
||||
p = self.root / "input"; p.write_bytes(b"12345")
|
||||
with patch.object(cleanup, "MAX_BYTES", 4):
|
||||
with self.assertRaisesRegex(ValueError, "bounded regular"):
|
||||
cleanup.regular_bytes(p)
|
||||
link = self.root / "link"
|
||||
try: link.symlink_to(p)
|
||||
except OSError: self.skipTest("symlinks unavailable")
|
||||
with self.assertRaisesRegex(ValueError, "symlink"):
|
||||
cleanup.regular_bytes(link)
|
||||
|
||||
@unittest.skipUnless(cleanup.SECURE_OPENAT_AVAILABLE, "requires secure POSIX openat")
|
||||
def test_buffer_consumed_once_and_changed_descriptor_rejected(self):
|
||||
p = self.root / "input"; p.write_bytes(b"old")
|
||||
actual_open = os.open
|
||||
def replace(name, flags, **kwargs):
|
||||
if str(name) == "input":
|
||||
replacement = self.root / "replacement"; replacement.write_bytes(b"new")
|
||||
replacement.replace(p)
|
||||
return actual_open(name, flags, **kwargs)
|
||||
with patch.object(cleanup.os, "open", replace):
|
||||
with self.assertRaisesRegex(ValueError, "opened input changed"):
|
||||
cleanup.regular_bytes(p)
|
||||
|
||||
@unittest.skipUnless(cleanup.SECURE_OPENAT_AVAILABLE, "requires secure POSIX openat")
|
||||
def test_parent_swap_before_legacy_lstat_cannot_authenticate_outside_bytes(self):
|
||||
pack = self.root / "pack"; folder = pack / "inputs"; folder.mkdir(parents=True)
|
||||
(folder / "input").write_bytes(b"inside")
|
||||
outside = self.root / "outside"; outside.mkdir(); (outside / "input").write_bytes(b"outside")
|
||||
actual_lstat = Path.lstat
|
||||
swapped = []; file_lstat_calls = []
|
||||
def swap_before_legacy_lstat(path, *args, **kwargs):
|
||||
if path == folder / "input":
|
||||
file_lstat_calls.append(True)
|
||||
# First call belongs to the legacy is_symlink precheck; the second
|
||||
# is its unchecked initial metadata read after all parent prechecks.
|
||||
if path == folder / "input" and len(file_lstat_calls) == 2 and not swapped:
|
||||
folder.rename(pack / "original-inputs"); folder.symlink_to(outside, target_is_directory=True)
|
||||
swapped.append(True)
|
||||
return actual_lstat(path, *args, **kwargs)
|
||||
binding = dict(path="inputs/input", sha256=hashlib.sha256(b"outside").hexdigest())
|
||||
with patch.object(Path, "lstat", swap_before_legacy_lstat):
|
||||
with self.assertRaises(ValueError): cleanup.checked(binding, pack)
|
||||
# V1 follows the replaced parent and returns outside bytes matching the hash.
|
||||
# V2 never uses this unchecked absolute lstat and reads the legitimate tree.
|
||||
|
||||
@unittest.skipUnless(cleanup.SECURE_OPENAT_AVAILABLE, "requires secure POSIX openat")
|
||||
def test_actual_openat_parent_component_swap_rejects_symlink(self):
|
||||
pack = self.root / "pack"; folder = pack / "inputs"; folder.mkdir(parents=True)
|
||||
(folder / "input").write_bytes(b"inside")
|
||||
outside = self.root / "outside"; outside.mkdir(); (outside / "input").write_bytes(b"outside")
|
||||
actual_open = os.open; swapped = []
|
||||
def swap_component(name, flags, **kwargs):
|
||||
if str(name) == "inputs" and not swapped:
|
||||
folder.rename(pack / "original-inputs"); folder.symlink_to(outside, target_is_directory=True)
|
||||
swapped.append(True)
|
||||
return actual_open(name, flags, **kwargs)
|
||||
with patch.object(cleanup.os, "open", swap_component):
|
||||
with self.assertRaisesRegex(ValueError, "traversal rejected"):
|
||||
cleanup.checked(dict(path="inputs/input", sha256=hashlib.sha256(b"outside").hexdigest()), pack)
|
||||
self.assertEqual(swapped, [True])
|
||||
|
||||
def test_existing_output_refused_before_recipe_read(self):
|
||||
out = self.root / "occupied"; out.mkdir()
|
||||
with self.assertRaisesRegex(ValueError, "fresh absent"):
|
||||
cleanup.run_recipe(self.root / "missing", self.root, out)
|
||||
|
||||
|
||||
if __name__ == "__main__": unittest.main()
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Synthetic inputs only: no game masters or real cleanup execution."""
|
||||
import copy,io,json,os,struct,tempfile,unittest,subprocess,sys
|
||||
from unittest import mock
|
||||
from types import SimpleNamespace
|
||||
from pathlib import Path
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
import hd_grating_gray_cleanup as m
|
||||
|
||||
class GrayCastTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name)
|
||||
pal=np.zeros((256,3),np.uint8);pal[11]=[18,18,18];pal[30]=[4,4,4]
|
||||
(self.root/'palette.bin').write_bytes(pal.tobytes())
|
||||
y,x=np.indices((128,256));rgb=np.stack([50+x%5+y%7,35+x%5+y%7,28+x%5+y%7],axis=2).astype('uint8')
|
||||
im=Image.fromarray(rgb);im.save(self.root/'parent.png')
|
||||
self.source=np.full((32,32),199,np.uint8);self.source[2:31,2:31]=11
|
||||
self.recipe=dict(format_version=1,method=m.METHOD,palette=self.bind('palette.bin'),parent_raw=self.bind('parent.png'),frames=[])
|
||||
for asset in ['203c6','203c7']:
|
||||
name=asset+'.bin';(self.root/name).write_bytes(struct.pack('<5i',1,1,0,32,32)+self.source.tobytes())
|
||||
self.recipe['frames'].append(dict(asset=asset,Source=self.bind(name),selected_cells_sha256=m.selected_digest(m.semantic_mask(asset,self.source))))
|
||||
def tearDown(self):self.temp.cleanup()
|
||||
def bind(self,n):return dict(path=n,sha256=m.sha((self.root/n).read_bytes()))
|
||||
def prepared(self):return m.prepare_inputs(self.recipe,self.root)
|
||||
def test_cast_neutral_preserves_generated_luminance(self):
|
||||
result=m.derive(*self.prepared());self.assertEqual(len(result),2)
|
||||
for a,out,metrics in result:
|
||||
self.assertEqual(out.shape,(128,128,4));self.assertTrue(np.all(out[:,:,3]==255))
|
||||
self.assertTrue(metrics['outside_resized_parent_exact']);self.assertEqual(metrics['exact_Original_pixel_copies'],0)
|
||||
self.assertLessEqual(metrics['max_luma_structure_rounding_error'],.5000001)
|
||||
self.assertGreater(metrics['generated_luma_variance'],1);self.assertGreater(metrics['output_luma_variance'],1)
|
||||
self.assertLess(abs(metrics['output_luma_variance']-metrics['generated_luma_variance']),2)
|
||||
def test_repeatability(self):
|
||||
a=m.derive(*self.prepared());b=m.derive(*self.prepared())
|
||||
self.assertTrue(all(np.array_equal(x[1],y[1])and x[2]==y[2]for x,y in zip(a,b)))
|
||||
def test_all_semantic_cells_not_error_picker(self):
|
||||
for asset in m.REGIONS:
|
||||
mask=m.semantic_mask(asset,self.source);x0,y0,x1,y1=m.REGIONS[asset]
|
||||
self.assertEqual(int(mask.sum()),int(np.count_nonzero(self.source[y0:y1,x0:x1]==11)))
|
||||
def test_missing_or_unknown_fields(self):
|
||||
for key in ['method','palette','frames']:
|
||||
r=copy.deepcopy(self.recipe);del r[key]
|
||||
with self.assertRaises(ValueError):m.validate_recipe(r)
|
||||
r=copy.deepcopy(self.recipe);r['gain']=1.2
|
||||
with self.assertRaises(ValueError):m.validate_recipe(r)
|
||||
def test_bool_version_rejected(self):
|
||||
r=copy.deepcopy(self.recipe);r['format_version']=True
|
||||
with self.assertRaises(ValueError):m.validate_recipe(r)
|
||||
def test_order_and_selector_are_fixed(self):
|
||||
r=copy.deepcopy(self.recipe);r['frames'].reverse()
|
||||
with self.assertRaises(ValueError):m.validate_recipe(r)
|
||||
p,raw,frames=self.prepared();frames[0][0]['selected_cells_sha256']='0'*64
|
||||
with self.assertRaises(ValueError):m.derive(p,raw,frames)
|
||||
def test_wrong_hash_and_escaping_paths(self):
|
||||
for ref in [dict(path='parent.png',sha256='0'*64),dict(path='../parent.png',sha256='0'*64),dict(path='/parent.png',sha256='0'*64),dict(path='a/../parent.png',sha256='0'*64)]:
|
||||
with self.assertRaises((ValueError,OSError)):m.read_bound(self.root,ref)
|
||||
def test_symlink_file_and_parent_rejected(self):
|
||||
(self.root/'alias.png').symlink_to(self.root/'parent.png')
|
||||
with self.assertRaises(OSError):m.read_bound(self.root,dict(path='alias.png',sha256=self.bind('parent.png')['sha256']))
|
||||
(self.root/'folder').symlink_to(self.root,target_is_directory=True)
|
||||
with self.assertRaises(OSError):m.read_bound(self.root,dict(path='folder/parent.png',sha256=self.bind('parent.png')['sha256']))
|
||||
def test_bytes_are_not_reopened(self):
|
||||
prepared=self.prepared();first=m.derive(*prepared)
|
||||
(self.root/'parent.png').write_bytes(b'corrupted after authenticated read')
|
||||
for a,b in zip(first,m.derive(*prepared)):self.assertTrue(np.array_equal(a[1],b[1]))
|
||||
def test_flat_parent_rejected(self):
|
||||
p,_,f=self.prepared();stream=io.BytesIO();Image.new('RGB',(256,128),(50,35,28)).save(stream,format='PNG')
|
||||
with self.assertRaisesRegex(ValueError,'witness'):m.derive(p,stream.getvalue(),f)
|
||||
def test_transparency_size_and_shift_caps(self):
|
||||
p,_,f=self.prepared()
|
||||
for image in [Image.new('RGBA',(256,128),(50,40,30,0)),Image.new('RGB',(128,128)),Image.new('RGB',(256,64))]:
|
||||
stream=io.BytesIO();image.save(stream,format='PNG')
|
||||
with self.assertRaises(ValueError):m.derive(p,stream.getvalue(),f)
|
||||
p[11]=[240,240,240]
|
||||
with self.assertRaisesRegex(ValueError,'fixed32'):m.derive(p,self.prepared()[1],f)
|
||||
def test_source_and_palette_validation(self):
|
||||
for b in [b'',struct.pack('<5i',2,1,0,32,32)+self.source.tobytes(),struct.pack('<5i',1,1,0,32,32)+bytes([254])*1024]:
|
||||
with self.assertRaises(ValueError):m.source_indices(b)
|
||||
for b in [bytes(767),bytes([64])*768]:
|
||||
with self.assertRaises(ValueError):m.palette_rgb(b)
|
||||
def test_fresh_publication_receipt_and_no_overwrite(self):
|
||||
(self.root/'recipe.json').write_text(json.dumps(self.recipe));out=self.root/'new-output'
|
||||
result=m.run('recipe.json',self.root,out)
|
||||
self.assertEqual(set(result['execution']),{'status','method','recipe_sha256','tool_sha256','source_sha256','parent_raw_sha256','output_sha256'})
|
||||
self.assertEqual(set(result['execution']['output_sha256']),{'203c6','203c7'})
|
||||
before=sorted(p.name for p in out.iterdir())
|
||||
with self.assertRaises(ValueError):m.run('recipe.json',self.root,out)
|
||||
self.assertEqual(before,sorted(p.name for p in out.iterdir()))
|
||||
def test_existing_empty_directory_no_replace(self):
|
||||
stage=self.root/'stage';target=self.root/'target';stage.mkdir();target.mkdir()
|
||||
with self.assertRaises(OSError):m.publish_no_replace(stage,target)
|
||||
self.assertTrue(stage.is_dir());self.assertTrue(target.is_dir())
|
||||
|
||||
def test_outside_and_alpha_exact_against_independent_baseline(self):
|
||||
palette,raw,frames=self.prepared();result=m.derive(palette,raw,frames)
|
||||
with Image.open(io.BytesIO(raw))as im:parent=np.asarray(im.convert('RGBA'))
|
||||
for col,((asset,out,_),(_,src))in enumerate(zip(result,frames)):
|
||||
baseline=np.asarray(Image.fromarray(parent[:,col*128:(col+1)*128]).resize((128,128),Image.Resampling.LANCZOS))
|
||||
area=np.repeat(np.repeat(m.semantic_mask(asset,src),4,0),4,1)
|
||||
np.testing.assert_array_equal(out[~area],baseline[~area])
|
||||
np.testing.assert_array_equal(out[:,:,3],baseline[:,:,3])
|
||||
ys,xs=np.nonzero(area);a=(ys[0],xs[0]);different=np.abs((baseline[:,:,:3]@m.LUMA)-(baseline[a][:3]@m.LUMA))>2
|
||||
by,bx=np.nonzero(area&different);b=(by[0],bx[0])
|
||||
original_difference=float((baseline[a][:3].astype(float)-baseline[b][:3])@m.LUMA)
|
||||
output_difference=float((out[a][:3].astype(float)-out[b][:3])@m.LUMA)
|
||||
self.assertLessEqual(abs(original_difference-output_difference),1.0000002)
|
||||
def test_gray_index_boundaries_and_exclusive_region_edges(self):
|
||||
for asset,(x0,y0,x1,y1)in m.REGIONS.items():
|
||||
source=np.full((32,32),11,np.uint8);source[y0,x0:x0+4]=[10,11,30,31]
|
||||
mask=m.semantic_mask(asset,source)
|
||||
self.assertEqual(mask[y0,x0:x0+4].tolist(),[False,True,True,False])
|
||||
self.assertTrue(mask[y1-1,x1-1]);self.assertFalse(mask[y1,x1-1]);self.assertFalse(mask[y0,x1])
|
||||
self.assertFalse(mask[y0,x0-1])
|
||||
if y0:self.assertFalse(mask[y0-1,x0])
|
||||
def test_clipping_is_rejected_without_luminance_compression(self):
|
||||
palette,raw,frames=self.prepared()
|
||||
with Image.open(io.BytesIO(raw))as im:a=np.asarray(im.convert('RGB')).copy()
|
||||
a[30,90]=255;stream=io.BytesIO();Image.fromarray(a).save(stream,format='PNG')
|
||||
with self.assertRaisesRegex(ValueError,'clipping'):m.derive(palette,stream.getvalue(),frames)
|
||||
def test_nonneutral_source_anchor_rejected(self):
|
||||
palette,raw,frames=self.prepared();palette[11]=[75,50,30]
|
||||
with self.assertRaisesRegex(ValueError,'near-neutral'):m.derive(palette,raw,frames)
|
||||
def test_non_rgb_parent_and_malformed_frames_rejected(self):
|
||||
p,raw,f=self.prepared()
|
||||
for mode in ['L','P']:
|
||||
stream=io.BytesIO();Image.new(mode,(256,128)).save(stream,format='PNG')
|
||||
with self.assertRaises(ValueError):m.derive(p,stream.getvalue(),f)
|
||||
for bad in [None,7,'203c6',[]]:
|
||||
r=copy.deepcopy(self.recipe);r['frames'][0]=bad
|
||||
with self.assertRaises(ValueError):m.validate_recipe(r)
|
||||
for bad in [None,7,[]]:
|
||||
with self.assertRaises(ValueError):m.read_regular(self.root,bad)
|
||||
def test_fifo_is_nonblocking_and_regular_rejected(self):
|
||||
os.mkfifo(self.root/'pipe')
|
||||
code="import hd_grating_gray_cleanup as m; m.read_regular(__import__('sys').argv[1], 'pipe')"
|
||||
env=dict(os.environ,PYTHONPATH=str(Path(m.__file__).parent))
|
||||
child=subprocess.run([sys.executable,'-c',code,str(self.root)],capture_output=True,timeout=3,env=env)
|
||||
self.assertNotEqual(child.returncode,0);self.assertIn(b'bounded regular input required',child.stderr)
|
||||
def test_oversize_short_and_changed_reads_rejected(self):
|
||||
with self.assertRaisesRegex(ValueError,'bounded regular'):m.read_regular(self.root,'parent.png',limit=2)
|
||||
with mock.patch.object(m.os,'fdopen',return_value=io.BytesIO(b'x')):
|
||||
with self.assertRaisesRegex(ValueError,'changed'):m.read_regular(self.root,'parent.png')
|
||||
real=m.os.fstat;calls=[0]
|
||||
def changed(fd):
|
||||
st=real(fd);calls[0]+=1
|
||||
values={k:getattr(st,k)for k in ['st_mode','st_dev','st_ino','st_size','st_mtime_ns','st_ctime_ns']}
|
||||
if calls[0]%2==0:values['st_mtime_ns']+=1
|
||||
return SimpleNamespace(**values)
|
||||
with mock.patch.object(m.os,'fstat',side_effect=changed):
|
||||
with self.assertRaisesRegex(ValueError,'changed'):m.read_regular(self.root,'parent.png')
|
||||
|
||||
if __name__=='__main__':unittest.main()
|
||||
Reference in new issue
Block a user