Coverage for klayout_pex/klayout/lvs_cache.py: 97%

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

25Cached LVS databases, which are reused only while the inputs of the LVS run are unchanged 

26(see protos/kpex/klayout/lvs_input_fingerprint.proto) 

27""" 

28 

29from __future__ import annotations 

30 

31from dataclasses import dataclass 

32import hashlib 

33import os 

34from pathlib import PurePath 

35import shutil 

36import tempfile 

37from typing import * 

38 

39import google.protobuf.json_format 

40 

41from klayout_pex_protobuf.kpex.klayout.lvs_input_fingerprint_pb2 import ( 

42 FileFingerprint, 

43 LVSDeckFingerprint, 

44 LVSInputFingerprint 

45) 

46 

47 

48# NOTE: increment whenever the fingerprint is computed differently, 

49# so that the fingerprints written by an older KPEX never match 

50FINGERPRINT_FORMAT_VERSION = 1 

51 

52# NOTE: an upgrade may change every file of a deck, only name a few of them 

53MAX_NAMED_DECK_FILES = 5 

54 

55 

56def file_sha256(path: str) -> str: 

57 h = hashlib.sha256() 

58 with open(path, 'rb') as f: 

59 for chunk in iter(lambda: f.read(1 << 20), b''): 

60 h.update(chunk) 

61 return h.hexdigest() 

62 

63 

64def file_fingerprint(path: str) -> FileFingerprint: 

65 return FileFingerprint(path=os.path.abspath(path), 

66 sha256=file_sha256(path)) 

67 

68 

69def deck_fingerprint(lvs_script_path: str) -> LVSDeckFingerprint: 

70 """ 

71 Every file in the directory of the LVS script and below, as the script includes further files 

72 (%include, require, load) in ways that are not practical to trace. 

73 

74 Hidden files (e.g. .DS_Store) are skipped, as are dangling symbolic links. 

75 """ 

76 script_path = os.path.abspath(lvs_script_path) 

77 deck_dir_path = os.path.dirname(script_path) 

78 fingerprint = LVSDeckFingerprint(script_path=script_path) 

79 for dir_path, dir_names, file_names in os.walk(deck_dir_path, followlinks=True): 

80 dir_names[:] = [n for n in dir_names if not n.startswith('.')] 

81 for file_name in file_names: 

82 path = os.path.join(dir_path, file_name) 

83 if file_name.startswith('.') or not os.path.isfile(path): 

84 continue 

85 relative_path = PurePath(os.path.relpath(path, deck_dir_path)).as_posix() 

86 fingerprint.sha256_by_relative_path[relative_path] = file_sha256(path) 

87 return fingerprint 

88 

89 

90def lvs_input_fingerprint(pdk: str, 

91 klayout_version: str, 

92 lvs_script_path: str, 

93 script_parameters: Dict[str, str], 

94 gds_path: str, 

95 cell_name: str, 

96 schematic_path: str) -> LVSInputFingerprint: 

97 fingerprint = LVSInputFingerprint(format_version=FINGERPRINT_FORMAT_VERSION, 

98 pdk=pdk, 

99 klayout_version=klayout_version, 

100 cell_name=cell_name) 

101 fingerprint.deck.CopyFrom(deck_fingerprint(lvs_script_path)) 

102 fingerprint.script_parameters.update(script_parameters) 

103 fingerprint.layout.CopyFrom(file_fingerprint(gds_path)) 

104 fingerprint.schematic.CopyFrom(file_fingerprint(schematic_path)) 

105 return fingerprint 

106 

107 

108def fingerprint_differences(cached: LVSInputFingerprint, 

109 current: LVSInputFingerprint) -> List[str]: 

110 """ 

111 What changed between the inputs of the cached LVS run and those of the current one, 

112 empty if the cached LVS database can be reused. Paths are not compared. 

