AskaEth/ppl2psd
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add --scale upscaling and stream encode to avoid OOM

Browse Source
This commit is contained in:
AskaEth committed 2026-08-29 00:37:08 +08:00
1 parent a08151337b
commit 6a97e68558
2 files changed
+127 -125
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

README.md
+5 -1
View File
@@ -46,15 +46,19 @@ python3 ppl2psd.py input.ppl -o out.psd --preview out.png
# PPL → PSD（文字转可编辑 TypeLayer）
python3 ppl2psd.py input.ppl -o out.psd --type-layer
# PPL → PSD（整个矢量项目等比放大 2 倍，画布/图层/文字/阴影 ×2）
python3 ppl2psd.py input.ppl -o out_2x.psd --scale 2
# PSD → PPL（反向）
python3 psd2ppl.py input.psd -o out.ppl
```
参数：
- `-o` / `--output`：输出路径（默认与输入同名）
- `--preview`：同时输出合成预览 PNG
- `--preview`：同时输出合成预览 PNG（`--scale` 非 1 时输出全分辨率预览）
- `-j` / `--workers`：并行编码线程数（默认 4）
- `--type-layer`：文字转 TypeLayer
- `--scale`：整个矢量项目等比放大倍数（默认 1.0；≥2 时自动降为单进程编码防爆内存）
---
ppl2psd.py
+122 -124
View File
@@ -28,6 +28,9 @@ from pathlib import Path
from typing import List, Optional, Tuple
from PIL import Image, ImageDraw, ImageFont
# We render huge 2x canvases (e.g. 21528x10764); disable PIL's decompression
# bomb guard so it does not reject our own output.
Image.MAX_IMAGE_PIXELS = None
from tqdm import tqdm
try:
@@ -44,6 +47,15 @@ try:
except ImportError: # pragma: no cover
_TYSHA = False
# Global output scale: everything (canvas, layer rects, text size, shadows)
# is multiplied by _SCALE, so the whole vector project is upscaled.
_SCALE = 1.0
def set_scale(s: float) -> None:
global _SCALE
_SCALE = max(0.1, float(s))
# --------------------------------------------------------------------------
# Fonts
# --------------------------------------------------------------------------
@@ -164,9 +176,9 @@ def parse_gradient(spec: str) -> dict:
# Layer rendering
# --------------------------------------------------------------------------
def bounding_rect(ly: dict) -> Tuple[float, float, float, float]:
"""boundingRect is (left, top, right, bottom), not (x, y, w, h)."""
"""boundingRect is (left, top, right, bottom), scaled by _SCALE."""
l, t, r, b = (float(v) for v in ly["boundingRect"].split(","))
return l, t, r, b
return l * _SCALE, t * _SCALE, r * _SCALE, b * _SCALE
def crop_to_canvas(l: float, t: float, r: float, b: float, cw: int, ch: int
@@ -254,7 +266,7 @@ def render_shape(ly: dict, cw: int, ch: int) -> Optional[Tuple[Image.Image, Tupl
img.paste(grad_img, (0, 0), mask)
# stroke
thickness = int(float(ly.get("shapeStrokeThickness", "0") or 0))
thickness = int(float(ly.get("shapeStrokeThickness", "0") or 0) * _SCALE)
if thickness > 0:
stroke = parse_gradient(ly.get("shapeStrokeColor", ""))
if stroke["type"] == "solid":
@@ -269,8 +281,8 @@ def render_shape(ly: dict, cw: int, ch: int) -> Optional[Tuple[Image.Image, Tupl
img.paste(stroke_img, (0, 0), ring)
# shadow (ppl renders a soft offset shadow per layer)
sd = float(ly.get("shadowDistance", "0") or 0)
sb = float(ly.get("shadowBlur", "0") or 0)
sd = float(ly.get("shadowDistance", "0") or 0) * _SCALE
sb = float(ly.get("shadowBlur", "0") or 0) * _SCALE
if (sd > 0 or sb > 0) and ly.get("_no_shadow") != "1":
img, ox, oy = _add_shadow(img, sd, sb,
float(ly.get("shadowAngle", "0") or 0),
@@ -325,7 +337,7 @@ def render_text(ly: dict, cw: int, ch: int) -> Optional[Tuple[Image.Image, Tuple
size_raw = float(ly.get("textSize", "12") or 12)
font_path = _font_path_for(ly)
prof = font_profile(font_path)
size = size_raw * prof["k"]
size = size_raw * prof["k"] * _SCALE
if font_path:
font = ImageFont.truetype(font_path, int(size))
else:
@@ -437,8 +449,8 @@ def render_text(ly: dict, cw: int, ch: int) -> Optional[Tuple[Image.Image, Tuple
img.paste(gimg, (0, 0), img.getchannel("A"))
# soft shadow (used by rasterized output / previews)
sd = float(ly.get("shadowDistance", "0") or 0)
sb = float(ly.get("shadowBlur", "0") or 0)
sd = float(ly.get("shadowDistance", "0") or 0) * _SCALE
sb = float(ly.get("shadowBlur", "0") or 0) * _SCALE
if (sd > 0 or sb > 0) and ly.get("_no_shadow") != "1":
img, ox, oy = _add_shadow(img, sd, sb,
float(ly.get("shadowAngle", "0") or 0),
@@ -514,42 +526,32 @@ def channel_blocks(img: Image.Image, workers: int,
progress: Optional[tqdm] = None) -> List[Tuple[int, bytes]]:
"""RLE-encode the planes of a cropped RGB(A) layer (parallel).
RGBA layers yield channel ids [-1, 0, 1, 2], RGB images yield [0, 1, 2]."""
arr = np.asarray(img) if np is not None else None
if arr is not None and arr.ndim == 3:
nch = arr.shape[2]
if nch == 4:
# zero RGB where alpha==0 for better compression
arr = arr.copy()
arr[arr[..., 3] == 0] = 0
# PSD layer channel ids: -1=alpha mask, 0=R, 1=G, 2=B
planes = [arr[..., 3].tobytes()] + [arr[..., i].tobytes() for i in range(3)]
else:
planes = [arr[..., i].tobytes() for i in range(nch)]
else:
bands = img.getbands()
if bands == ("R", "G", "B", "A"):
planes = [img.getchannel("A").tobytes()] + \
[img.getchannel(c).tobytes() for c in "RGB"]
else:
planes = [img.getchannel(c).tobytes() for c in bands]
RGBA layers yield channel ids [-1, 0, 1, 2], RGB images yield [0, 1, 2].
Bands are encoded one at a time and their raw bytes released right away,
so at most two full-image buffers (the raw plane + its RLE result) exist
at once. On 2x canvases the old copy-all-planes approach multiplied the
layer size by ~5 and blew up RAM/ZRAM/SWAP, freezing the phone."""
w = img.width
jobs = [(plane, w) for plane in planes]
if workers > 1 and len(jobs) > 1:
ids = [-1, 0, 1, 2] if img.mode == "RGBA" else [0, 1, 2]
order = ["A", "R", "G", "B"] if img.mode == "RGBA" else ["R", "G", "B"]
if workers > 1 and len(order) > 1:
from multiprocessing import Pool
with Pool(workers) as pool:
enc = []
for res in pool.imap(_pack_channel_worker, jobs, chunksize=1):
enc.append(res)
for band in order:
plane = img.getchannel(band).tobytes()
enc.append(pool.apply(_pack_channel_worker, ((plane, w),)))
del plane
if progress:
progress.update(1)
else:
enc = []
for j in jobs:
enc.append(_pack_channel_worker(j))
for band in order:
plane = img.getchannel(band).tobytes()
enc.append(_pack_channel_worker((plane, w)))
del plane
if progress:
progress.update(1)
ids = [-1, 0, 1, 2] if img.mode == "RGBA" else [0, 1, 2]
return list(zip(ids, enc))
@@ -594,7 +596,7 @@ def build_type_layer_tysh(ly: dict, font_ps: str, text_w: Optional[float] = None
pass
prof = font_profile(_font_path_for(ly))
size_px = float(ly.get("textSize", "12") or 12) * prof["k"]
size_px = float(ly.get("textSize", "12") or 12) * prof["k"] * _SCALE
ssd = eng["StyleRun"]["RunArray"][0]["StyleSheet"]["StyleSheetData"]
ssd["FontSize"] = Float(size_px)
@@ -710,71 +712,20 @@ def _layer_record(top, left, bottom, right, name, opacity, visible, channels,
return bytes(rec)
def build_psd(layers, width: int, height: int, out_path: str, workers: int,
progress: Optional[tqdm] = None) -> None:
"""layers: list of (name, opacity, visible, offset, Image[, tysh[, lsct]])
bottom->top. img=None marks a group divider layer (lsct present); it gets
empty channels (2-byte compression markers only)."""
# encode all channels (parallel)
chan_data = []
for layer in layers:
img = layer[4]
if img is None:
chan_data.append(None)
else:
ch = channel_blocks(img, workers, progress)
chan_data.append(ch)
def build_psd(records: bytearray, payload_f, merged: Image.Image, layers: list,
width: int, height: int, out_path: str) -> None:
"""Assemble the PSD file from pre-encoded pieces.
# layer records + channel payloads
records = bytearray()
payload = bytearray()
for li, layer in enumerate(layers):
name, opacity, visible, (lx, ly), img = layer[:5]
tysh = layer[5] if len(layer) > 5 else None
lsct = layer[6] if len(layer) > 6 else None
if img is None:
# group divider: 4 empty channels (RLE marker only, len 2)
ch = [(cid, b"\x00\x01") for cid in (-1, 0, 1, 2)]
l = t = r = b = 0
if lsct and lsct[1]:
cr = crop_to_canvas(*lsct[1], width, height)
if cr:
l, t, r, b = cr
records += _layer_record(t, l, b, r, name, opacity, visible, ch,
layer_id=li + 1,
lsct=lsct[0] if lsct else None)
for _, p in ch:
payload += p
continue
ch = chan_data[li]
l, t = lx, ly
r = l + img.width
b = t + img.height
records += _layer_record(t, l, b, r, name, opacity, visible, ch,
layer_id=li + 1, tysh=tysh, lsct=lsct)
for _, p in ch:
payload += p
# merged preview: composite visible layers over transparent
merged = Image.new("RGBA", (width, height), (0, 0, 0, 0))
for layer in layers:
name, opacity, visible, (lx, ly), img = layer[:5]
if img is not None and visible and opacity > 0:
_opacity_paste(merged, img, (lx, ly), opacity)
merged_rgb = None # keep the RGBA merged image for the 4-channel output
# Merged image data is written RAW with a single compression field and
# all planes in planar order (R G B A). 4 channels because the document
# has transparency (negative layer count references the merged alpha).
if np is not None:
arr = np.asarray(merged)
merged_block = b"\x00\x00" + b"".join(
arr[..., i].tobytes() for i in range(4))
else:
merged_block = b"\x00\x00" + b"".join(
merged.getchannel(c).tobytes() for c in "RGBA")
# assemble file. Header channels = 4 (document has transparency, like PS
# writes when layers have alpha).
The renderer already streamed every layer's RLE payloads into `payload_f`
(temp file) and composited the visible layers onto `merged`, so this only
writes header + records + streamed payloads + merged image. Nothing is
loaded back into RAM."""
import shutil
payload_f.seek(0, 2)
payload_len = payload_f.tell()
payload_f.seek(0)
# Header: channels=4 (document has transparency, like PS writes when
# layers have alpha)
fh = open(out_path, "wb")
fh.write(struct.pack(">4sH6sHIIHH", b"8BPS", 1, b"\x00" * 6, 4, height, width, 8, 3))
fh.write(struct.pack(">I", 0)) # color mode data
@@ -782,17 +733,23 @@ def build_psd(layers, width: int, height: int, out_path: str, workers: int,
# Layer info section: [Length(4)] + [layer count(2) + records + channel data]
# Negative layer count: abs = layer count, first alpha channel of the
# merged image holds the merged transparency (PS writes this when the
# document has transparent layers).
layer_info_body = struct.pack(">h", -len(layers)) + bytes(records) + bytes(payload)
layer_info = struct.pack(">I", len(layer_info_body)) + layer_info_body
# layer & mask section: length = layer info + global layer mask info
# (the outer length does NOT include its own 4-byte length field)
lam = layer_info + struct.pack(">I", 0) # global layer mask info
fh.write(struct.pack(">I", len(lam)) + lam)
# merged image holds the merged transparency.
body_len = 2 + len(records) + payload_len
# layer & mask section: length = [4-byte inner length] + body +
# global layer mask info (4); the outer length does NOT include its own
# 4-byte length field
fh.write(struct.pack(">I", 4 + body_len + 4))
fh.write(struct.pack(">I", body_len))
fh.write(struct.pack(">h", -len(layers)))
fh.write(bytes(records))
shutil.copyfileobj(payload_f, fh) # stream, never load payloads into RAM
fh.write(struct.pack(">I", 0)) # global layer mask info
# merged image data (RGB, RAW)
fh.write(merged_block)
# merged image data (RGB, RAW), one plane at a time
fh.write(b"\x00\x00")
for c in "RGBA":
fh.write(merged.getchannel(c).tobytes())
del merged
fh.close()
@@ -810,6 +767,10 @@ def _safe_workers(workers: int) -> int:
def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None,
workers: int = 4, type_layer: bool = False) -> None:
workers = _safe_workers(workers)
if _SCALE >= 2.0:
# 2x canvases are 4x the pixels; parallel channel encoding would
# duplicate layer buffers across processes and exhaust RAM/ZRAM/SWAP
workers = 1
import tempfile
with tempfile.TemporaryDirectory() as tmp:
with zipfile.ZipFile(ppl_path) as zf:
@@ -819,8 +780,8 @@ def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None,
layers, groups = parse_layers(text)
mw = re.search(r"^width=(\d+)", text, re.M)
mh = re.search(r"^height=(\d+)", text, re.M)
width = int(mw.group(1)) if mw else 0
height = int(mh.group(1)) if mh else 0
width = int(float(mw.group(1)) * _SCALE) if mw else 0
height = int(float(mh.group(1)) * _SCALE) if mh else 0
# --- group mapping ---
group_map = {}
@@ -866,10 +827,21 @@ def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None,
rendered = []
skipped = 0
# Streaming render+encode: each layer is rendered, composited onto
# the merged canvas, RLE-encoded into the payload temp file and then
# released. Only ~1 layer + the merged canvas stay in RAM, instead of
# all 142 layers' images (the 2x build previously exhausted RAM and
# swap and got OOM-killed).
merged = Image.new("RGBA", (width, height), (0, 0, 0, 0))
payload_f = tempfile.TemporaryFile()
records = bytearray()
rp = tqdm(total=len(layers), desc="渲染图层", unit="层", file=sys.stderr)
ep = tqdm(total=len(layers) * 4, desc="编码通道", unit="通道",
file=sys.stderr)
for i, ly in enumerate(layers):
gid = ly.get("groupId")
gname = gvis = gb = None
layer_id = len(rendered) + 1
if gid and gid in group_ranges:
g = group_map.get(gid, {})
gname = g.get("name", "") or f"group_{gid}"
@@ -880,6 +852,16 @@ def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None,
if group_ranges[gid][0] == i:
rendered.append((gname, 255, gvis, (0, 0), None, None,
(3, gb)))
# group open marker: 4 empty channels
ch = [(cid, b"\x00\x01") for cid in (-1, 0, 1, 2)]
cr = crop_to_canvas(*gb, width, height) if gb else None
l = t = r = b = 0
if cr:
l, t, r, b = cr
records += _layer_record(t, l, b, r, gname, 255, gvis,
ch, layer_id=layer_id, lsct=3)
for _, p in ch:
payload_f.write(p)
img_path = None
if ly["_type"] == "bitmap":
uri = ly.get("uri", "")
@@ -916,30 +898,41 @@ def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None,
except Exception as exc:
print(f" warning: text layer {name} -> TySh failed: {exc}",
file=sys.stderr)
rendered.append((name, opacity, visible, (lx, ly_), img, tysh))
if visible and opacity > 0:
_opacity_paste(merged, img, (lx, ly_), opacity)
ch = channel_blocks(img, workers, ep)
records += _layer_record(ly_, lx, ly_ + img.height, lx + img.width,
name, opacity, visible,
ch, layer_id=layer_id, tysh=tysh)
for _, p in ch:
payload_f.write(p)
del ch, img
rendered.append((name, opacity, visible, (lx, ly_), None, tysh,
None))
rp.update(1)
# group close marker (top of the app group)
if gid and gid in group_ranges and group_ranges[gid][1] == i:
rendered.append((gname, 255, gvis, (0, 0), None, None,
(1, None)))
ch = [(cid, b"\x00\x01") for cid in (-1, 0, 1, 2)]
records += _layer_record(0, 0, 0, 0, gname, 255, gvis,
ch, layer_id=layer_id + 1, lsct=1)
for _, p in ch:
payload_f.write(p)
rp.close()
ep.close()
print(f"渲染完成: {len(rendered)}/{len(layers)} 层 (跳过 {skipped}), "
f"画布 {width}x{height}")
ep = tqdm(total=len(rendered) * 4 + 3, desc="编码通道", unit="通道",
file=sys.stderr)
build_psd(rendered, width, height, out_path, workers, ep)
ep.close()
build_psd(records, payload_f, merged, rendered, width, height, out_path)
payload_f.close()
print(f"PSD 已写出: {out_path}")
if preview_path:
canvas = Image.new("RGBA", (width, height), (0, 0, 0, 0))
for layer in rendered:
name, opacity, visible, (lx, ly), img = layer[:5]
if img is not None and visible and opacity > 0:
_opacity_paste(canvas, img, (lx, ly), opacity)
# downscale for easy viewing
preview = canvas.convert("RGB")
if preview.width > 2000:
# downscale for easy viewing, unless the project was upscaled
# (then keep the full-res composite so the 2x detail is visible)
preview = merged.convert("RGB")
if _SCALE == 1.0 and preview.width > 2000:
scale = 2000 / preview.width
preview = preview.resize(
(int(preview.width * scale), int(preview.height * scale)),
@@ -959,8 +952,13 @@ def main():
ap.add_argument("--type-layer", action="store_true",
help="convert text layers to editable TypeLayers (TySh) "
"instead of rasterized pixels")
ap.add_argument("--scale", type=float, default=1.0,
help="upscale the whole vector project by this factor "
"(canvas, layer rects, text size, shadows) "
"(default: 1.0)")
args = ap.parse_args()
set_scale(args.scale)
out = args.output
if not out:
out = str(Path(args.input).with_suffix(".psd"))