Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / trainer / config / stress_detect.py: 100%

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"""Dedicated configuration for online stress detection.""" 

16 

17from dataclasses import dataclass, field 

18from typing import Literal 

19 

20 

21@dataclass 

22class OpStressDetectConfig: 

23 """Configure operator-level NPU stress detection rounds. 

24 

25 Args: 

26 detection_interval: Completed optimizer steps between rounds; zero disables 

27 callback creation and registration. 

28 num_detections: Consecutive checks per detection type in each round. 

29 detect_type: AIC, HCCS, or both types in AIC-then-HCCS order. 

30 exit_training_on_sdc: Raise on hardware precision failure when true; 

31 otherwise log an error and continue training. 

32 """ 

33 

34 detection_interval: int = 0 

35 num_detections: int = 1 

36 detect_type: Literal["aic", "hccs", "all"] = "aic" 

37 exit_training_on_sdc: bool = False 

38 

39 def __post_init__(self) -> None: 

40 """Reject invalid stress detection settings at the configuration boundary.""" 

41 if ( 

42 isinstance(self.detection_interval, bool) 

43 or not isinstance(self.detection_interval, int) 

44 or self.detection_interval < 0 

45 ): 

46 raise ValueError("stress_detect.op_stress_detect.detection_interval must be a non-negative integer") 

47 if ( 

48 isinstance(self.num_detections, bool) 

49 or not isinstance(self.num_detections, int) 

50 or self.num_detections < 1 

51 ): 

52 raise ValueError("stress_detect.op_stress_detect.num_detections must be a positive integer") 

53 if self.detect_type not in ("aic", "hccs", "all"): 

54 raise ValueError("stress_detect.op_stress_detect.detect_type must be one of: aic, hccs, all") 

55 if not isinstance(self.exit_training_on_sdc, bool): 

56 raise ValueError("stress_detect.op_stress_detect.exit_training_on_sdc must be a boolean") 

57 

58 

59@dataclass 

60class ModelStressDetectConfig: 

61 """Reserve independent model-level stress detection settings. 

62 

63 Args: 

64 detection_interval: Completed optimizer steps between rounds. Only zero 

65 is currently supported because model-level detection is not implemented. 

66 """ 

67 

68 detection_interval: int = 0 

69 

70 def __post_init__(self) -> None: 

71 """Reject invalid intervals and attempts to enable unimplemented detection.""" 

72 if ( 

73 isinstance(self.detection_interval, bool) 

74 or not isinstance(self.detection_interval, int) 

75 or self.detection_interval < 0 

76 ): 

77 raise ValueError("stress_detect.model_stress_detect.detection_interval must be a non-negative integer") 

78 if self.detection_interval > 0: 

79 raise NotImplementedError("stress_detect.model_stress_detect is reserved and not implemented yet") 

80 

81 

82@dataclass 

83class StressDetectConfig: 

84 """Group independent operator-level and model-level stress detection settings. 

85 

86 Args: 

87 op_stress_detect: Operator-level detection cadence and AIC/HCCS settings. 

88 model_stress_detect: Reserved model-level detection configuration. 

89 """ 

90 

91 op_stress_detect: OpStressDetectConfig = field(default_factory=OpStressDetectConfig) 

92 model_stress_detect: ModelStressDetectConfig = field(default_factory=ModelStressDetectConfig)