113 """ 

114 if cached.format_version != current.format_version: 

115 return [f"the fingerprint format changed ({cached.format_version} → {current.format_version})"] 

116 

117 differences: List[str] = [] 

118 

119 def compare(what: str, cached_value: str, current_value: str): 

120 if cached_value != current_value: 

121 differences.append(f"{what} changed ({cached_value or 'unknown'} → {current_value or 'unknown'})") 

122 

123 compare('the PDK', cached.pdk, current.pdk) 

124 compare('the KLayout version', cached.klayout_version, current.klayout_version) 

125 

126 cached_files = dict(cached.deck.sha256_by_relative_path) 

127 current_files = dict(current.deck.sha256_by_relative_path) 

128 changed_files = sorted(p for p in cached_files.keys() | current_files.keys() 

129 if cached_files.get(p) != current_files.get(p)) 

130 if changed_files: 

131 named = ', '.join(changed_files[:MAX_NAMED_DECK_FILES]) 

132 more = len(changed_files) - MAX_NAMED_DECK_FILES 

133 differences.append(f"the LVS deck changed ({named}{f' and {more} more files' if more > 0 else ''})") 

134 

135 cached_parameters = dict(cached.script_parameters) 

136 current_parameters = dict(current.script_parameters) 

137 changed_parameters = [f"{n}: {cached_parameters.get(n, 'unset')} → {current_parameters.get(n, 'unset')}" 

138 for n in sorted(cached_parameters.keys() | current_parameters.keys()) 

139 if cached_parameters.get(n) != current_parameters.get(n)] 

140 if changed_parameters: 

141 differences.append(f"the LVS script parameters changed ({', '.join(changed_parameters)})") 

142 

143 if cached.layout.sha256 != current.layout.sha256: 

144 differences.append("the input layout changed") 

145 compare('the cell', cached.cell_name, current.cell_name) 

146 if cached.schematic.sha256 != current.schematic.sha256: 

147 differences.append("the schematic changed") 

148 

149 return differences 

150 

151 

152def read_fingerprint(path: str) -> Optional[LVSInputFingerprint]: 

153 """ 

154 None if there is none, or it can't be read (e.g. written by an incompatible KPEX version) 

155 """ 

156 try: 

157 with open(path, 'r', encoding='utf-8') as f: 

158 return google.protobuf.json_format.Parse(f.read(), LVSInputFingerprint()) 

159 except (OSError, google.protobuf.json_format.ParseError): 

160 return None 

161 

162 

163def write_fingerprint(path: str, fingerprint: LVSInputFingerprint): 

164 json_str = google.protobuf.json_format.MessageToJson(fingerprint, 

165 preserving_proto_field_name=True, 

166 sort_keys=True, 

167 ensure_ascii=False) 

168 with open(path, 'w', encoding='utf-8') as f: 

169 f.write(json_str) 

170 

171 

172@dataclass(frozen=True) 

173class LVSCacheEntry: 

174 """ 

175 The cached LVS database of a cell, and the fingerprint of the LVS inputs it was created from 

176 """ 

177 dir_path: str 

178 cell_name: str 

179 

180 @property 

181 def lvsdb_path(self) -> str: 

182 return os.path.join(self.dir_path, f"{self.cell_name}.lvsdb.gz") 

183 

184 @property 

185 def fingerprint_path(self) -> str: 

186 return os.path.join(self.dir_path, f"{self.cell_name}.lvs_fingerprint.pb.json") 

187 

188 def cache_miss_reasons(self, fingerprint: LVSInputFingerprint) -> List[str]: 

189 """ 

190 Why the cached LVS database can't be reused, empty if it can 

191 """ 

192 if not os.path.isfile(self.lvsdb_path): 

193 return ["there is no cached LVSDB"] 

194 cached_fingerprint = read_fingerprint(self.fingerprint_path) 

195 if cached_fingerprint is None: 

196 return ["the cached LVSDB has no fingerprint of its LVS inputs"] 

197 return fingerprint_differences(cached_fingerprint, fingerprint) 

198 

199 def store(self, lvsdb_path: str, fingerprint: LVSInputFingerprint): 

200 os.makedirs(self.dir_path, exist_ok=True) 

201 # NOTE: the fingerprint is written last, so an interrupted store leaves an entry without one, 

202 # which is never reused 

203 if os.path.exists(self.fingerprint_path): 

204 os.remove(self.fingerprint_path) 

205 # NOTE: each file is written beside its place and renamed into it, so that a concurrent run 

206 # reading the entry (e.g. parallel tests sharing the cache) never reads a partly written file 

207 self._replace(self.lvsdb_path, lambda tmp_path: shutil.copyfile(lvsdb_path, tmp_path)) 

208 self._replace(self.fingerprint_path, lambda tmp_path: write_fingerprint(tmp_path, fingerprint)) 

209 

210 def _replace(self, path: str, write: Callable[[str], Any]): 

211 fd, tmp_path = tempfile.mkstemp(dir=self.dir_path, prefix=f".{os.path.basename(path)}.", suffix='.tmp') 

212 os.close(fd) 

213 try: 

214 write(tmp_path) 

215 os.replace(tmp_path, path) 

216 except BaseException: 

217 os.remove(tmp_path) 

218 raise