Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / codegen / source / compat.py: 0%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Sanitize modeling-source imports against the installed transformers. 

16 

17Generated modeling files embed their origin ``modeling_*.py`` verbatim. For 

18trusted-remote-code models that file is the version shipped inside the 

19checkpoint — it reflects whatever transformers the checkpoint author used, not 

20the one the artifact runs against. When that installed transformers is *newer* 

21in a breaking way, the verbatim source can fail to import even though the 

22guarded behavior is now redundant. 

23 

24The compatibility fix handles ``is_torch_fx_available``, 

25removed from ``transformers.utils.import_utils`` in transformers v5.0.0. Its 

26only consumer guard is no longer load-bearing (torch >= 2.x imports ``torch.fx`` 

27unconditionally), so the importing statement is dropped and a module-scope 

28callable fallback is spliced in its place, keeping the original 

29``if is_torch_fx_available():`` guard resolving and truthful on transformers v4 

30and v5 alike, with the guard body byte-identical. 

31 

32Deliberately narrow — one symbol, AST-located edits, no-op for every source 

33that does not import it. Any other stale-import breakage is reported, not 

34papered over (a new compat rule is an explicit, documented decision). 

35""" 

36 

37from __future__ import annotations 

38 

39import ast 

40from dataclasses import dataclass 

41 

42from hyper_parallel.codegen.astkit.edits import TextEdit, apply_edits 

43from hyper_parallel.codegen.astkit.index import _line_offsets 

44 

45# The symbol transformers v5.0.0 removed. 

46_REMOVED = "is_torch_fx_available" 

47 

48# Spliced where the import statement was. The guard CALLS the symbol, so this 

49# must be a callable returning truthy — not a bare True (``if True():`` would 

50# raise ``'bool' object is not callable``). 

51_FALLBACK = ( 

52 "# Codegen source compatibility: transformers >= 5.0 removed\n" 

53 "# this name from ``transformers.utils.import_utils``; torch >= 2.x imports\n" 

54 "# ``torch.fx`` unconditionally, so this (4.x-era) source's guard is always\n" 

55 "# true. Keep it callable so the guard resolves and stays truthful here.\n" 

56 "def is_torch_fx_available():\n" 

57 " return True\n" 

58) 

59 

60 

61@dataclass 

62class _Stmt: 

63 """One ``from`` import statement, resolved to byte spans and symbol names.""" 

64 

65 start: int # byte offset of the ``from`` keyword 

66 end: int # byte offset just past the statement's trailing newline 

67 module: str 

68 symbols: list[tuple[str, str | None]] 

69 

70 

71def sanitize_source_compat(text: str) -> str: 

72 """Drop ``is_torch_fx_available`` imports from a modeling source. 

73 

74 Returns ``text`` unchanged when the source does not import the symbol (the 

75 common case — byte-for-byte). When a statement imports it, that statement is 

76 dropped and a module-scope callable fallback is spliced at its position (so 

77 the symbol is bound before any guard below). Other imports are untouched. 

78 """ 

79 offenders = [ 

80 stmt 

81 for stmt in _import_statements(text) 

82 if stmt.module in {"transformers.utils", "transformers.utils.import_utils"} 

83 and any(name == _REMOVED for name, _ in stmt.symbols) 

84 ] 

85 if not offenders: 

86 return text 

87 

88 edits = [TextEdit(stmt.start, stmt.end, _replacement(stmt)) for stmt in offenders] 

89 return apply_edits(text, edits) 

90 

91 

92def _replacement(stmt: _Stmt) -> str: 

93 """Render the surviving names from one affected import plus the fallback.""" 

94 remaining = [(name, alias) for name, alias in stmt.symbols if name != _REMOVED] 

95 if not remaining: 

96 return _FALLBACK 

97 rendered = ", ".join( 

98 f"{name} as {alias}" if alias is not None else name for name, alias in remaining 

99 ) 

100 return f"from {stmt.module} import {rendered}\n{_FALLBACK}" 

101 

102 

103# --- parse + locate offending import statements -------------------------------- 

104 

105 

106def _import_statements(text: str) -> list[_Stmt]: 

107 """Return top-level ``from ... import ...`` statements with source spans.""" 

108 try: 

109 tree = ast.parse(text) 

110 except SyntaxError: 

111 return [] 

112 

113 offsets = _line_offsets(text) 

114 statements = [] 

115 for node in tree.body: 

116 if not isinstance(node, ast.ImportFrom) or node.module is None: 

117 continue 

118 start = offsets[node.lineno - 1] + node.col_offset 

119 end = offsets[node.end_lineno - 1] + node.end_col_offset 

120 if end < len(text) and text[end] == "\n": 

121 end += 1 

122 statements.append( 

123 _Stmt( 

124 start=start, 

125 end=end, 

126 module=node.module, 

127 symbols=[(alias.name, alias.asname) for alias in node.names], 

128 ) 

129 ) 

130 return statements