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

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# 

24 

25""" 

26Naming, format inference and stdio handling for PEX25D artifacts. 

27 

28There are two orthogonal properties of every PEX25D artifact on disk: 

29 

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. 

33 

34``format`` 

35 How that message is encoded — the normative PEX25D *text* format, binary 

36 protobuf, or protobuf text format. 

37 

38Both are inferred from the file name, and either can be overridden 

39(``--kind`` / ``--format``). The naming convention is:: 

40 

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 

46 

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""" 

51 

52from __future__ import annotations 

53 

54import contextlib 

55import gzip 

56import io 

57import os 

58import sys 

59from dataclasses import dataclass 

60from enum import StrEnum 

61from typing import * 

62 

63 

64# A single '-' as a path means stdin (for inputs) or stdout (for outputs). 

65STDIO_PATH = '-' 

66 

67 

68class ArtifactKind(StrEnum): 

69 """Which PEX25D message an artifact holds.""" 

70 

71 AUTO = 'auto' # infer from the path 

72 FILE = 'file' # kpex.pex25d.PEX25DFile — literal, unresolved 

73 SCENE = 'scene' # kpex.pex25d.PEX25DScene — resolved, adapter-ready 

74 

75 

76class ArtifactFormat(StrEnum): 

77 """How a PEX25D message is encoded.""" 

78 

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 

83 

84 

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] 

96 

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} 

102 

103 

104class ArtifactNamingError(ValueError): 

105 """The kind or format of an artifact could not be determined.""" 

106 

107 

108@dataclass(frozen=True) 

109class ArtifactSpec: 

110 """A fully determined artifact: where it lives, what it holds, how it is encoded.""" 

111 

112 path: str 

113 kind: ArtifactKind 

114 format: ArtifactFormat 

115 gzipped: bool = False 

116 

117 @property 

118 def is_stdio(self) -> bool: 

119 return self.path == STDIO_PATH 

120 

121 @property 

122 def is_binary(self) -> bool: 

123 return self.format == ArtifactFormat.PB or self.gzipped 

124 

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 ''})" 

128 

129 

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 

142 

143 

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. 

152 

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``. 

157 

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 

164 

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 ) 

177 

178 effective_kind = inferred_kind if kind == ArtifactKind.AUTO else kind 

179 effective_format = inferred_format if format == ArtifactFormat.AUTO else format 

180 

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 ) 

187 

188 return ArtifactSpec(path=path, 

189 kind=effective_kind, 

190 format=effective_format, 

191 gzipped=gzipped) 

192 

193 

194@contextlib.contextmanager 

195def open_artifact_read(spec: ArtifactSpec) -> Iterator[BinaryIO]: 

196 """ 

197 Open an artifact for reading, as bytes. 

198 

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 

214 

215 opener = gzip.open if spec.gzipped else open 

216 with opener(spec.path, 'rb') as f: 

217 yield cast(BinaryIO, f) 

218 

219 

220@contextlib.contextmanager 

221def open_artifact_write(spec: ArtifactSpec) -> Iterator[BinaryIO]: 

222 """ 

223 Open an artifact for writing, as bytes. 

224 

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 

239 

240 parent = os.path.dirname(os.path.abspath(spec.path)) 

241 os.makedirs(parent, exist_ok=True) 

242 

243 opener = gzip.open if spec.gzipped else open 

244 with opener(spec.path, 'wb') as f: 

245 yield cast(BinaryIO, f) 

246 

247 

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 

258 

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] 

268 

269 return os.path.join(os.path.dirname(source_path), 

270 name + CANONICAL_SUFFIX[(kind, format)])