LayMesh文档
EN

python/laymesh/bridge.py

python/laymesh/bridge.py
"""Use the LayMesh CLI with the same DSL as Rust and the browser. Bare geometry defaults to mm (canvas unit); typography and line widths use pt.Bindings substitute {{name}} only outside DSL strings, comments and LCSS.Use save_source to preserve a layout and data for independent CLI rendering.""" from __future__ import annotations import hashlibimport ioimport jsonimport mathimport osimport reimport shutilimport subprocessimport tempfileimport warningsfrom contextvars import ContextVarfrom dataclasses import dataclassfrom pathlib import Pathfrom typing import Mapping from ._runtime import bundled_command, development_command, override_commandfrom ._svg import SvgCompatibilityError, normalize_matplotlib_svg  _warning_state: ContextVar[tuple[bool, set[str]] | None] = ContextVar("laymesh_warning_state", default=None)  def _warn(message: str, *, stacklevel: int = 3) -> None:    state = _warning_state.get()    if state is not None:        enabled, seen = state        if not enabled or message in seen:            return        seen.add(message)    warnings.warn(message, stacklevel=stacklevel)  def _warning_mode(value: bool | None) -> bool:    if value is not None:        if not isinstance(value, bool):            raise LayMeshBridgeError("show_warnings 须为 None、True 或 False")        return value    mode = os.environ.get("LAYMESH_WARNINGS", "show")    if mode not in ("show", "hide"):        raise LayMeshBridgeError("LAYMESH_WARNINGS 须为 show 或 hide")    return mode == "show"  class LayMeshBridgeError(RuntimeError):    """A binding, asset, or CLI error with a message suitable for a Notebook."""  @dataclass(frozen=True)class RenderResult:    preview_svg: str    output: Path | None    saved_source: Path | None  _NAME = re.compile(r"[A-Za-z_][A-Za-z_0-9]*\Z")_GENERATED = "# Generated by LayMesh Jupyter; edit the Notebook cell instead.\n"_FORMATS = {".svg", ".pdf", ".png", ".jpg", ".jpeg", ".tif", ".tiff", ".webp", ".bmp", ".gif", ".ico", ".pnm", ".pbm", ".pgm", ".ppm", ".pam", ".tga"}_RASTER_FORMATS = _FORMATS - {".svg", ".pdf"}  def _command() -> list[str]:    try:        command = override_command() or bundled_command() or development_command()    except RuntimeError as error:        raise LayMeshBridgeError(str(error)) from error    if command:        return command    installed = shutil.which("laymesh")    if installed:        return [installed]    raise LayMeshBridgeError("找不到 LayMesh 原生 CLI;请运行 cargo build --release -p laymesh-cli,或设置 LAYMESH_CLI")  def _run(cli: list[str], args: list[str], base_dir: Path, source: Path | None = None) -> str:    try:        result = subprocess.run(            [*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,            encoding="utf-8", errors="replace", check=False,        )    except OSError as exc:        raise LayMeshBridgeError(f"无法启动 LayMesh CLI:{exc}") from exc    message = result.stderr.strip()    if source is not None:        message = message.replace(str(source), "<LayMesh cell>")    if result.returncode:        raise LayMeshBridgeError(message or result.stdout.strip() or f"LayMesh 退出状态 {result.returncode}")    if message:        for diagnostic in message.splitlines():            _warn(diagnostic, stacklevel=3)    return result.stdout  def _temporary_lay(base_dir: Path, source: str):    handle = tempfile.NamedTemporaryFile(        mode="w", encoding="utf-8", prefix=".laymesh-", suffix=".lay",        dir=base_dir, delete=False,    )    try:        handle.write(source)        handle.close()        return Path(handle.name)    except Exception:        handle.close()        Path(handle.name).unlink(missing_ok=True)        raise  def _figure_type():    try:        from matplotlib.figure import Figure    except ImportError:        return ()    return Figure  def _numpy_scalar(value):    try:        import numpy as np    except ImportError:        return value    return value.item() if isinstance(value, np.generic) else value  class _Bindings:    def __init__(        self, namespace: Mapping[str, object], cli: list[str], base_dir: Path,        scratch: Path, plot_dpi: float, saved_source: Path | None,    ) -> None:        self.namespace = namespace        self.cli = cli        self.base_dir = base_dir        self.scratch = scratch        self.plot_dpi = plot_dpi        self.saved_source = saved_source        self.assets_dir = saved_source.with_suffix(".assets") if saved_source else None        self.cache: dict[str, str] = {}     def _asset(self, name: str, data: bytes, extension: str) -> str:        if self.assets_dir:            self.assets_dir.mkdir(exist_ok=True)            filename = f"{name}-{hashlib.sha256(data).hexdigest()[:16]}{extension}"            target = self.assets_dir / filename            if not target.exists():                target.write_bytes(data)            return f"{self.assets_dir.name}/{filename}"        target = self.scratch / f"{name}{extension}"        target.write_bytes(data)        return str(target)     def _svg_accepted(self, svg: bytes, name: str) -> bool:        candidate = self.scratch / f"{name}.svg"        candidate.write_bytes(svg)        probe = _temporary_lay(            self.base_dir,            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',        )        try:            result = subprocess.run(                [*self.cli, "validate", str(probe)], cwd=self.base_dir,                capture_output=True, text=True, encoding="utf-8", errors="replace", check=False,            )            if result.returncode == 0:                return True            if "E_SVG:" in result.stderr:                return False            raise LayMeshBridgeError(result.stderr.strip() or result.stdout.strip())        except OSError as exc:            raise LayMeshBridgeError(f"无法启动 LayMesh CLI:{exc}") from exc        finally:            probe.unlink(missing_ok=True)     @staticmethod    def _warn_plot_bounds(name: str, figure) -> None:        # Both boxes use inches. Keep savefig's original bounds unchanged: this        # check only explains clipping that an inline tight preview can hide.        try:            canvas = figure.canvas            canvas.draw()            content = figure.get_tightbbox(canvas.get_renderer())            frame = figure.bbox_inches            if content is None:                return            overhang = (                ("left", frame.x0 - content.x0),                ("bottom", frame.y0 - content.y0),                ("right", content.x1 - frame.x1),                ("top", content.y1 - frame.y1),            )            outside = [(side, float(amount)) for side, amount in overhang                       if math.isfinite(amount) and amount > 1 / 72]        except Exception:            # Some Matplotlib canvases have no renderer available here. Saving            # the figure must still follow the existing path.            return        if outside:            sides = ", ".join(f"{side} {amount * 25.4:.1f} mm" for side, amount in outside)            _warn(                f"W_PLOT_BOUNDS: Visible content in Matplotlib figure '{name}' extends beyond "                f"the figure canvas ({sides}). Exporting with the original bounds may clip it, "                "so the LayMesh output can differ from Jupyter's inline preview. "                'Call fig.tight_layout() or create the figure with layout="constrained".',                stacklevel=7,            )     def _figure(self, name: str, figure) -> str:        import matplotlib as mpl         self._warn_plot_bounds(name, figure)        reason = "SVG 使用了当前 LayMesh 不支持的图形内容"        try:            with mpl.rc_context({"svg.fonttype": "path"}):                stream = io.StringIO()                figure.savefig(stream, format="svg")            svg = normalize_matplotlib_svg(stream.getvalue()).encode("utf-8")            if self._svg_accepted(svg, name):                return json.dumps(self._asset(name, svg, ".svg"), ensure_ascii=False)        except SvgCompatibilityError as exc:            reason = str(exc)        _warn(            f"Matplotlib 图 {name} 已改用 {self.plot_dpi:g} DPI PNG:{reason}",            stacklevel=3,        )        stream = io.BytesIO()        figure.savefig(stream, format="png", dpi=self.plot_dpi)        return json.dumps(self._asset(name, stream.getvalue(), ".png"), ensure_ascii=False)     def literal(self, name: str, line: int) -> str:        if name in self.cache:            return self.cache[name]        if name not in self.namespace:            raise LayMeshBridgeError(f"第 {line} 行引用了未定义的 Notebook 变量 {{{{{name}}}}}")        value = _numpy_scalar(self.namespace[name])        data_literal = self._data_literal(name, value, line)        if data_literal is not None:            literal = data_literal        elif isinstance(value, Mapping):            literal = self._dict_literal(name, value, line)        elif isinstance(value, _figure_type()):            literal = self._figure(name, value)        elif isinstance(value, (str, Path)):            literal = json.dumps(str(value), ensure_ascii=False)        elif isinstance(value, bool):            literal = "true" if value else "false"        elif isinstance(value, (int, float)):            try:                finite = math.isfinite(value)            except (OverflowError, ValueError):                finite = False            if not finite or isinstance(value, int) and abs(value) > 2**53 - 1:                raise LayMeshBridgeError(f"第 {line} 行变量 {name} 必须是有限且可精确表示的数值")            literal = repr(value)        else:            raise LayMeshBridgeError(                f"第 {line} 行变量 {name} 类型不支持;支持标量、字典、数值数组、DataFrame 和 Matplotlib Figure"            )        self.cache[name] = literal        return literal     def _dict_literal(self, name: str, value: Mapping, line: int) -> str:        """Serialize JSON containers to an asset, retaining mapping insertion order."""        active: set[int] = set()         def fail(path, reason):            raise LayMeshBridgeError(f"第 {line} 行字典 {path}: {reason}")         def normalize(item, path, depth=0):            if depth > 100:                fail(path, "嵌套超过 100 层")            item = _numpy_scalar(item)            if item is None or isinstance(item, (str, bool)):                return item            if isinstance(item, (int, float)):                if isinstance(item, float) and math.isnan(item):                    return None                try:                    finite = math.isfinite(item)                except (OverflowError, ValueError):                    finite = False                if not finite or isinstance(item, int) and abs(item) > 2**53 - 1:                    fail(path, "数值必须有限且能精确表示")                return item            is_numpy = type(item).__module__.split(".")[0] == "numpy"            if not isinstance(item, (Mapping, list, tuple)) and not is_numpy:                fail(path, f"不支持的类型 {type(item).__name__}")            identity = id(item)            if identity in active:                fail(path, "循环引用")            active.add(identity)            try:                if is_numpy:                    import numpy as np                    if not isinstance(item, np.ndarray) or item.ndim not in (1, 2):                        fail(path, "需要一维或二维 NumPy 数组")                    return normalize(item.tolist(), path, depth + 1)                if isinstance(item, Mapping):                    result = {}                    for key, val in item.items():                        if not isinstance(key, str):                            fail(path, f"键 {key!r} 必须为字符串")                        result[key] = normalize(val, f"{path}[{key!r}]", depth + 1)                    return result                return [normalize(val, f"{path}[{i}]", depth + 1) for i, val in enumerate(item)]            finally:                active.remove(identity)         data = json.dumps(normalize(value, name), ensure_ascii=False, allow_nan=False,                          separators=(",", ":")).encode("utf-8")        asset = self._asset(name, data, ".json")        return f"dict(src={json.dumps(asset, ensure_ascii=False)})"     def _data_literal(self, name: str, value, line: int) -> str | None:        """Bind data as strict JSON assets, never as generated per-point DSL."""        kind = "array"        is_table = False        module = type(value).__module__.split(".")[0]        pandas = None        if module == "pandas":            import pandas            if isinstance(value, pandas.DataFrame):                if value.empty or not value.columns.is_unique or any(                    not isinstance(c, str) or not c for c in value.columns                ):                    raise LayMeshBridgeError(f"第 {line} 行 {name} 需要非空 DataFrame 和唯一字符串列名")                value = {column: value[column].tolist() for column in value.columns}                kind, is_table = "table", True            elif isinstance(value, pandas.Series):                value = value.tolist()            else:                return None        elif module == "numpy":            import numpy as np            if not isinstance(value, np.ndarray):                return None            if value.ndim not in (1, 2) or not value.size:                raise LayMeshBridgeError(f"第 {line} 行 {name} 需要非空一维或二维数组")            value = value.tolist()        elif not isinstance(value, (list, tuple)):            return None         def scalar(item):            if item is None or pandas is not None and item is pandas.NA:                return None            item = _numpy_scalar(item)            if isinstance(item, bool):                raise LayMeshBridgeError(f"第 {line} 行 {name} 的绘图数据不接受布尔值")            if isinstance(item, (int, float)):                if isinstance(item, float) and math.isnan(item):                    return None                try:                    finite = math.isfinite(item)                except OverflowError:                    finite = False                if not finite or isinstance(item, int) and abs(item) > 2**53 - 1:                    raise LayMeshBridgeError(f"第 {line} 行 {name} 数据必须有限且可精确表示")                return item            if is_table and isinstance(item, str):                return item            raise LayMeshBridgeError(f"第 {line} 行 {name} 包含不支持的数据类型 {type(item).__name__}")         if is_table:            normalized = {column: [scalar(item) for item in items] for column, items in value.items()}        else:            if not value:                raise LayMeshBridgeError(f"第 {line} 行 {name} 数组不能为空")            if isinstance(value[0], (list, tuple)):                width = len(value[0])                if not width or any(not isinstance(row, (list, tuple)) or len(row) != width for row in value):                    raise LayMeshBridgeError(f"第 {line} 行 {name} 二维数组必须为非空矩形")                normalized = [[scalar(item) for item in row] for row in value]            else:                normalized = [scalar(item) for item in value]        data = json.dumps(normalized, ensure_ascii=False, allow_nan=False, separators=(",", ":")).encode("utf-8")        asset = self._asset(name, data, ".json")        return f"{kind}(src={json.dumps(asset, ensure_ascii=False)})"     def expand(self, source: str) -> str:        """Expand bindings only in DSL code, never in strings, comments or LCSS."""        out: list[str] = []        index = 0        line = 1        style_depth = 0        while index < len(source):            if not style_depth and source[index] == "#":                end = source.find("\n", index)                if end < 0:                    out.append(source[index:])                    break                out.append(source[index:end])                index = end                continue            match = re.match(r"(?:rf|fr|r|f)?([\"'])", source[index:], re.I)            if match:                quote = match.group(1)                opening = index + len(match.group(0)) - 1                delimiter = quote * 3 if source.startswith(quote * 3, opening) else quote                end = opening + len(delimiter)                while end < len(source):                    if source[end] == "\\":                        end += 2                    elif source.startswith(delimiter, end):                        end += len(delimiter)                        break                    else:                        end += 1                literal = source[index:end]                out.append(literal)                line += literal.count("\n")                index = end                continue            if source.startswith("/*", index):                end = source.find("*/", index + 2)                end = len(source) if end < 0 else end + 2                value = source[index:end]                line += value.count("\n")                out.append(value)                index = end                continue            if re.match(r"style\s*\{", source[index:]) and (index == 0 or not source[index-1].isalnum()):                end = index + re.match(r"style\s*\{", source[index:]).end()                value = source[index:end]                out.append(value)                line += value.count("\n")                index = end                style_depth = 1                continue            if style_depth:                if source[index] == "{": style_depth += 1                elif source[index] == "}": style_depth -= 1            if not style_depth and source.startswith("{{", index):                end = source.find("}}", index + 2)                if end < 0:                    raise LayMeshBridgeError(f"第 {line} 行占位符缺少 }}}}")                name = source[index + 2:end].strip()                if not _NAME.fullmatch(name):                    raise LayMeshBridgeError(f"第 {line} 行只允许 {{{{变量名}}}},不能写 Python 表达式")                out.append(self.literal(name, line))                index = end + 2                continue            if source[index] == "\n":                line += 1            out.append(source[index])            index += 1        return "".join(out)  def _export_args(output: Path | None, options: Mapping[str, object]) -> list[str]:    """Validate before staging assets; the native encoder validates the same options again."""    explicit = {key: value for key, value in options.items() if value is not None}    if not explicit:        return []    if output is None:        raise LayMeshBridgeError("导出编码参数需要 output 文件")    suffix = output.suffix.lower()    jpeg = suffix in {".jpg", ".jpeg"}    tiff = suffix in {".tif", ".tiff"}    webp = suffix == ".webp"    quality = explicit.get("quality")    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):        raise LayMeshBridgeError("quality 仅用于 JPEG(1–100)或 WebP(0–100)")    compression = explicit.get("compression")    if compression is not None and compression not in ({"none", "lzw", "deflate", "packbits"} if tiff else {"fast", "default", "best"} if suffix == ".png" else set()):        raise LayMeshBridgeError("compression:PNG 使用 fast/default/best;TIFF 使用 none/lzw/deflate/packbits")    background = explicit.get("background")    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)):        raise LayMeshBridgeError("background 仅用于 JPEG/GIF/PPM/PGM/PBM,须为 #RRGGBB")    for key in ("webp_lossless", "webp_method", "webp_alpha_quality", "webp_near_lossless"):        if key not in explicit:            continue        value = explicit[key]        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))):            raise LayMeshBridgeError("webp_* 仅用于 WebP:lossless 为 bool,method 为 0–6,alpha_quality/near_lossless 为 0–100 整数")    if explicit.get("webp_lossless") is False and "webp_near_lossless" in explicit:        raise LayMeshBridgeError("webp_near_lossless 仅用于无损 WebP 模式")    if explicit.get("webp_lossless", True) and explicit.get("webp_alpha_quality", 100) != 100:        raise LayMeshBridgeError("降低 webp_alpha_quality 须选择有损 WebP 模式")    return [part for key, value in explicit.items() for part in ("--" + key.replace("_", "-"), str(value).lower() if isinstance(value, bool) else str(value))]  def _render_impl(    source: str, *, namespace: Mapping[str, object], base_dir: Path,    output: Path | str | None, dpi: float | None, plot_dpi: float,    save_source: Path | str | None, original: Path | None,    export_options: Mapping[str, object],) -> RenderResult:    base_dir = base_dir.resolve()    if not base_dir.is_dir():        raise LayMeshBridgeError(f"素材目录不存在:{base_dir}")    if not math.isfinite(plot_dpi) or plot_dpi <= 0:        raise LayMeshBridgeError("--plot-dpi 必须是正数")    if dpi is not None and (not math.isfinite(dpi) or dpi <= 0 or dpi > 25400):        raise LayMeshBridgeError("--dpi 必须为 0–25400 之间的正数")    output_path = Path(output).resolve() if output is not None else None    if output_path and output_path.suffix.lower() not in _FORMATS:        raise LayMeshBridgeError("输出文件须为 SVG/PDF/PNG/JPEG/TIFF/WebP/BMP/GIF/ICO/PNM/TGA")    if dpi is not None and (not output_path or output_path.suffix.lower() not in _RASTER_FORMATS):        raise LayMeshBridgeError("--dpi 只用于位图输出")    option_args = _export_args(output_path, export_options)    saved_path = (base_dir / save_source).resolve() if save_source else None    if saved_path:        if saved_path.suffix != ".lay" or saved_path.parent != base_dir:            raise LayMeshBridgeError("--save-source 须是素材目录中的 .lay 文件")        if saved_path.exists() and not saved_path.read_text(encoding="utf-8").startswith(_GENERATED):            raise LayMeshBridgeError(f"不会覆盖非生成的布局文件:{saved_path}")    cli = _command()    with tempfile.TemporaryDirectory(prefix="laymesh-assets-") as scratch_name:        binding = _Bindings(namespace, cli, base_dir, Path(scratch_name), plot_dpi, saved_path)        expanded = binding.expand(source)        file = original if original and expanded == source and not saved_path else _temporary_lay(base_dir, expanded)        cleanup = file != original        try:            if output_path:                output_path.parent.mkdir(parents=True, exist_ok=True)                args = ["render", str(file), "-o", str(output_path)]                if dpi is not None:                    args += ["--dpi", str(dpi)]                args += option_args                _run(cli, args, base_dir, file if cleanup else None)            if output_path and output_path.suffix.lower() == ".svg":                preview = output_path.read_text(encoding="utf-8")            else:                preview_file = Path(scratch_name) / "preview.svg"                _run(cli, ["render", str(file), "-o", str(preview_file)], base_dir, file if cleanup else None)                preview = preview_file.read_text(encoding="utf-8")            if saved_path:                staged = _temporary_lay(base_dir, _GENERATED + expanded)                try:                    os.replace(staged, saved_path)                finally:                    staged.unlink(missing_ok=True)            return RenderResult(preview, output_path, saved_path)        finally:            if cleanup:                file.unlink(missing_ok=True)  def _render(source: str, *, show_warnings: bool | None = None, **kwargs) -> RenderResult:    token = _warning_state.set((_warning_mode(show_warnings), set()))    try:        return _render_impl(source, **kwargs)    finally:        _warning_state.reset(token)  def render_source(    source: str, *, namespace: Mapping[str, object] | None = None,    base_dir: Path | str | None = None, output: Path | str | None = None,    dpi: float | None = None, plot_dpi: float = 300,    save_source: Path | str | None = None, show_warnings: bool | None = None,    quality: float | None = None, compression: str | None = None, background: str | None = None,    webp_lossless: bool | None = None, webp_method: int | None = None,    webp_alpha_quality: int | None = None, webp_near_lossless: int | None = None,) -> RenderResult:    """Render inline .lay source with optional Notebook variable bindings."""    return _render(        source, namespace=namespace or {}, base_dir=Path(base_dir or Path.cwd()),        output=output, dpi=dpi, plot_dpi=plot_dpi,        save_source=save_source, original=None, show_warnings=show_warnings,        export_options=dict(quality=quality, compression=compression, background=background,                            webp_lossless=webp_lossless, webp_method=webp_method,                            webp_alpha_quality=webp_alpha_quality, webp_near_lossless=webp_near_lossless),    )  def render_file(    file: Path | str, *, namespace: Mapping[str, object] | None = None,    output: Path | str | None = None, dpi: float | None = None,    plot_dpi: float = 300, save_source: Path | str | None = None, show_warnings: bool | None = None,    quality: float | None = None, compression: str | None = None, background: str | None = None,    webp_lossless: bool | None = None, webp_method: int | None = None,    webp_alpha_quality: int | None = None, webp_near_lossless: int | None = None,) -> RenderResult:    """Render an existing .lay file; its relative assets stay relative to it."""    file = Path(file).resolve()    if file.suffix != ".lay" or not file.is_file():        raise LayMeshBridgeError(f"布局文件不存在或不是 .lay:{file}")    return _render(        file.read_text(encoding="utf-8"), namespace=namespace or {},        base_dir=file.parent, output=output, dpi=dpi,        plot_dpi=plot_dpi, save_source=save_source, original=file, show_warnings=show_warnings,        export_options=dict(quality=quality, compression=compression, background=background,                            webp_lossless=webp_lossless, webp_method=webp_method,                            webp_alpha_quality=webp_alpha_quality, webp_near_lossless=webp_near_lossless),    )