@dataclass
Class Tile
tile_id: str
Document_id : str
source: str
Kind:
Int: page
Int: seq
Int
y1:
The path is:
ocr_text: str = ""
title: str ""
def _doc_id_from_source(src: str) -> str:
tail = src.rstrip("/").split("/")[-1] If you want to know more about the src, click here.
tail = re.sub(r".(html?|pdf|png|jpg)$", "", tail, flags=re.I)
Sub(r) return is a re."[^A-Za-z0-9_.-()]+", "_", tail)[:80] or hashlib.md5(src.encode()).hexdigest()[:10]
def _ahash(img, size: int = 8) -> int:
"""64-bit average hash — cheap near-duplicate detection for repeated headers."""
Import numpy as an np
Convert g to img."L").resize((size, size))
A = np.asarray (g, dtype="float32")
bits = (a > a.mean()).flatten()
Out = 0.
For b, in bits
Out = (out)
return bin(a ^ b).count("1")
def _is_informative(img, cfg: Config) -> bool:
"""Reject blank / solid-colour tiles before they ever reach the GPU."""
Numpy can be imported as np
a = np.asarray(img.convert("L"), dtype="float32")
Return float(a.std()) >= cfg.blank_std_threshold
def _save_tile(img, out_dir: Path, name: str) -> str:
out_dir.mkdir(parents=True, exist_ok=True)
p = out_dir/f"{name}.png"
img.convert("RGB").save(p, format="PNG", optimize=True)
Return str (p)
def slice_image_to_tiles(img, cfg: Config, *, doc_id: str, source: str, kind: str,
Page: out_dir, int: path, start_seq, int: 0,
seen_hashes: Optional[List[int]] = None,
Title: str = "") -> List[Tile]:
"""Vertical sliding window with overlap. Used for PDFs and text fallback."""
PIL Import Image
If none else, see_hashes equals seen_hashes []
Image size = width and height
if W!=cfg.tile_width
new_h = max(1, int(H * cfg.tile_width / W))
img = img.resize((cfg.tile_width, new_h))
Size = W x H
step = max(1, cfg.tile_height - cfg.tile_overlap)
tiles: List[Tile] = []
y, seq = 0, start_seq
Start_seq
Breaking News
crop = img.crop((0, y, W, y + h))
if _is_informative(crop, cfg):
Hsh = "_ahash"(crop).
if all(_hamming(hsh, s) > cfg.dedup_hamming for s in seen_hashes):
seen_hashes.append(hsh)
tid = f"{doc_id}__p{page}__t{seq}"
tiles.append(Tile(
tile_id=tid, doc_id=doc_id, source=source, kind=kind, page=page,
seq=seq, y0=y, y1=y + h, title=title,
path=_save_tile(crop, out_dir, tid),
))
seq += 1
Step = y
return tiles
_JS_AUTOSCROLL = """
Async () => {
await new Promise((resolve) => {
let y = 0;
const timer = setInterval(() => {
window.scrollBy(0, 800);
y += 800;
if (y >= document.body.scrollHeight || y > 40000) {
clearInterval(timer);
window.scrollTo(0, 0);
setTimeout(resolve, 250);
}
}, 40);
});
}
"""
_JS_FLATTEN = """
() => {
document.querySelectorAll('*').forEach((el) => s.position === 'sticky') el.style.position = 'absolute';
);
document.querySelectorAll('[role="dialog"], .cookie, #cookie-banner, .cc-banner')
.forEach((el) => el.remove());
}
"""
_CSS_CLEANUP = """
* { animation: none !important; transition: none !important;
scroll-behavior: auto !important; }
html { -webkit-font-smoothing: antialiased; }
Video in iframe[src*="youtube"] { visibility: hidden !important; }
"""
_UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0 Safari/537.36 PixelRAG-Tutorial/1.0")
async def _render_urls_async(urls: List[str], cfg: Config, out_dir: Path) -> List[Tile]:
From playwright.async_api, import async_playwright
PIL Import Image
list of all tiles[Tile] = []
Async playback with async_playwright() Asp:
browser = await pw.chromium.launch(headless=True, args=cfg.headless_args)
ctx = await browser.new_context(
viewport={"width": cfg.tile_width, "height": cfg.tile_height},
device_scale_factor=cfg.device_scale,
user_agent=_UA,
java_script_enabled=True,
)
Use urls to find urls
doc_id = _doc_id_from_source(url)
page = await ctx.new_page()
try:
await page.goto(url, wait_until="domcontentloaded", timeout=cfg.nav_timeout_ms)
try:
await page.wait_for_load_state("networkidle", timeout=12000)
Except Exception
pass
await page.evaluate(_JS_AUTOSCROLL)
await page.add_style_tag(content=_CSS_CLEANUP)
await page.evaluate(_JS_FLATTEN)
title = (await page.title()() or Doc_id
height = await page.evaluate(
"() => Math.max(document.body.scrollHeight, "
"document.documentElement.scrollHeight)")
height = int(min(height, cfg.max_page_height))
step = max(1, cfg.tile_height - cfg.tile_overlap)
seen: List[int] = []
y, seq = 0, 0
If y is 0, then 0 will be the result.
Break Away
buf = await page.screenshot(
full_page=True, type="png",
clip={"x": 0, "y": y, "width": cfg.tile_width, "height": h})
img = Image.open(io.BytesIO(buf)).convert("RGB")
Img.size[0] != cfg.tile_width:
img = img.resize((cfg.tile_width,
max(1, int(img.size[1] * cfg.tile_width / img.size[0]))))
If the image is informative, then _is_informative() will be called.
The _ahash (img) is hsh.
if all(_hamming(hsh, s) > cfg.dedup_hamming for s in seen):
seen.append(hsh)
tid = f"{doc_id}__p0__t{seq}"
all_tiles.append(Tile(
tile_id=tid, doc_id=doc_id, source=url, kind="web",
page=0, seq=seq, y0=y, y1=y + h, title=title,
path=_save_tile(img, out_dir, tid)))
seq += 1
The step y is equal to the y-value.
log.info(" rendered %-34s -> %2d tiles (page %dpx)", doc_id, seq, height)
Exc:
log.warning(" FAILED %s (%s)", url, type(exc).__name__)
finally:
await page.close()
await ctx.close()
await browser.close()
All_tiles return
def render_urls(urls: List[str], cfg: Config, out_dir: Path) -> List[Tile]:
"""Screenshot every URL into tiles; degrade to the text renderer on failure."""
try:
tiles = run_async(_render_urls_async(urls, cfg, out_dir))
if tiles:
return tiles
log.warning("Browser produced no tiles — using text-render fallback.")
Except Exception Excl.
log.warning("Playwright unavailable (%s: %s) — using text-render fallback.",
type(exc).__name__, str(exc)[:160])
Return to the Homepage [t for u in urls for t in render_url_as_text(u, cfg, out_dir)]
def _strip_html(html: str) -> str:
HTML = re.sub."(?is)", " ", html)
HTML = re.sub."(?s)", " ", html)
HTML = re.sub."(?i)(p|div|h[1-6]|li|tr|br)>", "n", html)
Text = re.sub."(?s)]+>", " ", html)
For example, "a" and "b" are both acceptable. [(" ", " "), ("&", "&"), ("", ">"), (""", '"')]:
Text = replace(a,b)
Text = Re.sub(r"[d+]", "", text)
Text = Re.sub(r"[ t]+", " ", text)
Return re.sub."n{2,}", "n", text).strip()
When you use def _mono_font, the size of your font is int (20).
ImageFont can be imported from PIL
For cand in"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
"/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf"):
if os.path.exists(cand):
return ImageFont.truetype(cand, size)
try:
import matplotlib.font_manager as fm
return ImageFont.truetype(fm.findfont("DejaVu Sans"), size)
Except Exception
return ImageFont.load_default()
Def text_to_image (text: str, Config: cfg, title: string = "") -> Any:
"""Render plain text onto a tall white canvas — a browser-free stand-in."""
ImageDraw imports images from PIL
font = _mono_font (20), _mono_font (31)
Wrap = max (20, (cfg.tile_width.80)/11)
Lines:[str] = []
Split text for paragraphs."n"):
para = para.strip()
if you don't para
Continue reading
while len(para) > wrap:
Find = cut (" ", 0, wrap)
cut = cut if cut > 0 else wrap
lines.append(para[:cut])
para = para[cut:].lstrip()
lines.append(para)
lines = lines[:900]
height = pad * 2 + 60 + lh * len(lines)
image = new(img)"RGB", (cfg.tile_width, max(cfg.tile_height, height)), "white")
d = ImageDraw.Draw(img)
d.text((pad, pad), title[:60], font=tfont, fill=(15, 15, 15))
If i is greater than ln, then enumerate the lines:
d.text((pad, pad + 60 + i * lh), ln, font=font, fill=(35, 35, 35))
Return img
def render_url_as_text(url: str, cfg: Config, out_dir: Path) -> List[Tile]:
Import RequestsRequests.get(url; timeout=30); headers =
doc_id = _doc_id_from_source(url)
try:
r = requests.get(url, timeout=30, headers={"User-Agent": _UA})
r.raise_for_status()
Body = "_strip_html"(r.text
Search for m using re.search (r"(?is)(.*?) ", r.text)
Title = m.group(1).strip() if m else doc_id
Except Exception Excl.
log.warning(" fetch failed for %s (%s)", url, type(exc).__name__)
You can return to your original language by clicking here. []
Title = title
log.info(" text-rendered %-30s -> canvas %dpx", doc_id, img.size[1])
return slice_image_to_tiles(img, cfg, doc_id=doc_id, source=url, kind="text",
page=0, out_dir=out_dir, title=title)
def render_pdf(pdf_path: str, cfg: Config, out_dir: Path, dpi: int = 150) -> List[Tile]:
import fitz
PIL Import Image
doc_id = _doc_id_from_source(pdf_path)
tiles: List[Tile] = []
Fitz.open (pdf_path as document):
title = (doc.metadata or {}).get("title"() or Doc_Id
n_pages = doc.page_count
for pno in range(n_pages):
Images = Doc[pno].get_pixmap(dpi=dpi)
Img = Image.frombytes"RGB", (pix.width, pix.height), pix.samples)
tiles += slice_image_to_tiles(img, cfg, doc_id=doc_id, source=pdf_path,
kind="pdf", page=pno, out_dir=out_dir,
title=title)
log.info(" rendered %-34s -> %2d tiles (%d pages)", doc_id, len(tiles), n_pages)
return tiles
def make_synthetic_pdf(path: Path) -> str:
"""A tiny PDF so the tutorial always exercises the PDF path, offline or not."""
import fitz
Body = [
("PixelRAG Internal Note", 22),
("", 12),
("Why pixel-native retrieval?", 16),
("Parsers are per-site glue code. A renderer is one code path for every", 11),
("document type: HTML, PDF, scanned fax, spreadsheet export, dashboard.", 11),
("", 11),
("Tiling policy", 16),
("Tiles are 1024x1024 with 128px of vertical overlap. Overlap keeps a", 11),
("sentence or table row from being split across two embeddings, which is", 11),
("the single biggest source of recall loss in naive screenshot pipelines.", 11),
("", 11),
("Serving", 16),
("FAISS inner-product over L2-normalised vectors equals cosine similarity.", 11),
("Tile scores are max-pooled per document so one strong tile can surface", 11),
("a long page, mirroring late-interaction retrieval behaviour.", 11),
("", 11),
("The mitochondria reference is a joke; the overlap advice is not.", 11),
]
doc = fitz.open()
New page is doc.new_page()
y = 72
For line, body size:
page.insert_text((72, y), line, fontsize=size, fontname="helv")
y += size + 8
doc.save(str(path))
doc.close()
Return str (path)
Apple and Duke Researchers Present a Reinforcement Learning Approach That
