"""TileMosaic: bounded 8-bit PNG, integer-translation microscopy mosaic engine. No network, hardware control, diagnostics, rotation correction or generative fill. """ from __future__ import annotations import argparse import base64 import binascii import csv import hashlib import io import json import math import os from pathlib import Path import re import stat import struct import sys import tempfile import zipfile import numpy as np from PIL import Image, __version__ as PILLOW_VERSION SCHEMA = 'tilemosaic/v1' VERSION = '1.2.0' MIN_DIM, MAX_DIM = 16, 2048 MAX_TILES, MAX_TOTAL_PIXELS, MAX_CANVAS_PIXELS = 64, 16_777_216, 32_000_000 MAX_JSON_BYTES, MAX_FILE_BYTES, MAX_UPLOAD_BYTES = 96 * 1024**2, 16 * 1024**2, 64 * 1024**2 MAX_DEPTH, MAX_NODES = 12, 16_800_000 NCC_MIN, PEAK_MARGIN, RESIDUAL_TOL = 0.985, 0.015, 1e-6 class Invalid(ValueError): pass def _json_bytes(value): return json.dumps(value, sort_keys=True, ensure_ascii=True, allow_nan=False, separators=(',', ':')).encode('ascii') def _sha(data): return hashlib.sha256(data).hexdigest() def _bounded(value): stack = [(value, 0)]; count = 0 while stack: obj, depth = stack.pop(); count += 1 if count > MAX_NODES or depth > MAX_DEPTH: raise Invalid('Input node/depth limit exceeded') if isinstance(obj, dict): if any(not isinstance(k, str) or len(k) > 120 for k in obj): raise Invalid('Object keys must be short strings') stack.extend((v, depth + 1) for v in obj.values()) elif isinstance(obj, list): stack.extend((v, depth + 1) for v in obj) elif type(obj) in (int, float): if type(obj) is float and not math.isfinite(obj): raise Invalid('Nonfinite input number') elif isinstance(obj, str): if len(obj) > 1024: raise Invalid('Input string exceeds 1024 characters') elif obj is not None and type(obj) is not bool: raise Invalid('Only JSON values are supported') def _pairs(pairs): out = {} for key, value in pairs: if key in out: raise Invalid('Duplicate JSON key: ' + key[:80]) out[key] = value return out def _constant(value): raise Invalid('Nonfinite JSON constant: ' + value) def _parse(text): if not isinstance(text, str) or len(text) > MAX_JSON_BYTES or len(text.encode('utf-8')) > MAX_JSON_BYTES: raise Invalid('JSON input must be text within 96 MiB') depth = 0; quoted = False; escaped = False for c in text: if quoted: if escaped: escaped = False elif c == '\\': escaped = True elif c == '"': quoted = False elif c == '"': quoted = True elif c in '[{': depth += 1 if depth > MAX_DEPTH: raise Invalid('JSON nesting exceeds 12 levels') elif c in ']}': depth -= 1 try: value = json.loads(text, object_pairs_hook=_pairs, parse_constant=_constant) except (ValueError, RecursionError) as exc: raise Invalid('Invalid JSON: ' + str(exc)[:180]) from exc _bounded(value) return value def _keys(value, allowed, required, where): if not isinstance(value, dict): raise Invalid(where + ' must be an object') extra, missing = set(value) - set(allowed), set(required) - set(value) if extra or missing: raise Invalid(where + ': unexpected keys ' + repr(sorted(extra)) + '; missing ' + repr(sorted(missing))) def _integer(value, lo, hi, where): if type(value) is not int or not lo <= value <= hi: raise Invalid(where + f' must be an integer in [{lo}, {hi}]') return value def _number(value, lo, hi, where): if type(value) not in (int, float) or not lo <= value <= hi or not math.isfinite(value): raise Invalid(where + f' must be finite in [{lo}, {hi}]') return float(value) def _validate(payload): _bounded(payload) _keys(payload, ('schema','grid','overlap','intensity_mode','pixel_scale_um','search_radius_px','tiles','note'), ('schema','grid','overlap','tiles'), 'payload') if payload['schema'] != SCHEMA: raise Invalid('schema must be tilemosaic/v1') _keys(payload['grid'], ('rows','cols'), ('rows','cols'), 'grid') rows = _integer(payload['grid']['rows'], 1, MAX_TILES, 'grid.rows') cols = _integer(payload['grid']['cols'], 1, MAX_TILES, 'grid.cols') if rows * cols > MAX_TILES: raise Invalid('At most 64 tiles are supported') overlap = _number(payload['overlap'], 0.15, 0.75, 'overlap') if payload.get('intensity_mode', 'grayscale') != 'grayscale': raise Invalid('Only grayscale intensity_mode is supported') scale = payload.get('pixel_scale_um') if scale is not None: scale = _number(scale, 1e-9, 1e9, 'pixel_scale_um') note = payload.get('note', '') if not isinstance(note, str) or len(note) > 500: raise Invalid('note must be a string up to 500 characters') tiles = payload['tiles'] if not isinstance(tiles, list) or len(tiles) != rows * cols: raise Invalid('tiles must completely fill the declared rectangular grid') shape = None; seen = set(); converted = []; stage_count = 0 for t in tiles: _keys(t, ('row','col','label','pixels','stage_x_px','stage_y_px'), ('row','col','pixels'), 'tile') r = _integer(t['row'], 0, rows - 1, 'tile.row'); c = _integer(t['col'], 0, cols - 1, 'tile.col') if (r,c) in seen: raise Invalid('Duplicate grid cell') seen.add((r,c)); label = t.get('label', '') if not isinstance(label, str) or len(label) > 120 or any(ord(x) < 32 and x not in '\t\r\n' for x in label): raise Invalid('label must be text up to 120 characters without unsupported control characters') px = t['pixels'] if not isinstance(px, list) or not MIN_DIM <= len(px) <= MAX_DIM or not isinstance(px[0], list) or not MIN_DIM <= len(px[0]) <= MAX_DIM: raise Invalid('Tile dimensions must each be 16..2048 pixels') h, w = len(px), len(px[0]) if h*w*len(tiles) > MAX_TOTAL_PIXELS: raise Invalid('Total decoded pixels exceed 16,777,216') if shape is not None and shape != (h,w): raise Invalid('All tile shapes must match') shape = (h,w) for line in px: if not isinstance(line, list) or len(line) != w: raise Invalid('Pixels must form a rectangular array') if any(type(x) is not int or not 0 <= x <= 255 for x in line): raise Invalid('Pixels must be integer 8-bit values 0..255; booleans are not pixels') item = {'row':r,'col':c,'label':label,'array':np.asarray(px, dtype=np.uint8)} if 'stage_x_px' in t or 'stage_y_px' in t: if not {'stage_x_px','stage_y_px'} <= set(t): raise Invalid('Stage coordinates require both axes') item['stage_x_px'] = _integer(t['stage_x_px'], -100000, 100000, 'stage_x_px') item['stage_y_px'] = _integer(t['stage_y_px'], -100000, 100000, 'stage_y_px'); stage_count += 1 converted.append(item) if stage_count not in (0,len(tiles)): raise Invalid('Supply stage coordinates for every tile or none') radius = payload.get('search_radius_px', max(2, min(16, math.ceil(min(shape)*0.1)))) radius = _integer(radius, 1, 32, 'search_radius_px') return {'rows':rows,'cols':cols,'overlap':overlap,'shape':shape,'scale':scale,'radius':radius, 'tiles':sorted(converted,key=lambda t:(t['row'],t['col'])),'stage':bool(stage_count)} def _integral(a): return np.pad(a.cumsum(0).cumsum(1), ((1,0),(1,0))) def _rect(ii, y0, x0, y1, x1): return ii[y1,x1] - ii[y0,x1] - ii[y1,x0] + ii[y0,x0] def _fft_shape(h,w): return (1 << (2*h-2).bit_length(), 1 << (2*w-2).bit_length()) def _periodic(array): # Exact repeating fields can look unique inside a narrow stage-hint window. # Sample strips cheaply screen candidate periods; full-image equality confirms. for axis in (0,1): length=array.shape[axis] probe=array[:,np.linspace(0,array.shape[1]-1,min(8,array.shape[1]),dtype=int)] if axis==0 else array[np.linspace(0,array.shape[0]-1,min(8,array.shape[0]),dtype=int),:] for shift in range(1,length//2+1): left=[slice(None),slice(None)];right=left.copy();left[axis]=slice(shift,None);right[axis]=slice(None,-shift) if np.array_equal(probe[tuple(left)],probe[tuple(right)]) and np.array_equal(array[tuple(left)],array[tuple(right)]): return True return False def _register(a, b, expected_y, expected_x, radius): """Evaluate EVERY integer shift in the declared search window with exact-overlap NCC. FFT computes cross-products; integral sums normalize the actual overlap, not padding. """ a=a.astype(np.float64);b=b.astype(np.float64);h,w=a.shape if float(a.std()) < 1.0 or float(b.std()) < 1.0: return {'accepted':False,'reason':'blank_or_low_contrast'} ys=np.arange(max(1-h,expected_y-radius),min(h-1,expected_y+radius)+1,dtype=np.int64) xs=np.arange(max(1-w,expected_x-radius),min(w-1,expected_x+radius)+1,dtype=np.int64) if not len(ys) or not len(xs): return {'accepted':False,'reason':'no_plausible_overlap'} dy,dx=np.meshgrid(ys,xs,indexing='ij');dy=dy.ravel();dx=dx.ravel() y0=np.maximum(0,dy);x0=np.maximum(0,dx);y1=np.minimum(h,h+dy);x1=np.minimum(w,w+dx) count=(y1-y0)*(x1-x0) enough=count>=max(64,math.ceil(h*w*0.04)) ia,ib=_integral(a),_integral(b);ia2,ib2=_integral(a*a),_integral(b*b) sa=_rect(ia,y0,x0,y1,x1);sb=_rect(ib,y0-dy,x0-dx,y1-dy,x1-dx) va=np.maximum(0,_rect(ia2,y0,x0,y1,x1)-sa*sa/count) vb=np.maximum(0,_rect(ib2,y0-dy,x0-dx,y1-dy,x1-dx)-sb*sb/count) correction=sa*sb/count # Release four full-size integral images before allocating FFT buffers. del ia,ib,ia2,ib2,sa,sb size=_fft_shape(h,w) spectrum=np.fft.rfft2(a,s=size) other=np.fft.rfft2(b,s=size);np.conjugate(other,out=other) spectrum*=other;del other cross=np.fft.irfft2(spectrum,s=size) cov=cross[dy%size[0],dx%size[1]]-correction del cross,correction valid=enough & (va/count>=1.0) & (vb/count>=1.0) score=np.full(len(dy),-2.0) score[valid]=np.clip(cov[valid]/np.sqrt(va[valid]*vb[valid]),-1,1) order=sorted(np.flatnonzero(valid),key=lambda i:(-round(float(score[i]),12),abs(int(dy[i])-expected_y)+abs(int(dx[i])-expected_x),int(dy[i]),int(dx[i]))) if not order: return {'accepted':False,'reason':'blank_or_insufficient_overlap'} best=order[0];second=order[1] if len(order)>1 else None value=float(score[best]);runner=float(score[second]) if second is not None else None margin=value-runner if runner is not None else 2.0 # Reuse the raw cross-spectrum for phase correlation. It is a diagnostic, # not the overlap-normalized acceptance score. Avoid two redundant FFTs. magnitude=np.abs(spectrum) np.divide(spectrum,magnitude,out=spectrum,where=magnitude>1e-9) spectrum[magnitude<=1e-9]=0 del magnitude phase=np.fft.irfft2(spectrum,s=size) del spectrum result={'dy':int(dy[best]),'dx':int(dx[best]),'ncc':round(value,9),'runner_up_ncc':round(runner,9) if runner is not None else None, 'margin':round(margin,9),'overlap_pixels':int(count[best]),'phase_peak':round(float(phase[dy[best]%size[0],dx[best]%size[1]]),9), 'candidates_tested':len(order)} if value < NCC_MIN: reason='low_match_or_unsupported_geometry' elif margin < PEAK_MARGIN: reason='ambiguous_or_periodic' elif abs(result['dy']-expected_y)==radius or abs(result['dx']-expected_x)==radius: reason='peak_on_search_boundary' else: reason='accepted' result.update(accepted=reason=='accepted',reason=reason) return result def _solve_positions(n, edges): if n==1: return np.zeros((1,2)),0.0 matrix=np.zeros((len(edges),n-1));values=np.zeros((len(edges),2)) for k,(i,j,dy,dx) in enumerate(edges): if i: matrix[k,i-1]=-1 if j: matrix[k,j-1]=1 values[k]=[dy,dx] solution,_,rank,_=np.linalg.lstsq(matrix,values,rcond=None) if rank!=n-1: return None,None residual=float(np.max(np.abs(matrix@solution-values))) return np.vstack((np.zeros((1,2)),solution)),residual def _provenance(payload=None): return {'product':'tilemosaic','version':VERSION,'schema':SCHEMA, 'input_sha256':_sha(_json_bytes(payload)) if payload is not None else None, 'method':'All-window exact-overlap FFT NCC; phase diagnostic; anchored least squares; integer mean render', 'dependencies':{'numpy':np.__version__,'pillow':PILLOW_VERSION}, 'scope':'Materials imagery, fixed magnification, 8-bit grayscale, integer translation only. No professional judgment.'} def _report(status, summary, tables, findings, payload=None): return {'status':status,'summary':summary,'tables':tables,'findings':findings,'provenance':_provenance(payload)} def _compute_inner(payload): try: info=_validate(payload) except (Invalid,ValueError,OverflowError,TypeError,RecursionError) as exc: return _report('FAIL',{}, {},[{'code':'invalid_input','message':str(exc)}]),None h,w=info['shape'];tiles=info['tiles'];rows=info['rows'];cols=info['cols'];n=len(tiles) summary={'tile_count':n,'grid_rows':rows,'grid_cols':cols,'tile_h':h,'tile_w':w,'resolved':False, 'pixel_scale_um':info['scale'],'calibration_status':'USER_SUPPLIED' if info['scale'] is not None else 'UNKNOWN'} measured=[];edges=[] periodic={i for i,t in enumerate(tiles) if n>1 and _periodic(t['array'])} for i,t in enumerate(tiles): for j,kind in ((i+1,'right'),(i+cols,'down')): if (kind=='right' and t['col']==cols-1) or j>=n: continue u=tiles[j] ey=round((1-info['overlap'])*h) if kind=='down' else 0 ex=round((1-info['overlap'])*w) if kind=='right' else 0 if info['stage']: ey=u['stage_y_px']-t['stage_y_px'];ex=u['stage_x_px']-t['stage_x_px'] reg={'accepted':False,'reason':'globally_periodic_tile'} if i in periodic or j in periodic else _register(t['array'],u['array'],ey,ex,info['radius']) row={'from_row':t['row'],'from_col':t['col'],'to_row':u['row'],'to_col':u['col'],'kind':kind, 'expected_dy':ey,'expected_dx':ex,**reg};measured.append(row) if reg['accepted']: edges.append((i,j,reg['dy'],reg['dx'])) summary.update(edges_accepted=len(edges),edges_rejected=len(measured)-len(edges)) if any(not row['accepted'] for row in measured): return _report('UNKNOWN',summary,{'alignment-residuals.csv':measured},[{'code':'unresolved_neighbors','message':'At least one declared neighbor is unobservable, ambiguous, outside the search interior or below the match threshold. No final mosaic was produced.'}],payload),None positions,residual=_solve_positions(n,edges) if positions is None or residual>RESIDUAL_TOL: summary['max_residual_px']=None if residual is None else round(residual,9) return _report('UNKNOWN',summary,{'alignment-residuals.csv':measured},[{'code':'inconsistent_graph','message':'Neighbor translations do not form a connected, cycle-consistent integer layout.'}],payload),None for row,(i,j,dy,dx) in zip(measured,edges): row['residual_dy']=round(float(positions[j,0]-positions[i,0]-dy),9) row['residual_dx']=round(float(positions[j,1]-positions[i,1]-dx),9) integer=np.rint(positions).astype(np.int64) if np.max(np.abs(positions-integer))>RESIDUAL_TOL: return _report('UNKNOWN',summary,{'alignment-residuals.csv':measured},[{'code':'fractional_solution','message':'Subpixel reconstruction is unsupported.'}],payload),None origin=integer.min(axis=0);placed=integer-origin;canvas_h=int(placed[:,0].max()+h);canvas_w=int(placed[:,1].max()+w) if canvas_h*canvas_w>MAX_CANVAS_PIXELS: return _report('UNKNOWN',summary,{'alignment-residuals.csv':measured},[{'code':'canvas_limit','message':'Resolved canvas exceeds 32,000,000 pixels.'}],payload),None acc=np.zeros((canvas_h,canvas_w),np.uint32);coverage=np.zeros((canvas_h,canvas_w),np.uint8) transforms=[] for k,t in enumerate(tiles): y,x=map(int,placed[k]);acc[y:y+h,x:x+w]+=t['array'];coverage[y:y+h,x:x+w]+=1 transforms.append({'row':t['row'],'col':t['col'],'label':t['label'],'y_px':y,'x_px':x, 'anchor_y_px':int(integer[k,0]),'anchor_x_px':int(integer[k,1]),'width_px':w,'height_px':h}) denominator=np.maximum(coverage,1).astype(np.uint32) mosaic=((acc+denominator//2)//denominator).astype(np.uint8) summary.update(resolved=True,mosaic_h=canvas_h,mosaic_w=canvas_w,max_residual_px=round(residual,9), covered_pixels=int(np.count_nonzero(coverage)),uncovered_pixels=int(np.count_nonzero(coverage==0)), canvas_origin_anchor_y_px=int(origin[0]),canvas_origin_anchor_x_px=int(origin[1]), registration_precision='integer pixels',registration_status='COPY_ONLY' if n==1 else 'RESOLVED',output_mode='L',output_bits=8) artifacts={'mosaic':mosaic,'coverage':coverage,'tiles':tiles} return _report('PASS',summary,{'transforms.csv':transforms,'alignment-residuals.csv':measured},[],payload),artifacts def _compute(payload): try: return _compute_inner(payload) except MemoryError: return _report('UNKNOWN',{}, {},[{'code':'resource_exhausted','message':'Insufficient memory for this job. No alignment or final image is claimed.'}]),None def run(payload): return _compute(payload)[0] def run_json(text): try: result=run(_parse(text)) except (Invalid,UnicodeError) as exc: result=_report('FAIL',{}, {},[{'code':'invalid_json','message':str(exc)}]) return _json_bytes(result).decode('ascii') def _csv_bytes(rows): cols=sorted({key for row in rows for key in row});buffer=io.StringIO(newline='') writer=csv.DictWriter(buffer,fieldnames=cols,lineterminator='\n');writer.writeheader() for row in rows: out={} for key in cols: value=row.get(key,'') if isinstance(value,str) and (value.lstrip().startswith(('=','+','-','@')) or value.startswith(('\t','\r','\n'))): value="'"+value out[key]=value writer.writerow(out) return buffer.getvalue().encode('utf-8') def _png(array): out=io.BytesIO();Image.fromarray(np.asarray(array,dtype=np.uint8)).save(out,format='PNG',compress_level=9,optimize=False) return out.getvalue() def _half(array): h,w=array.shape;target=np.zeros(((h+1)//2,(w+1)//2),np.uint32);counts=np.zeros_like(target) for dy in (0,1): for dx in (0,1): chunk=array[dy::2,dx::2];ch,cw=chunk.shape;target[:ch,:cw]+=chunk;counts[:ch,:cw]+=1 return ((target+counts//2)//counts).astype(np.uint8) def _pack(payload, report, art): files={'report.json':_json_bytes(report),'input.json':_json_bytes(payload)} for name,rows in report['tables'].items(): files[name]=_csv_bytes(rows) if art is not None: mosaic=art['mosaic'];coverage=art['coverage'] files['mosaic.png']=_png(mosaic);files['coverage-mask.png']=_png((coverage>0).astype(np.uint8)*255) files['coverage-count.png']=_png(coverage) sources=[] for t in art['tiles']: name=f"sources/r{t['row']}-c{t['col']}.png";files[name]=_png(t['array']) sources.append({'row':t['row'],'col':t['col'],'label':t['label'],'path':name, 'decoded_pixel_sha256':_sha(t['array'].tobytes()),'png_sha256':_sha(files[name])}) files['source-manifest.json']=_json_bytes({'scope':'Exact input grayscale pixels reencoded as PNG; original RGB colors and original file metadata are not retained.','tiles':sources}) scale=report['summary']['pixel_scale_um'] files['calibration.json']=_json_bytes({'pixel_scale_um':scale,'status':report['summary']['calibration_status'], 'scope':'User-supplied isotropic scale, not independently calibrated. Coordinates are pixel top-left positions.', 'width_um':mosaic.shape[1]*scale if scale is not None else None,'height_um':mosaic.shape[0]*scale if scale is not None else None, 'canvas_origin_anchor_x_px':report['summary']['canvas_origin_anchor_x_px'], 'canvas_origin_anchor_y_px':report['summary']['canvas_origin_anchor_y_px']}) levels=[];level=0;current=mosaic while True: h,w=current.shape;paths=[] for y in range(0,h,256): for x in range(0,w,256): name=f'pyramid/level-{level}/{x//256}-{y//256}.png';files[name]=_png(current[y:y+256,x:x+256]);paths.append(name) levels.append({'level':level,'width_px':w,'height_px':h,'downsample':2**level, 'pixel_scale_um':scale*(2**level) if scale is not None else None,'tiles':paths}) if h==1 and w==1: break current=_half(current);level+=1 files['pyramid.json']=_json_bytes({'tile_size':256,'mode':'L','bits':8,'levels':levels, 'downsample':'2x2 arithmetic mean, half-up rounded, actual edge sample count; uncovered pixels remain zero before downsampling.'}) files['README.txt']=(f'TileMosaic {VERSION}; schema {SCHEMA}; status {report["status"]}.\n' 'Scope: materials images, fixed magnification, integer translation, one 8-bit grayscale channel.\n' 'PASS is a bounded computation, not a scientific or professional judgment.\n' 'report.json and CSVs record all neighbor decisions and anchored placements. input.json preserves supplied JSON values.\n' 'PASS bundles include mosaic.png, binary coverage-mask.png, tile-count coverage-count.png, calibration.json,\n' 'decoded source PNGs and source-manifest.json, plus an actual multilevel 256px PNG pyramid and pyramid.json.\n' 'Overlap intensities are arithmetic means, half-up rounded; uncovered pixels are black and identified by the mask.\n' 'Raw grayscale source pixels are retained; RGB input becomes floor((299R+587G+114B+500)/1000).\n' 'Original RGB colors, PNG metadata and original compressed bytes are not retained. Keep original image files.\n' 'UNKNOWN bundles contain observations only, with no final image, placements, pyramid or calibration claim.\n' 'Scale, when supplied, is user-declared micrometers per pixel, never independently validated. No OME-TIFF, TIFF,\n' 'subpixel reconstruction, rotation, perspective, z-stacks, camera connection, viewer or medical use is supported.\n' 'manifest.json hashes every other ZIP member. ZIP and PNG bytes are repeatable in the recorded dependency versions.\n').encode() files['manifest.json']=_json_bytes({'product':'tilemosaic','version':VERSION,'status':report['status'],'sha256':{name:_sha(data) for name,data in sorted(files.items())}}) out=io.BytesIO() with zipfile.ZipFile(out,'w',compression=zipfile.ZIP_STORED) as z: for name,data in sorted(files.items()): entry=zipfile.ZipInfo(name,date_time=(1980,1,1,0,0,0));entry.external_attr=0o600<<16;entry.create_system=3;z.writestr(entry,data) return out.getvalue() def bundle(payload): report,art=_compute(payload) if report['status']=='FAIL': raise Invalid('Cannot bundle invalid input: '+str(report['findings'])) return _pack(payload,report,art) def _filename(name): if not isinstance(name,str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,119}',name) or '..' in name: raise Invalid('Filenames must be unambiguous ASCII basenames, at most 120 characters') def _decode(entry, name, expected): _keys(entry,('encoding','content'),('encoding','content'),name) if entry['encoding']!=expected or not isinstance(entry['content'],str): raise Invalid(name+' must use '+expected+' text encoding') limit=MAX_JSON_BYTES if name=='input.json' else MAX_FILE_BYTES content=entry['content'] if len(content)>limit*4//3+4: raise Invalid(name+' encoded file size exceeds limit') try: raw=base64.b64decode(content,validate=True) if expected=='base64' else content.encode('utf-8') except (ValueError,UnicodeError,binascii.Error) as exc: raise Invalid(name+' has invalid encoding') from exc if len(raw)>limit: raise Invalid(name+' file size exceeds limit') return raw def _png_to_gray(raw,name): if not raw.startswith(b'\x89PNG\r\n\x1a\n'): raise Invalid(name+' must contain PNG bytes') offset=8;ended=False while offset+12<=len(raw): size=struct.unpack('>I',raw[offset:offset+4])[0];kind=raw[offset+4:offset+8] if size>MAX_FILE_BYTES or offset+12+size>len(raw): raise Invalid(name+' has truncated or oversized PNG chunks') offset+=12+size if kind==b'IEND': if size!=0 or offset!=len(raw): raise Invalid(name+' has trailing data or invalid IEND') ended=True;break if not ended: raise Invalid(name+' is missing complete PNG IEND') try: with Image.open(io.BytesIO(raw)) as im: if im.format!='PNG' or im.mode not in ('L','RGB') or getattr(im,'n_frames',1)!=1: raise Invalid(name+' must be one-frame L or RGB PNG') w,h=im.size if not MIN_DIM<=w<=MAX_DIM or not MIN_DIM<=h<=MAX_DIM: raise Invalid(name+' dimensions must be 16..2048 before decoding') im.verify() with Image.open(io.BytesIO(raw)) as im: im.load();array=np.asarray(im) if im.mode=='RGB': rgb=array.astype(np.uint32);array=((299*rgb[:,:,0]+587*rgb[:,:,1]+114*rgb[:,:,2]+500)//1000).astype(np.uint8) return array.tolist() except (OSError,ValueError,SyntaxError,Image.DecompressionBombError) as exc: raise Invalid(name+' unreadable or unsupported PNG: '+str(exc)[:150]) from exc def files_to_payload(files): if not isinstance(files,dict) or not 1<=len(files)<=MAX_TILES+1: raise Invalid('files must contain one input.json, or layout.json plus 1..64 PNG tiles') for name in files: _filename(name) if 'input.json' in files: if set(files)!={'input.json'}: raise Invalid('input.json cannot be mixed with other files') payload=_parse(_decode(files['input.json'],'input.json','utf8').decode('utf-8'));_validate(payload);return payload if 'layout.json' not in files: raise Invalid('Missing layout.json') layout=_parse(_decode(files['layout.json'],'layout.json','utf8').decode('utf-8')) _keys(layout,('schema','grid','overlap','intensity_mode','pixel_scale_um','search_radius_px','tiles','note'),('schema','grid','overlap','tiles'),'layout') if not isinstance(layout['tiles'],list) or not 1<=len(layout['tiles'])<=MAX_TILES: raise Invalid('layout tiles must contain 1..64 records') references=set();converted=[];total=0;pixels=0 for t in layout['tiles']: _keys(t,('row','col','label','file','stage_x_px','stage_y_px'),('row','col','file'),'layout tile') name=t['file'];_filename(name) if not name.endswith('.png') or name in references or name not in files: raise Invalid('Each tile must reference one distinct supplied .png basename') references.add(name);raw=_decode(files[name],name,'base64');total+=len(raw) if total>MAX_UPLOAD_BYTES: raise Invalid('Decoded upload total exceeds 64 MiB') array=_png_to_gray(raw,name);pixels+=len(array)*len(array[0]) if pixels>MAX_TOTAL_PIXELS: raise Invalid('Decoded tile total exceeds 16,777,216 pixels') converted.append({**{k:v for k,v in t.items() if k!='file'},'pixels':array}) if set(files)!=references|{'layout.json'}: raise Invalid('Unexpected unreferenced files') payload={**layout,'tiles':converted};_validate(payload);return payload def demo(): # Integer LCG is defined here only for reproducible demo data; tests use an independent random.Random oracle. state=9381;plane=[] for y in range(48): line=[] for x in range(48): state=(1664525*state+1013904223)&0xffffffff;line.append(20+(state>>24)%216) plane.append(line) return {'schema':SCHEMA,'grid':{'rows':2,'cols':2},'overlap':0.5,'pixel_scale_um':0.5, 'intensity_mode':'grayscale','note':'SYNTHETIC demonstration; not specimen observations.', 'tiles':[{'row':r,'col':c,'label':f'Synthetic {r},{c}','pixels':[line[c*16:c*16+32] for line in plane[r*16:r*16+32]]} for r in range(2) for c in range(2)]} def _plain(path): result=Path(os.path.abspath(path)) if any(p.is_symlink() for p in (result,*result.parents)): raise Invalid('Symlink input/output paths or ancestors are refused') return result def _atomic(path,data): fd,temp=tempfile.mkstemp(prefix='.tilemosaic-',dir=path.parent) try: with os.fdopen(fd,'wb') as f: f.write(data);f.flush();os.fsync(f.fileno()) os.replace(temp,path) finally: if os.path.exists(temp): os.unlink(temp) def main(argv=None): parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('input');parser.add_argument('output') args=parser.parse_args(argv) try: inp=_plain(args.input);out=_plain(args.output) if not inp.is_file() or not stat.S_ISREG(inp.stat().st_mode) or inp.stat().st_size>MAX_JSON_BYTES: raise Invalid('Input must be an ordinary JSON file within 96 MiB') if out.exists() and (not out.is_dir() or any(out.iterdir())): raise Invalid('Output must be absent or an empty ordinary directory; stale outputs are refused') if out==inp or inp in out.parents: raise Invalid('Input and output must be disjoint') with inp.open('rb') as stream: raw=stream.read(MAX_JSON_BYTES+1) payload=_parse(raw.decode('utf-8'));report,art=_compute(payload) archive=_pack(payload,report,art) if report['status']!='FAIL' else None out.mkdir(parents=True,exist_ok=True) _atomic(out/'report.json',_json_bytes(report)) if archive is not None: _atomic(out/'result.zip',archive) print(_json_bytes({'status':report['status'],'output_dir':str(out)}).decode()) return {'PASS':0,'FAIL':2,'UNKNOWN':3}[report['status']] except MemoryError: print(_json_bytes({'status':'UNKNOWN','error':'Insufficient memory to read or package this job; no successful output is claimed.'}).decode());return 3 except (OSError,Invalid,ValueError,UnicodeError,TypeError) as exc: print(_json_bytes({'status':'FAIL','error':str(exc)[:400]}).decode());return 2 if __name__=='__main__': raise SystemExit(main())