from __future__ import annotations import base64 import io import json import os import random import time from html import escape from pathlib import Path import gradio as gr import numpy as np from PIL import Image try: import spaces except ImportError: class _LocalSpaces: @staticmethod def GPU(*_args, **_kwargs): return lambda function: function spaces = _LocalSpaces() import torch from anypaint import prepare_anypaint from quantized_runtime import build_quantized_pipeline, install_fixed_lora REPO_ID = "yijunwang2/krea2-anypaint" WEIGHT_NAME = "krea2_anypaint_rank32.safetensors" MAX_CANVAS_EDGE = 1280 MAX_SEED = 2**31 - 1 MIN_SCALE = 0.2 MAX_SCALE = 1.0 DEFAULT_RATIO = "16:9" ASPECT_RATIOS = ( ("1:1", 1, 1), ("4:3", 4, 3), ("3:4", 3, 4), ("16:9", 16, 9), ("9:16", 9, 16), ) ASSETS = Path(__file__).resolve().parent / "assets" if os.environ.get("ANYPAINT_UI_ONLY") == "1": pipe = None else: pipe = build_quantized_pipeline(REPO_ID) pipe._anypaint_lora_hooks = install_fixed_lora(pipe.transformer, REPO_ID, WEIGHT_NAME) print(f"[AnyPaint runtime] {json.dumps(pipe._anypaint_runtime, sort_keys=True)}", flush=True) def _round16(value: float) -> int: return max(16, int(round(value / 16.0)) * 16) def _ratio_of(label: str) -> tuple[int, int]: return next(((rw, rh) for name, rw, rh in ASPECT_RATIOS if name == label), (1, 1)) def _default_state(ratio: str = DEFAULT_RATIO) -> dict: return { "ratio": ratio, "offset": 0.5, "offset_cross": 0.5, "scale": 1.0, "tool": "move", "brush_size": 48, "mask_data": "", } def parse_state(value) -> dict: state = _default_state() if isinstance(value, dict): incoming = value elif isinstance(value, str) and value.strip().startswith("{"): try: incoming = json.loads(value) except json.JSONDecodeError: incoming = {} else: incoming = {} state.update({key: incoming[key] for key in state if key in incoming}) if state["ratio"] not in {item[0] for item in ASPECT_RATIOS}: state["ratio"] = DEFAULT_RATIO for key in ("offset", "offset_cross"): try: state[key] = min(1.0, max(0.0, float(state[key]))) except (TypeError, ValueError): state[key] = 0.5 try: state["scale"] = min(MAX_SCALE, max(MIN_SCALE, float(state["scale"]))) except (TypeError, ValueError): state["scale"] = 1.0 if state["tool"] not in {"move", "paint", "erase"}: state["tool"] = "move" try: state["brush_size"] = min(256, max(4, int(state["brush_size"]))) except (TypeError, ValueError): state["brush_size"] = 48 state["mask_data"] = str(state.get("mask_data") or "") return state def plan_canvas( source_width: int, source_height: int, ratio_width: int, ratio_height: int, offset: float, scale: float, offset_cross: float, ) -> tuple[int, int, tuple[int, int, int, int]]: try: offset = min(1.0, max(0.0, float(offset))) except (TypeError, ValueError): offset = 0.5 try: offset_cross = min(1.0, max(0.0, float(offset_cross))) except (TypeError, ValueError): offset_cross = 0.5 try: scale = min(MAX_SCALE, max(MIN_SCALE, float(scale))) except (TypeError, ValueError): scale = 1.0 source_ratio = source_width / source_height target_ratio = ratio_width / ratio_height if target_ratio >= source_ratio: canvas_height = _round16(min(source_height, MAX_CANVAS_EDGE)) full_height = canvas_height full_width = _round16(full_height * source_ratio) canvas_width = max(_round16(canvas_height * target_ratio), full_width) if canvas_width > MAX_CANVAS_EDGE: factor = MAX_CANVAS_EDGE / canvas_width canvas_width = _round16(canvas_width * factor) canvas_height = _round16(canvas_height * factor) full_height = canvas_height full_width = _round16(full_height * source_ratio) else: canvas_width = _round16(min(source_width, MAX_CANVAS_EDGE)) full_width = canvas_width full_height = _round16(full_width / source_ratio) canvas_height = max(_round16(canvas_width / target_ratio), full_height) if canvas_height > MAX_CANVAS_EDGE: factor = MAX_CANVAS_EDGE / canvas_height canvas_height = _round16(canvas_height * factor) canvas_width = _round16(canvas_width * factor) full_width = canvas_width full_height = _round16(full_width / source_ratio) box_width = full_width box_height = full_height if scale < 1.0: box_width = max(16, round(box_width * scale)) box_height = max(16, round(box_width / source_ratio)) box_width = min(box_width, canvas_width) box_height = min(box_height, canvas_height) if target_ratio >= source_ratio: x0 = round((canvas_width - box_width) * offset) y0 = round((canvas_height - box_height) * offset_cross) else: x0 = round((canvas_width - box_width) * offset_cross) y0 = round((canvas_height - box_height) * offset) x1, y1 = x0 + box_width, y0 + box_height x0 = max(0, min(x0, canvas_width - 16)) y0 = max(0, min(y0, canvas_height - 16)) x1 = max(x0 + 16, min(x1, canvas_width)) y1 = max(y0 + 16, min(y1, canvas_height)) return canvas_width, canvas_height, (int(x0), int(y0), int(x1), int(y1)) def _axis_of(source_width: int, source_height: int, ratio_width: int, ratio_height: int) -> str: source_ratio = source_width / source_height target_ratio = ratio_width / ratio_height if abs(target_ratio - source_ratio) < 1e-3: return "none" return "horizontal" if target_ratio > source_ratio else "vertical" def _image_data_uri(image: Image.Image, max_edge: int = 720) -> str: image = image.convert("RGB") if max(image.size) > max_edge: scale = max_edge / max(image.size) image = image.resize( (max(1, round(image.width * scale)), max(1, round(image.height * scale))), Image.Resampling.LANCZOS, ) with io.BytesIO() as buffer: image.save(buffer, format="JPEG", quality=90) return "data:image/jpeg;base64," + base64.b64encode(buffer.getvalue()).decode("ascii") def _mask_data_uri(path: Path) -> str: with Image.open(path) as image, io.BytesIO() as buffer: image.convert("L").save(buffer, format="PNG") return "data:image/png;base64," + base64.b64encode(buffer.getvalue()).decode("ascii") def _space_examples() -> tuple[list[list], dict[str, str]]: manifest_path = ASSETS / "manifest.json" if not manifest_path.is_file(): return [], {} records = json.loads(manifest_path.read_text(encoding="utf-8")) examples = [] states = {} for record in records: source_path = ASSETS / record["source"] mask_path = ASSETS / record["mask"] if not source_path.is_file() or not mask_path.is_file(): continue canvas_width, canvas_height = record["canvas_size"] target_ratio = canvas_width / canvas_height ratio = min(ASPECT_RATIOS, key=lambda item: abs(item[1] / item[2] - target_ratio))[0] state = _default_state(ratio) state["mask_data"] = _mask_data_uri(mask_path) states[record["id"]] = json.dumps(state) examples.append([str(source_path), record["prompt"], record["id"]]) return examples, states SPACE_EXAMPLES, EXAMPLE_STATES = _space_examples() def _mask_from_data(value: str, size: tuple[int, int]) -> Image.Image: if not value.startswith("data:image/"): return Image.new("L", size, 0) try: payload = base64.b64decode(value.split(",", 1)[1]) mask = Image.open(io.BytesIO(payload)).convert("L") except Exception: return Image.new("L", size, 0) if mask.size != size: mask = mask.resize(size, Image.Resampling.NEAREST) return mask.point(lambda pixel: 255 if pixel > 32 else 0) def _stage_html(source: Image.Image | None, state_value: str) -> str: state = parse_state(state_value) if source is None: return ( "
" "Upload an image to start painting or extending the canvas.
" ) source = source.convert("RGB") rw, rh = _ratio_of(state["ratio"]) canvas_width, canvas_height, bbox = plan_canvas( source.width, source.height, rw, rh, state["offset"], state["scale"], state["offset_cross"], ) x0, y0, x1, y1 = bbox axis = _axis_of(source.width, source.height, rw, rh) source_uri = _image_data_uri(source) mask_uri = escape(state["mask_data"], quote=True) ratio_tiles = [] for label, tile_rw, tile_rh in ASPECT_RATIOS: tile_width, tile_height, tile_bbox = plan_canvas( source.width, source.height, tile_rw, tile_rh, state["offset"], state["scale"], state["offset_cross"], ) tile_scale = 58 / max(tile_width, tile_height) tx0, ty0, tx1, ty1 = tile_bbox selected = label == state["ratio"] ratio_tiles.append( f"
" "
" f"
" f"
" f"
{label}
" f"
{tile_width}x{tile_height}
" "
" ) ratio_grid = ( "
" + "".join(ratio_tiles) + "
" ) stage_size = 520.0 display_scale = stage_size / max(canvas_width, canvas_height) display_width = canvas_width * display_scale display_height = canvas_height * display_scale source_left = x0 * display_scale source_top = y0 * display_scale source_width = (x1 - x0) * display_scale source_height = (y1 - y0) * display_scale if axis == "horizontal": anchors = (("Left", 0.0), ("Center", 0.5), ("Right", 1.0)) elif axis == "vertical": anchors = (("Top", 0.0), ("Center", 0.5), ("Bottom", 1.0)) else: anchors = () anchor_buttons = "".join( f"" for name, value in anchors ) anchor_row = ( f"
{anchor_buttons}
" if anchors else "" ) _, _, full_bbox = plan_canvas( source.width, source.height, rw, rh, state["offset"], 1.0, state["offset_cross"] ) full_width = full_bbox[2] - full_bbox[0] full_height = full_bbox[3] - full_bbox[1] def tool_button(label: str, value: str) -> str: active = state["tool"] == value return ( f"" ) tools = ( "
" + tool_button("Move", "move") + tool_button("Paint", "paint") + tool_button("Erase", "erase") + "" + "
" ) return f"""
{tools}
×
{anchor_row}
Canvas {canvas_width}x{canvas_height} ยท source ({x0},{y0})-({x1},{y1})
source generated
{ratio_grid}
""" def _placed_canvas( source: Image.Image, canvas_size: tuple[int, int], bbox: tuple[int, int, int, int], ) -> Image.Image: values = np.asarray(source.convert("RGB")) fill = tuple(np.median(values.reshape(-1, 3), axis=0).round().astype(np.uint8)) canvas = Image.new("RGB", canvas_size, fill) resized = source.convert("RGB").resize( (bbox[2] - bbox[0], bbox[3] - bbox[1]), Image.Resampling.LANCZOS ) canvas.paste(resized, bbox[:2]) return canvas @spaces.GPU(duration=90) def generate(source, prompt, state_value, seed, randomize_seed, invert_mask): request_started = time.perf_counter() if source is None: raise gr.Error("Upload a source image first.") if pipe is None: raise gr.Error("The local UI-only preview does not load model weights.") source = source.convert("RGB") prompt = str(prompt or "").strip() or "a high definition image, complete coherent composition" state = parse_state(state_value) rw, rh = _ratio_of(state["ratio"]) canvas_width, canvas_height, bbox = plan_canvas( source.width, source.height, rw, rh, state["offset"], state["scale"], state["offset_cross"], ) brush = _mask_from_data(state["mask_data"], (canvas_width, canvas_height)) if invert_mask: brush = Image.fromarray(255 - np.asarray(brush, dtype=np.uint8), mode="L") prepared = prepare_anypaint(source, brush, (canvas_width, canvas_height), bbox) prepared_at = time.perf_counter() actual_seed = random.randint(0, MAX_SEED) if randomize_seed else int(seed) result = pipe( prompt=prompt, image=prepared.condition, width=canvas_width, height=canvas_height, num_inference_steps=8, guidance_scale=0.0, generator=torch.Generator(device="cpu").manual_seed(actual_seed), reference_max_pixels=384 * 384, reference_placements=[prepared.reference_placement], encode_reference_in_prompt=True, kv_cache=True, known_image=prepared.known_image, known_mask=prepared.keep_mask, ).images[0] generated_at = time.perf_counter() print( "[AnyPaint timing] " f"gpu={torch.cuda.get_device_name(0)} " f"prepare={prepared_at - request_started:.3f}s " f"pipeline={generated_at - prepared_at:.3f}s " f"function={generated_at - request_started:.3f}s", flush=True, ) return result, actual_seed def reset_on_upload(image, state_value): ratio = parse_state(state_value)["ratio"] state = json.dumps(_default_state(ratio)) return _stage_html(image, state), state, gr.update(visible=image is None) def refresh_stage(image, state_value): return _stage_html(image, state_value) def clear_source(state_value): ratio = parse_state(state_value)["ratio"] state = json.dumps(_default_state(ratio)) return _stage_html(None, state), state, gr.update(value=None, visible=True) def run_example(image, text, example_id): state_value = EXAMPLE_STATES[str(example_id)] result, actual_seed = generate(image, text, state_value, 42, True, False) return result, actual_seed, state_value, _stage_html(image, state_value), gr.update(visible=False) CSS = """ #ap-container { max-width:1180px; margin:0 auto; } .dark .gradio-container { color:var(--body-text-color); } #ap-stage-wrap { min-height:40px; } #ap-state { position:absolute!important; width:0; height:0; overflow:hidden; padding:0; margin:0; border:0; opacity:0; pointer-events:none; } #ap-clear-source { position:absolute!important; width:0; height:0; overflow:hidden; padding:0; margin:0; border:0; opacity:0; pointer-events:none; } """ CANVAS_JS = r""" () => { const root = () => { const app=document.querySelector('gradio-app'); return app&&app.shadowRoot?app.shadowRoot:document; }; const stateBox = () => root().querySelector('#ap-state textarea'); const readState = () => { let s={ratio:'16:9',offset:.5,offset_cross:.5,scale:1,tool:'move',brush_size:48,mask_data:''}; const b=stateBox(); if(b&&b.value){try{Object.assign(s,JSON.parse(b.value));}catch(e){}} return s; }; const writeState = (s) => { const b=stateBox(); if(!b)return; const set=Object.getOwnPropertyDescriptor(HTMLTextAreaElement.prototype,'value').set; set.call(b,JSON.stringify(s)); b.dispatchEvent(new Event('input',{bubbles:true})); b.dispatchEvent(new Event('change',{bubbles:true})); }; window._apSelectRatio = (ratio) => { const s=readState(); s.ratio=ratio; s.mask_data=''; writeState(s); }; window._apSetOffset = (offset) => { const s=readState(); s.offset=Math.min(1,Math.max(0,offset)); writeState(s); }; window._apClearSource = () => { const b=root().querySelector('#ap-clear-source button')||root().querySelector('#ap-clear-source'); if(b)b.click(); }; window._apSetTool = (tool) => { const st=root().querySelector('#ap-stage'); if(!st)return; st.dataset.tool=tool; root().querySelectorAll('[data-tool]').forEach(b=>{const active=b.dataset.tool===tool;b.style.border=active?'2px solid var(--color-accent)':'1px solid var(--border-color-primary)';b.style.color=active?'var(--color-accent)':'var(--body-text-color)';}); syncPointer(st); }; window._apSetBrushSize = (value) => { const st=root().querySelector('#ap-stage'); if(st)st.dataset.brush=value; const out=root().querySelector('#ap-size-value'); if(out)out.value=value; }; const syncPointer = (st) => { const c=st.querySelector('#ap-mask'),img=st.querySelector('#ap-source'),h=st.querySelector('#ap-resize'); const move=st.dataset.tool==='move'; c.style.pointerEvents=move?'none':'auto'; c.style.cursor=move?'default':'crosshair'; img.style.cursor=move?'grab':'default'; h.style.display=move?'block':'none'; }; const attach = () => { const st=root().querySelector('#ap-stage'); if(!st||st.dataset.bound==='1')return; st.dataset.bound='1'; const img=st.querySelector('#ap-source'), canvas=st.querySelector('#ap-mask'), seed=st.querySelector('#ap-mask-seed'), handle=st.querySelector('#ap-resize'), remove=st.querySelector('#ap-remove'); const cw=+st.dataset.cw,ch=+st.dataset.ch,buffer=document.createElement('canvas'); buffer.width=cw;buffer.height=ch; const bctx=buffer.getContext('2d'),ctx=canvas.getContext('2d'); const render=()=>{ctx.clearRect(0,0,cw,ch);ctx.save();ctx.globalAlpha=.42;ctx.fillStyle='#f59e0b';ctx.globalCompositeOperation='source-over';ctx.drawImage(buffer,0,0);ctx.globalCompositeOperation='source-in';ctx.fillRect(0,0,cw,ch);ctx.restore();}; if(seed&&seed.src){const load=()=>{bctx.clearRect(0,0,cw,ch);bctx.drawImage(seed,0,0,cw,ch);const data=bctx.getImageData(0,0,cw,ch);for(let i=0;i32||data.data[i+1]>32||data.data[i+2]>32;data.data[i]=255;data.data[i+1]=255;data.data[i+2]=255;data.data[i+3]=on?255:0;}bctx.putImageData(data,0,0);render();};if(seed.complete)load();else seed.onload=load;} const commitMask=()=>{const s=readState();s.tool=st.dataset.tool;s.brush_size=+st.dataset.brush;s.mask_data=buffer.toDataURL('image/png');writeState(s);}; window._apClearMask=()=>{bctx.clearRect(0,0,cw,ch);render();commitMask();}; const point=e=>{const r=canvas.getBoundingClientRect(),p=e.touches?e.touches[0]:e;return[(p.clientX-r.left)*cw/r.width,(p.clientY-r.top)*ch/r.height];}; let drawing=false,last=null; const stroke=e=>{if(!drawing)return;const p=point(e),size=+st.dataset.brush||48;bctx.save();bctx.globalCompositeOperation=st.dataset.tool==='erase'?'destination-out':'source-over';bctx.strokeStyle='white';bctx.fillStyle='white';bctx.lineWidth=size;bctx.lineCap='round';bctx.beginPath();bctx.moveTo(...(last||p));bctx.lineTo(...p);bctx.stroke();bctx.beginPath();bctx.arc(p[0],p[1],size/2,0,Math.PI*2);bctx.fill();bctx.restore();last=p;render();e.preventDefault();}; canvas.addEventListener('pointerdown',e=>{if(st.dataset.tool==='move')return;drawing=true;last=point(e);canvas.setPointerCapture(e.pointerId);stroke(e);});canvas.addEventListener('pointermove',stroke);canvas.addEventListener('pointerup',e=>{if(!drawing)return;drawing=false;last=null;commitMask();}); const syncHandles=()=>{handle.style.left=(img.offsetLeft+img.offsetWidth-9)+'px';handle.style.top=(img.offsetTop+img.offsetHeight-9)+'px';remove.style.left=(img.offsetLeft-11)+'px';remove.style.top=(img.offsetTop-11)+'px';};syncHandles();syncPointer(st); const commitPlacement=()=>{const freeX=st.clientWidth-img.offsetWidth,freeY=st.clientHeight-img.offsetHeight,axis=st.dataset.axis;const s=readState();const ox=freeX>0?img.offsetLeft/freeX:.5,oy=freeY>0?img.offsetTop/freeY:.5;s.offset=axis==='vertical'?oy:ox;s.offset_cross=axis==='vertical'?ox:oy;const liveScale=st.clientWidth/cw,fullW=+st.dataset.fullw*liveScale,fullH=+st.dataset.fullh*liveScale;s.scale=Math.min(1,Math.max(.2,axis==='vertical'?img.offsetWidth/fullW:img.offsetHeight/fullH));s.tool=st.dataset.tool;s.brush_size=+st.dataset.brush;s.mask_data=buffer.toDataURL('image/png');writeState(s);}; let drag=false,sx=0,sy=0,sl=0,stp=0;img.addEventListener('pointerdown',e=>{if(st.dataset.tool!=='move')return;drag=true;sx=e.clientX;sy=e.clientY;sl=img.offsetLeft;stp=img.offsetTop;img.setPointerCapture(e.pointerId);img.style.cursor='grabbing';e.preventDefault();});img.addEventListener('pointermove',e=>{if(!drag)return;img.style.left=Math.min(Math.max(sl+e.clientX-sx,0),Math.max(0,st.clientWidth-img.offsetWidth))+'px';img.style.top=Math.min(Math.max(stp+e.clientY-sy,0),Math.max(0,st.clientHeight-img.offsetHeight))+'px';syncHandles();});img.addEventListener('pointerup',()=>{if(!drag)return;drag=false;img.style.cursor='grab';commitPlacement();}); let resizing=false,rx=0,rw=0,rh=0;handle.addEventListener('pointerdown',e=>{resizing=true;rx=e.clientX;rw=img.offsetWidth;rh=img.offsetHeight;handle.setPointerCapture(e.pointerId);e.preventDefault();});handle.addEventListener('pointermove',e=>{if(!resizing)return;let nw=Math.max(32,rw+e.clientX-rx),nh=nw*rh/rw;if(img.offsetLeft+nw>st.clientWidth){nw=st.clientWidth-img.offsetLeft;nh=nw*rh/rw;}if(img.offsetTop+nh>st.clientHeight){nh=st.clientHeight-img.offsetTop;nw=nh*rw/rh;}img.style.width=nw+'px';img.style.height=nh+'px';syncHandles();});handle.addEventListener('pointerup',()=>{if(!resizing)return;resizing=false;commitPlacement();}); }; const observer=new MutationObserver(attach);observer.observe(document.body,{childList:true,subtree:true});const app=document.querySelector('gradio-app');if(app&&app.shadowRoot)observer.observe(app.shadowRoot,{childList:true,subtree:true});attach();setInterval(attach,800); } """ with gr.Blocks(title="Krea 2 AnyPaint") as demo: with gr.Column(elem_id="ap-container"): gr.Markdown( "# Krea 2 AnyPaint\n" "Arbitrary-mask inpainting, outpainting, and image editing for Krea 2 Turbo." ) state = gr.Textbox( value=json.dumps(_default_state()), show_label=False, container=False, elem_id="ap-state", ) preview_record = SPACE_EXAMPLES[0] if os.environ.get("ANYPAINT_UI_EXAMPLE") == "1" and SPACE_EXAMPLES else None preview_image = Image.open(preview_record[0]).convert("RGB") if preview_record else None preview_state = EXAMPLE_STATES[preview_record[2]] if preview_record else state.value with gr.Row(): with gr.Column(scale=1): prompt = gr.Textbox( label="Prompt", lines=2, placeholder="Describe the complete desired output image", ) source = gr.Image( value=preview_image, type="pil", label="Source image", height=300, visible=preview_image is None, ) stage = gr.HTML(_stage_html(preview_image, preview_state), elem_id="ap-stage-wrap") clear_source_button = gr.Button("clear source", elem_id="ap-clear-source") run = gr.Button("Generate", variant="primary") with gr.Column(scale=1): result = gr.Image(label="AnyPaint result", interactive=False) with gr.Accordion("Advanced settings", open=False): with gr.Row(): seed = gr.Number(value=42, precision=0, label="Seed") randomize_seed = gr.Checkbox(value=True, label="Randomize seed") invert_mask = gr.Checkbox(value=False, label="Invert painted mask") example_id = gr.Textbox(label="Case", visible=False) if SPACE_EXAMPLES: gr.Examples( examples=SPACE_EXAMPLES, inputs=[source, prompt, example_id], outputs=[result, seed, state, stage, source], fn=run_example, cache_examples=False, run_on_click=True, ) source.change(reset_on_upload, [source, state], [stage, state, source]) source.clear(clear_source, [state], [stage, state, source]) clear_source_button.click(clear_source, [state], [stage, state, source]) state.change(refresh_stage, [source, state], [stage], show_progress="hidden") run.click( generate, [source, prompt, state, seed, randomize_seed, invert_mask], [result, seed], api_name="generate", ) demo.load(fn=None, js=CANVAS_JS) if __name__ == "__main__": demo.queue(max_size=20).launch(theme=gr.themes.Citrus(), css=CSS)