python/laymesh/bridge.py 1"""Use the LayMesh CLI with the same DSL as Rust and the browser.2 3Bare geometry defaults to mm (canvas unit); typography and line widths use pt.4Bindings substitute {{name}} only outside DSL strings, comments and LCSS.5Use save_source to preserve a layout and data for independent CLI rendering.6"""7 8from __future__ import annotations9 10import hashlib11import io12import json13import math14import os15import re16import shutil17import subprocess18import tempfile19import warnings20from contextvars import ContextVar21from dataclasses import dataclass22from pathlib import Path23from typing import Mapping24 25from ._runtime import bundled_command, development_command, override_command26from ._svg import SvgCompatibilityError, normalize_matplotlib_svg27 28 29_warning_state: ContextVar[tuple[bool, set[str]] | None] = ContextVar("laymesh_warning_state", default=None)30 31 32def _warn(message: str, *, stacklevel: int = 3) -> None:33 state = _warning_state.get()34 if state is not None:35 enabled, seen = state36 if not enabled or message in seen:37 return38 seen.add(message)39 warnings.warn(message, stacklevel=stacklevel)40 41 42def _warning_mode(value: bool | None) -> bool:43 if value is not None:44 if not isinstance(value, bool):45 raise LayMeshBridgeError("show_warnings 须为 None、True 或 False")46 return value47 mode = os.environ.get("LAYMESH_WARNINGS", "show")48 if mode not in ("show", "hide"):49 raise LayMeshBridgeError("LAYMESH_WARNINGS 须为 show 或 hide")50 return mode == "show"51 52 53class LayMeshBridgeError(RuntimeError):54 """A binding, asset, or CLI error with a message suitable for a Notebook."""55 56 57@dataclass(frozen=True)58class RenderResult:59 preview_svg: str60 output: Path | None61 saved_source: Path | None62 63 64_NAME = re.compile(r"[A-Za-z_][A-Za-z_0-9]*\Z")65_GENERATED = "# Generated by LayMesh Jupyter; edit the Notebook cell instead.\n"66_FORMATS = {".svg", ".pdf", ".png", ".jpg", ".jpeg", ".tif", ".tiff", ".webp", ".bmp", ".gif", ".ico", ".pnm", ".pbm", ".pgm", ".ppm", ".pam", ".tga"}67_RASTER_FORMATS = _FORMATS - {".svg", ".pdf"}68 69 70def _command() -> list[str]:71 try:72 command = override_command() or bundled_command() or development_command()73 except RuntimeError as error:74 raise LayMeshBridgeError(str(error)) from error75 if command:76 return command77 installed = shutil.which("laymesh")78 if installed:79 return [installed]80 raise LayMeshBridgeError("找不到 LayMesh 原生 CLI;请运行 cargo build --release -p laymesh-cli,或设置 LAYMESH_CLI")81 82 83def _run(cli: list[str], args: list[str], base_dir: Path, source: Path | None = None) -> str:84 try:85 result = subprocess.run(86 [*cli, *args, "--warnings", "show" if _warning_state.get() is None or _warning_state.get()[0] else "hide"], cwd=base_dir, capture_output=True, text=True,87 encoding="utf-8", errors="replace", check=False,88 )89 except OSError as exc:90 raise LayMeshBridgeError(f"无法启动 LayMesh CLI:{exc}") from exc91 message = result.stderr.strip()92 if source is not None:93 message = message.replace(str(source), "<LayMesh cell>")94 if result.returncode:95 raise LayMeshBridgeError(message or result.stdout.strip() or f"LayMesh 退出状态 {result.returncode}")96 if message:97 for diagnostic in message.splitlines():98 _warn(diagnostic, stacklevel=3)99 return result.stdout100 101 102def _temporary_lay(base_dir: Path, source: str):103 handle = tempfile.NamedTemporaryFile(104 mode="w", encoding="utf-8", prefix=".laymesh-", suffix=".lay",105 dir=base_dir, delete=False,106 )107 try:108 handle.write(source)109 handle.close()110 return Path(handle.name)111 except Exception:112 handle.close()113 Path(handle.name).unlink(missing_ok=True)114 raise115 116 117def _figure_type():118 try:119 from matplotlib.figure import Figure120 except ImportError:121 return ()122 return Figure123 124 125def _numpy_scalar(value):126 try:127 import numpy as np128 except ImportError:129 return value130 return value.item() if isinstance(value, np.generic) else value131 132 133class _Bindings:134 def __init__(135 self, namespace: Mapping[str, object], cli: list[str], base_dir: Path,136 scratch: Path, plot_dpi: float, saved_source: Path | None,137 ) -> None:138 self.namespace = namespace139 self.cli = cli140 self.base_dir = base_dir141 self.scratch = scratch142 self.plot_dpi = plot_dpi143 self.saved_source = saved_source144 self.assets_dir = saved_source.with_suffix(".assets") if saved_source else None145 self.cache: dict[str, str] = {}146 147 def _asset(self, name: str, data: bytes, extension: str) -> str:148 if self.assets_dir:149 self.assets_dir.mkdir(exist_ok=True)150 filename = f"{name}-{hashlib.sha256(data).hexdigest()[:16]}{extension}"151 target = self.assets_dir / filename152 if not target.exists():153 target.write_bytes(data)154 return f"{self.assets_dir.name}/{filename}"155 target = self.scratch / f"{name}{extension}"156 target.write_bytes(data)157 return str(target)158 159 def _svg_accepted(self, svg: bytes, name: str) -> bool:160 candidate = self.scratch / f"{name}.svg"161 candidate.write_bytes(svg)162 probe = _temporary_lay(163 self.base_dir,164 f'page=canvas(size=(10 mm,10 mm))\nplot=image(src={json.dumps(str(candidate))})\npage.add(plot,size=(8 mm, auto),target=page.top_left)\n',165 )166 try:167 result = subprocess.run(168 [*self.cli, "validate", str(probe)], cwd=self.base_dir,169 capture_output=True, text=True, encoding="utf-8", errors="replace", check=False,170 )171 if result.returncode == 0:172 return True173 if "E_SVG:" in result.stderr:174 return False175 raise LayMeshBridgeError(result.stderr.strip() or result.stdout.strip())176 except OSError as exc:177 raise LayMeshBridgeError(f"无法启动 LayMesh CLI:{exc}") from exc178 finally:179 probe.unlink(missing_ok=True)180 181 @staticmethod182 def _warn_plot_bounds(name: str, figure) -> None:183 # Both boxes use inches. Keep savefig's original bounds unchanged: this184 # check only explains clipping that an inline tight preview can hide.185 try:186 canvas = figure.canvas187 canvas.draw()188 content = figure.get_tightbbox(canvas.get_renderer())189 frame = figure.bbox_inches190 if content is None:191 return192 overhang = (193 ("left", frame.x0 - content.x0),194 ("bottom", frame.y0 - content.y0),195 ("right", content.x1 - frame.x1),196 ("top", content.y1 - frame.y1),197 )198 outside = [(side, float(amount)) for side, amount in overhang199 if math.isfinite(amount) and amount > 1 / 72]200 except Exception:201 # Some Matplotlib canvases have no renderer available here. Saving202 # the figure must still follow the existing path.203 return204 if outside:205 sides = ", ".join(f"{side} {amount * 25.4:.1f} mm" for side, amount in outside)206 _warn(207 f"W_PLOT_BOUNDS: Visible content in Matplotlib figure '{name}' extends beyond "208 f"the figure canvas ({sides}). Exporting with the original bounds may clip it, "209 "so the LayMesh output can differ from Jupyter's inline preview. "210 'Call fig.tight_layout() or create the figure with layout="constrained".',211 stacklevel=7,212 )213 214 def _figure(self, name: str, figure) -> str:215 import matplotlib as mpl216 217 self._warn_plot_bounds(name, figure)218 reason = "SVG 使用了当前 LayMesh 不支持的图形内容"219 try:220 with mpl.rc_context({"svg.fonttype": "path"}):221 stream = io.StringIO()222 figure.savefig(stream, format="svg")223 svg = normalize_matplotlib_svg(stream.getvalue()).encode("utf-8")224 if self._svg_accepted(svg, name):225 return json.dumps(self._asset(name, svg, ".svg"), ensure_ascii=False)226 except SvgCompatibilityError as exc:227 reason = str(exc)228 _warn(229 f"Matplotlib 图 {name} 已改用 {self.plot_dpi:g} DPI PNG:{reason}",230 stacklevel=3,231 )232 stream = io.BytesIO()233 figure.savefig(stream, format="png", dpi=self.plot_dpi)234 return json.dumps(self._asset(name, stream.getvalue(), ".png"), ensure_ascii=False)235 236 def literal(self, name: str, line: int) -> str:237 if name in self.cache:238 return self.cache[name]239 if name not in self.namespace:240 raise LayMeshBridgeError(f"第 {line} 行引用了未定义的 Notebook 变量 {{{{{name}}}}}")241 value = _numpy_scalar(self.namespace[name])242 data_literal = self._data_literal(name, value, line)243 if data_literal is not None:244 literal = data_literal245 elif isinstance(value, Mapping):246 literal = self._dict_literal(name, value, line)247 elif isinstance(value, _figure_type()):248 literal = self._figure(name, value)249 elif isinstance(value, (str, Path)):250 literal = json.dumps(str(value), ensure_ascii=False)251 elif isinstance(value, bool):252 literal = "true" if value else "false"253 elif isinstance(value, (int, float)):254 try:255 finite = math.isfinite(value)256 except (OverflowError, ValueError):257 finite = False258 if not finite or isinstance(value, int) and abs(value) > 2**53 - 1:259 raise LayMeshBridgeError(f"第 {line} 行变量 {name} 必须是有限且可精确表示的数值")260 literal = repr(value)261 else:262 raise LayMeshBridgeError(263 f"第 {line} 行变量 {name} 类型不支持;支持标量、字典、数值数组、DataFrame 和 Matplotlib Figure"264 )265 self.cache[name] = literal266 return literal267 268 def _dict_literal(self, name: str, value: Mapping, line: int) -> str:269 """Serialize JSON containers to an asset, retaining mapping insertion order."""270 active: set[int] = set()271 272 def fail(path, reason):273 raise LayMeshBridgeError(f"第 {line} 行字典 {path}: {reason}")274 275 def normalize(item, path, depth=0):276 if depth > 100:277 fail(path, "嵌套超过 100 层")278 item = _numpy_scalar(item)279 if item is None or isinstance(item, (str, bool)):280 return item281 if isinstance(item, (int, float)):282 if isinstance(item, float) and math.isnan(item):283 return None284 try:285 finite = math.isfinite(item)286 except (OverflowError, ValueError):287 finite = False288 if not finite or isinstance(item, int) and abs(item) > 2**53 - 1:289 fail(path, "数值必须有限且能精确表示")290 return item291 is_numpy = type(item).__module__.split(".")[0] == "numpy"292 if not isinstance(item, (Mapping, list, tuple)) and not is_numpy:293 fail(path, f"不支持的类型 {type(item).__name__}")294 identity = id(item)295 if identity in active:296 fail(path, "循环引用")297 active.add(identity)298 try:299 if is_numpy:300 import numpy as np301 if not isinstance(item, np.ndarray) or item.ndim not in (1, 2):302 fail(path, "需要一维或二维 NumPy 数组")303 return normalize(item.tolist(), path, depth + 1)304 if isinstance(item, Mapping):305 result = {}306 for key, val in item.items():307 if not isinstance(key, str):308 fail(path, f"键 {key!r} 必须为字符串")309 result[key] = normalize(val, f"{path}[{key!r}]", depth + 1)310 return result311 return [normalize(val, f"{path}[{i}]", depth + 1) for i, val in enumerate(item)]312 finally:313 active.remove(identity)314 315 data = json.dumps(normalize(value, name), ensure_ascii=False, allow_nan=False,316 separators=(",", ":")).encode("utf-8")317 asset = self._asset(name, data, ".json")318 return f"dict(src={json.dumps(asset, ensure_ascii=False)})"319 320 def _data_literal(self, name: str, value, line: int) -> str | None:321 """Bind data as strict JSON assets, never as generated per-point DSL."""322 kind = "array"323 is_table = False324 module = type(value).__module__.split(".")[0]325 pandas = None326 if module == "pandas":327 import pandas328 if isinstance(value, pandas.DataFrame):329 if value.empty or not value.columns.is_unique or any(330 not isinstance(c, str) or not c for c in value.columns331 ):332 raise LayMeshBridgeError(f"第 {line} 行 {name} 需要非空 DataFrame 和唯一字符串列名")333 value = {column: value[column].tolist() for column in value.columns}334 kind, is_table = "table", True335 elif isinstance(value, pandas.Series):336 value = value.tolist()337 else:338 return None339 elif module == "numpy":340 import numpy as np341 if not isinstance(value, np.ndarray):342 return None343 if value.ndim not in (1, 2) or not value.size:344 raise LayMeshBridgeError(f"第 {line} 行 {name} 需要非空一维或二维数组")345 value = value.tolist()346 elif not isinstance(value, (list, tuple)):347 return None348 349 def scalar(item):350 if item is None or pandas is not None and item is pandas.NA:351 return None352 item = _numpy_scalar(item)353 if isinstance(item, bool):354 raise LayMeshBridgeError(f"第 {line} 行 {name} 的绘图数据不接受布尔值")355 if isinstance(item, (int, float)):356 if isinstance(item, float) and math.isnan(item):357 return None358 try:359 finite = math.isfinite(item)360 except OverflowError:361 finite = False362 if not finite or isinstance(item, int) and abs(item) > 2**53 - 1:363 raise LayMeshBridgeError(f"第 {line} 行 {name} 数据必须有限且可精确表示")364 return item365 if is_table and isinstance(item, str):366 return item367 raise LayMeshBridgeError(f"第 {line} 行 {name} 包含不支持的数据类型 {type(item).__name__}")368 369 if is_table:370 normalized = {column: [scalar(item) for item in items] for column, items in value.items()}371 else:372 if not value:373 raise LayMeshBridgeError(f"第 {line} 行 {name} 数组不能为空")374 if isinstance(value[0], (list, tuple)):375 width = len(value[0])376 if not width or any(not isinstance(row, (list, tuple)) or len(row) != width for row in value):377 raise LayMeshBridgeError(f"第 {line} 行 {name} 二维数组必须为非空矩形")378 normalized = [[scalar(item) for item in row] for row in value]379 else:380 normalized = [scalar(item) for item in value]381 data = json.dumps(normalized, ensure_ascii=False, allow_nan=False, separators=(",", ":")).encode("utf-8")382 asset = self._asset(name, data, ".json")383 return f"{kind}(src={json.dumps(asset, ensure_ascii=False)})"384 385 def expand(self, source: str) -> str:386 """Expand bindings only in DSL code, never in strings, comments or LCSS."""387 out: list[str] = []388 index = 0389 line = 1390 style_depth = 0391 while index < len(source):392 if not style_depth and source[index] == "#":393 end = source.find("\n", index)394 if end < 0:395 out.append(source[index:])396 break397 out.append(source[index:end])398 index = end399 continue400 match = re.match(r"(?:rf|fr|r|f)?([\"'])", source[index:], re.I)401 if match:402 quote = match.group(1)403 opening = index + len(match.group(0)) - 1404 delimiter = quote * 3 if source.startswith(quote * 3, opening) else quote405 end = opening + len(delimiter)406 while end < len(source):407 if source[end] == "\\":408 end += 2409 elif source.startswith(delimiter, end):410 end += len(delimiter)411 break412 else:413 end += 1414 literal = source[index:end]415 out.append(literal)416 line += literal.count("\n")417 index = end418 continue419 if source.startswith("/*", index):420 end = source.find("*/", index + 2)421 end = len(source) if end < 0 else end + 2422 value = source[index:end]423 line += value.count("\n")424 out.append(value)425 index = end426 continue427 if re.match(r"style\s*\{", source[index:]) and (index == 0 or not source[index-1].isalnum()):428 end = index + re.match(r"style\s*\{", source[index:]).end()429 value = source[index:end]430 out.append(value)431 line += value.count("\n")432 index = end433 style_depth = 1434 continue435 if style_depth:436 if source[index] == "{": style_depth += 1437 elif source[index] == "}": style_depth -= 1438 if not style_depth and source.startswith("{{", index):439 end = source.find("}}", index + 2)440 if end < 0:441 raise LayMeshBridgeError(f"第 {line} 行占位符缺少 }}}}")442 name = source[index + 2:end].strip()443 if not _NAME.fullmatch(name):444 raise LayMeshBridgeError(f"第 {line} 行只允许 {{{{变量名}}}},不能写 Python 表达式")445 out.append(self.literal(name, line))446 index = end + 2447 continue448 if source[index] == "\n":449 line += 1450 out.append(source[index])451 index += 1452 return "".join(out)453 454 455def _export_args(output: Path | None, options: Mapping[str, object]) -> list[str]:456 """Validate before staging assets; the native encoder validates the same options again."""457 explicit = {key: value for key, value in options.items() if value is not None}458 if not explicit:459 return []460 if output is None:461 raise LayMeshBridgeError("导出编码参数需要 output 文件")462 suffix = output.suffix.lower()463 jpeg = suffix in {".jpg", ".jpeg"}464 tiff = suffix in {".tif", ".tiff"}465 webp = suffix == ".webp"466 quality = explicit.get("quality")467 if quality is not None and (not (jpeg or webp) or isinstance(quality, bool) or not isinstance(quality, (int, float)) or not math.isfinite(quality) or not (1 if jpeg else 0) <= quality <= 100):468 raise LayMeshBridgeError("quality 仅用于 JPEG(1–100)或 WebP(0–100)")469 compression = explicit.get("compression")470 if compression is not None and compression not in ({"none", "lzw", "deflate", "packbits"} if tiff else {"fast", "default", "best"} if suffix == ".png" else set()):471 raise LayMeshBridgeError("compression:PNG 使用 fast/default/best;TIFF 使用 none/lzw/deflate/packbits")472 background = explicit.get("background")473 if background is not None and (suffix not in {".jpg", ".jpeg", ".gif", ".ppm", ".pgm", ".pbm"} or not isinstance(background, str) or not re.fullmatch(r"#[0-9a-fA-F]{6}", background)):474 raise LayMeshBridgeError("background 仅用于 JPEG/GIF/PPM/PGM/PBM,须为 #RRGGBB")475 for key in ("webp_lossless", "webp_method", "webp_alpha_quality", "webp_near_lossless"):476 if key not in explicit:477 continue478 value = explicit[key]479 if not webp or (key == "webp_lossless" and not isinstance(value, bool)) or (key != "webp_lossless" and (type(value) is not int or not 0 <= value <= (6 if key == "webp_method" else 100))):480 raise LayMeshBridgeError("webp_* 仅用于 WebP:lossless 为 bool,method 为 0–6,alpha_quality/near_lossless 为 0–100 整数")481 if explicit.get("webp_lossless") is False and "webp_near_lossless" in explicit:482 raise LayMeshBridgeError("webp_near_lossless 仅用于无损 WebP 模式")483 if explicit.get("webp_lossless", True) and explicit.get("webp_alpha_quality", 100) != 100:484 raise LayMeshBridgeError("降低 webp_alpha_quality 须选择有损 WebP 模式")485 return [part for key, value in explicit.items() for part in ("--" + key.replace("_", "-"), str(value).lower() if isinstance(value, bool) else str(value))]486 487 488def _render_impl(489 source: str, *, namespace: Mapping[str, object], base_dir: Path,490 output: Path | str | None, dpi: float | None, plot_dpi: float,491 save_source: Path | str | None, original: Path | None,492 export_options: Mapping[str, object],493) -> RenderResult:494 base_dir = base_dir.resolve()495 if not base_dir.is_dir():496 raise LayMeshBridgeError(f"素材目录不存在:{base_dir}")497 if not math.isfinite(plot_dpi) or plot_dpi <= 0:498 raise LayMeshBridgeError("--plot-dpi 必须是正数")499 if dpi is not None and (not math.isfinite(dpi) or dpi <= 0 or dpi > 25400):500 raise LayMeshBridgeError("--dpi 必须为 0–25400 之间的正数")501 output_path = Path(output).resolve() if output is not None else None502 if output_path and output_path.suffix.lower() not in _FORMATS:503 raise LayMeshBridgeError("输出文件须为 SVG/PDF/PNG/JPEG/TIFF/WebP/BMP/GIF/ICO/PNM/TGA")504 if dpi is not None and (not output_path or output_path.suffix.lower() not in _RASTER_FORMATS):505 raise LayMeshBridgeError("--dpi 只用于位图输出")506 option_args = _export_args(output_path, export_options)507 saved_path = (base_dir / save_source).resolve() if save_source else None508 if saved_path:509 if saved_path.suffix != ".lay" or saved_path.parent != base_dir:510 raise LayMeshBridgeError("--save-source 须是素材目录中的 .lay 文件")511 if saved_path.exists() and not saved_path.read_text(encoding="utf-8").startswith(_GENERATED):512 raise LayMeshBridgeError(f"不会覆盖非生成的布局文件:{saved_path}")513 cli = _command()514 with tempfile.TemporaryDirectory(prefix="laymesh-assets-") as scratch_name:515 binding = _Bindings(namespace, cli, base_dir, Path(scratch_name), plot_dpi, saved_path)516 expanded = binding.expand(source)517 file = original if original and expanded == source and not saved_path else _temporary_lay(base_dir, expanded)518 cleanup = file != original519 try:520 if output_path:521 output_path.parent.mkdir(parents=True, exist_ok=True)522 args = ["render", str(file), "-o", str(output_path)]523 if dpi is not None:524 args += ["--dpi", str(dpi)]525 args += option_args526 _run(cli, args, base_dir, file if cleanup else None)527 if output_path and output_path.suffix.lower() == ".svg":528 preview = output_path.read_text(encoding="utf-8")529 else:530 preview_file = Path(scratch_name) / "preview.svg"531 _run(cli, ["render", str(file), "-o", str(preview_file)], base_dir, file if cleanup else None)532 preview = preview_file.read_text(encoding="utf-8")533 if saved_path:534 staged = _temporary_lay(base_dir, _GENERATED + expanded)535 try:536 os.replace(staged, saved_path)537 finally:538 staged.unlink(missing_ok=True)539 return RenderResult(preview, output_path, saved_path)540 finally:541 if cleanup:542 file.unlink(missing_ok=True)543 544 545def _render(source: str, *, show_warnings: bool | None = None, **kwargs) -> RenderResult:546 token = _warning_state.set((_warning_mode(show_warnings), set()))547 try:548 return _render_impl(source, **kwargs)549 finally:550 _warning_state.reset(token)551 552 553def render_source(554 source: str, *, namespace: Mapping[str, object] | None = None,555 base_dir: Path | str | None = None, output: Path | str | None = None,556 dpi: float | None = None, plot_dpi: float = 300,557 save_source: Path | str | None = None, show_warnings: bool | None = None,558 quality: float | None = None, compression: str | None = None, background: str | None = None,559 webp_lossless: bool | None = None, webp_method: int | None = None,560 webp_alpha_quality: int | None = None, webp_near_lossless: int | None = None,561) -> RenderResult:562 """Render inline .lay source with optional Notebook variable bindings."""563 return _render(564 source, namespace=namespace or {}, base_dir=Path(base_dir or Path.cwd()),565 output=output, dpi=dpi, plot_dpi=plot_dpi,566 save_source=save_source, original=None, show_warnings=show_warnings,567 export_options=dict(quality=quality, compression=compression, background=background,568 webp_lossless=webp_lossless, webp_method=webp_method,569 webp_alpha_quality=webp_alpha_quality, webp_near_lossless=webp_near_lossless),570 )571 572 573def render_file(574 file: Path | str, *, namespace: Mapping[str, object] | None = None,575 output: Path | str | None = None, dpi: float | None = None,576 plot_dpi: float = 300, save_source: Path | str | None = None, show_warnings: bool | None = None,577 quality: float | None = None, compression: str | None = None, background: str | None = None,578 webp_lossless: bool | None = None, webp_method: int | None = None,579 webp_alpha_quality: int | None = None, webp_near_lossless: int | None = None,580) -> RenderResult:581 """Render an existing .lay file; its relative assets stay relative to it."""582 file = Path(file).resolve()583 if file.suffix != ".lay" or not file.is_file():584 raise LayMeshBridgeError(f"布局文件不存在或不是 .lay:{file}")585 return _render(586 file.read_text(encoding="utf-8"), namespace=namespace or {},587 base_dir=file.parent, output=output, dpi=dpi,588 plot_dpi=plot_dpi, save_source=save_source, original=file, show_warnings=show_warnings,589 export_options=dict(quality=quality, compression=compression, background=background,590 webp_lossless=webp_lossless, webp_method=webp_method,591 webp_alpha_quality=webp_alpha_quality, webp_near_lossless=webp_near_lossless),592 )