Coverage for klayout_pex/pex25d/artifact.py: 76%
99 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-06 13:14 +0000
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-06 13:14 +0000
1#
2# --------------------------------------------------------------------------------
3# SPDX-FileCopyrightText: 2024-2026 Martin Jan Köhler and Harald Pretl
4# Johannes Kepler University, Institute for Integrated Circuits.
5#
6# This file is part of KPEX
7# (see https://github.com/iic-jku/klayout-pex).
8#
9# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <http://www.gnu.org/licenses/>.
21# SPDX-License-Identifier: GPL-3.0-or-later
22# --------------------------------------------------------------------------------
23#
25"""
26Naming, format inference and stdio handling for PEX25D artifacts.
28There are two orthogonal properties of every PEX25D artifact on disk:
30``kind``
31 Which message it holds — the unresolved ``PEX25DFile`` as written by a
32 reader, or the resolved ``PEX25DScene`` as consumed by a solver adapter.
34``format``
35 How that message is encoded — the normative PEX25D *text* format, binary
36 protobuf, or protobuf text format.
38Both are inferred from the file name, and either can be overridden
39(``--kind`` / ``--format``). The naming convention is::
41 foo.pex25d text format, PEX25DFile
42 foo.pex25d.pb binary protobuf, PEX25DFile
43 foo.pex25d.textpb protobuf text format, PEX25DFile
44 foo.pex25d.scene.pb binary protobuf, PEX25DScene
45 foo.pex25d.scene.textpb protobuf text format, PEX25DScene
47i.e. the ``.scene`` infix selects the kind and the final extension selects the
48format. There is deliberately no text-format spelling of a scene: the text
49format is the *unresolved* format, and resolution is not reversible.
50"""
52from __future__ import annotations
54import contextlib
55import gzip
56import io
57import os
58import sys
59from dataclasses import dataclass
60from enum import StrEnum
61from typing import *
64# A single '-' as a path means stdin (for inputs) or stdout (for outputs).
65STDIO_PATH = '-'
68class ArtifactKind(StrEnum):
69 """Which PEX25D message an artifact holds."""
71 AUTO = 'auto' # infer from the path
72 FILE = 'file' # kpex.pex25d.PEX25DFile — literal, unresolved
73 SCENE = 'scene' # kpex.pex25d.PEX25DScene — resolved, adapter-ready
76class ArtifactFormat(StrEnum):
77 """How a PEX25D message is encoded."""
79 AUTO = 'auto' # infer from the path
80 TEXT = 'text' # the normative PEX25D text format (PEX25DFile only)
81 PB = 'pb' # binary protobuf
82 TEXTPB = 'textpb' # protobuf text format
85# Suffix table, longest first so that '.pex25d.scene.pb' is matched before
86# '.pex25d.pb' would be. Gzip is handled separately, by stripping a trailing
87# '.gz' before the lookup — the repo already writes '.gds.gz' and layout-sized
88# PEX25D text compresses extremely well.
89_SUFFIXES: List[Tuple[str, ArtifactKind, ArtifactFormat]] = [
90 ('.pex25d.scene.textpb', ArtifactKind.SCENE, ArtifactFormat.TEXTPB),
91 ('.pex25d.scene.pb', ArtifactKind.SCENE, ArtifactFormat.PB),
92 ('.pex25d.textpb', ArtifactKind.FILE, ArtifactFormat.TEXTPB),
93 ('.pex25d.pb', ArtifactKind.FILE, ArtifactFormat.PB),
94 ('.pex25d', ArtifactKind.FILE, ArtifactFormat.TEXT),
95]
97# Canonical spelling for each (kind, format) pair, used when a verb has to
98# derive an output name from an input name.
99CANONICAL_SUFFIX: Dict[Tuple[ArtifactKind, ArtifactFormat], str] = {
100 (kind, fmt): suffix for suffix, kind, fmt in reversed(_SUFFIXES)
101}
104class ArtifactNamingError(ValueError):
105 """The kind or format of an artifact could not be determined."""
108@dataclass(frozen=True)
109class ArtifactSpec:
110 """A fully determined artifact: where it lives, what it holds, how it is encoded."""
112 path: str
113 kind: ArtifactKind
114 format: ArtifactFormat
115 gzipped: bool = False
117 @property
118 def is_stdio(self) -> bool:
119 return self.path == STDIO_PATH
121 @property
122 def is_binary(self) -> bool:
123 return self.format == ArtifactFormat.PB or self.gzipped
125 def __str__(self) -> str:
126 where = '<stdin>/<stdout>' if self.is_stdio else self.path
127 return f"{where} ({self.kind.value}, {self.format.value}{', gzipped' if self.gzipped else ''})"
130def named_kind_and_format(path: str) -> Optional[Tuple[ArtifactKind, ArtifactFormat]]:
131 """
132 The kind and encoding the suffix of the path names (see the table above, a trailing '.gz' aside),
133 None for stdio and for a path without a conventional suffix
134 """
135 if path == STDIO_PATH:
136 return None
137 name = os.path.basename(path).removesuffix('.gz')
138 for suffix, suffix_kind, suffix_format in _SUFFIXES:
139 if name.endswith(suffix):
140 return suffix_kind, suffix_format
141 return None
144def infer_artifact_spec(path: str,
145 kind: ArtifactKind = ArtifactKind.AUTO,
146 format: ArtifactFormat = ArtifactFormat.AUTO,
147 default_kind: ArtifactKind = ArtifactKind.FILE,
148 default_format: ArtifactFormat = ArtifactFormat.TEXT) -> ArtifactSpec:
149 """
150 Determine an artifact's kind and encoding from its path, honouring explicit
151 overrides.
153 Inference is by suffix (see the table above). For ``-`` (stdio) there is no
154 suffix to inspect, so the explicit values are used, falling back to
155 ``default_kind`` / ``default_format`` — which is why every verb that accepts
156 ``-`` states its defaults in ``--help``.
158 :raises ArtifactNamingError: if the path carries no recognized suffix and no
159 override was given, or if the resulting combination cannot exist.
160 """
161 gzipped = path != STDIO_PATH and os.path.basename(path).endswith('.gz')
162 inferred_kind = default_kind
163 inferred_format = default_format
165 if path != STDIO_PATH:
166 named = named_kind_and_format(path)
167 if named is not None:
168 inferred_kind, inferred_format = named
169 elif kind == ArtifactKind.AUTO or format == ArtifactFormat.AUTO:
170 raise ArtifactNamingError(
171 f"Can't tell what kind of PEX25D artifact '{path}' is meant to be. "
172 f"Use one of the conventional suffixes "
173 f"({', '.join(suffix for suffix, _, _ in reversed(_SUFFIXES))}), "
174 f"or state the encoding explicitly — see --help for the "
175 f"--format / --kind options this command offers."
176 )
178 effective_kind = inferred_kind if kind == ArtifactKind.AUTO else kind
179 effective_format = inferred_format if format == ArtifactFormat.AUTO else format
181 if effective_kind == ArtifactKind.SCENE and effective_format == ArtifactFormat.TEXT:
182 raise ArtifactNamingError(
183 "The PEX25D text format has no spelling for a resolved scene — it is the "
184 "unresolved format, and resolution is not reversible. Write the scene as "
185 "'pb' or 'textpb', or write the unresolved file as 'text'."
186 )
188 return ArtifactSpec(path=path,
189 kind=effective_kind,
190 format=effective_format,
191 gzipped=gzipped)
194@contextlib.contextmanager
195def open_artifact_read(spec: ArtifactSpec) -> Iterator[BinaryIO]:
196 """
197 Open an artifact for reading, as bytes.
199 Bytes rather than text even for the text formats: the PEX25D reader scans
200 decimal digits straight into scaled integers and wants to control decoding
201 itself (the format is UTF-8 by definition, so there is nothing to negotiate).
202 """
203 if spec.is_stdio:
204 # sys.__stdin__ rather than sys.stdin: when an artifact is written to
205 # stdout the process redirects sys.stdout to stderr so that logging can
206 # never corrupt the stream, and the artifact I/O must bypass that.
207 stream: BinaryIO = sys.__stdin__.buffer
208 if spec.gzipped:
209 with gzip.GzipFile(fileobj=stream, mode='rb') as f:
210 yield cast(BinaryIO, f)
211 else:
212 yield stream
213 return
215 opener = gzip.open if spec.gzipped else open
216 with opener(spec.path, 'rb') as f:
217 yield cast(BinaryIO, f)
220@contextlib.contextmanager
221def open_artifact_write(spec: ArtifactSpec) -> Iterator[BinaryIO]:
222 """
223 Open an artifact for writing, as bytes.
225 Note that when the destination is stdout, *nothing else may be written
226 there* — diagnostics and progress go to stderr. See
227 :func:`klayout_pex.pex25d.diagnostics.diagnostics_stream`.
228 """
229 if spec.is_stdio:
230 # The real stdout, not the redirected sys.stdout — see open_artifact_read.
231 stream: BinaryIO = sys.__stdout__.buffer
232 if spec.gzipped:
233 with gzip.GzipFile(fileobj=stream, mode='wb') as f:
234 yield cast(BinaryIO, f)
235 else:
236 yield stream
237 stream.flush()
238 return
240 parent = os.path.dirname(os.path.abspath(spec.path))
241 os.makedirs(parent, exist_ok=True)
243 opener = gzip.open if spec.gzipped else open
244 with opener(spec.path, 'wb') as f:
245 yield cast(BinaryIO, f)
248def derive_path(source_path: str,
249 kind: ArtifactKind,
250 format: ArtifactFormat) -> str:
251 """
252 Derive a sibling artifact path from an existing one, e.g. ``foo.pex25d`` →
253 ``foo.pex25d.scene.pb``. Used only where a verb offers a default output
254 location; a verb never writes to a derived path without saying so.
255 """
256 if source_path == STDIO_PATH:
257 return STDIO_PATH
259 name = os.path.basename(source_path)
260 if name.endswith('.gz'):
261 name = name[:-len('.gz')]
262 for suffix, _, _ in _SUFFIXES:
263 if name.endswith(suffix):
264 name = name[:-len(suffix)]
265 break
266 else:
267 name = os.path.splitext(name)[0]
269 return os.path.join(os.path.dirname(source_path),
270 name + CANONICAL_SUFFIX[(kind, format)])