Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / codegen / emit / replacement.py: 86%

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"""Translate native replacement observations for source emission.""" 

16from __future__ import annotations 

17 

18from typing import Any 

19 

20 

21def replacement_records_from_snapshot(snapshot: Any) -> tuple[dict[str, Any], ...]: 

22 """Translate native replacement observations into emission records. 

23 

24 The conversion consumes immutable observation data and never recompiles 

25 rules or inspects a live model. 

26 """ 

27 payload = snapshot._emission_view() if hasattr(snapshot, "_emission_view") else snapshot # pylint: disable=protected-access 

28 if not isinstance(payload, dict): 

29 raise TypeError("replacement snapshot must be a CodegenSnapshot or dict") 

30 replacements = payload.get("replacements", {}) 

31 declarations = replacements.get("declarations", ()) 

32 applications = replacements.get("applications", ()) 

33 declarations_by_id = {item["rule_id"]: item for item in declarations} 

34 if len(declarations_by_id) != len(declarations): 

35 raise ValueError("replacement snapshot contains duplicate rule_id values") 

36 records = [] 

37 for application in applications: 

38 rule_id = application.get("rule_id") 

39 declaration = declarations_by_id.get(rule_id) 

40 if declaration is None: 

41 raise ValueError( 

42 f"replacement application references unknown rule_id {rule_id!r}" 

43 ) 

44 source_identity = application.get("source_type") or declaration.get("source_type") 

45 if not isinstance(source_identity, str) or not source_identity: 

46 raise ValueError( 

47 f"replacement application {rule_id!r} has no resolved source_type" 

48 ) 

49 source_type = source_identity.replace(":", ".", 1) 

50 record = { 

51 "match": list(declaration["match"]), 

52 "fqn": application["primary_fqn"], 

53 "fqns": list(application["module_fqns"]), 

54 "module_type": source_type, 

55 "factory": application["replacement_type"].replace(":", ".", 1), 

56 "replacement_target": declaration["replacement_target"], 

57 "exact_type": bool(declaration["exact_type"]), 

58 } 

59 if declaration.get("static_arguments"): 

60 record["target_config"] = dict(declaration["static_arguments"]) 

61 records.append(record) 

62 return tuple(records) 

63 

64 

65__all__ = ["replacement_records_from_snapshot"]