Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / auto_parallel / sapp_nd / nd / common / config.py: 88%

1# Copyright 2025 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"""parallel dimensions""" 

16 

17import os 

18import copy 

19import json 

20import yaml 

21import toml 

22 

23from hyper_parallel.auto_parallel.sapp_nd.nd.logger import logger 

24 

25 

26class YamlObject: 

27 """Attributed dictionary""" 

28 

29 def from_dict(self, field): 

30 """init from dictionary""" 

31 for k in field: 

32 if isinstance(field[k], dict): 

33 setattr(self, k, YamlObject(field[k])) 

34 else: 

35 setattr(self, k, field[k]) 

36 

37 def __init__(self, field): 

38 self.from_dict(field) 

39 

40 def __getattr__(self, attr): 

41 if attr not in self.__dict__: 

42 logger.warning( 

43 "Attribute %s does not exist. Value '0' will be assigned.", 

44 str(attr), 

45 ) 

46 return 0 

47 return self.__dict__[attr] 

48 

49 def __copy__(self): 

50 cls = self.__class__ 

51 res = cls.__new__(cls) 

52 res.__dict__.update(self.__dict__) 

53 return res 

54 

55 def __deepcopy__(self, memo): 

56 cls = self.__class__ 

57 res = cls.__new__(cls) 

58 for k, v in self.__dict__.items(): 

59 setattr(res, k, copy.deepcopy(v, memo)) 

60 return res 

61 

62 def __getstate__(self): 

63 return self.to_dict() 

64 

65 def __setstate__(self, field): 

66 return self.from_dict(field) 

67 

68 def to_dict(self): 

69 """Transform from Config to a strict dict class""" 

70 return_dict = {} 

71 for key, val in self.__dict__.items(): 

72 if isinstance(val, YamlObject): 

73 val = val.to_dict() 

74 return_dict[key] = val 

75 return return_dict 

76 

77 def dump(self, file_name, folder=None): 

78 """Dump to given file in given folder""" 

79 if not folder: 

80 folder = os.path.dirname(__file__) 

81 full_file_name = os.path.join(folder, "config_" + file_name) + ".yaml" 

82 with open(full_file_name, "w", encoding="utf-8") as outfile: 

83 yaml.dump(self.to_dict(), outfile, default_flow_style=False) 

84 

85 

86class Config(YamlObject): 

87 """Yaml config""" 

88 

89 def __init__(self, input_config): 

90 if isinstance(input_config, str): 

91 with open(input_config, encoding="utf-8") as f: 

92 if input_config.endswith("yaml"): 

93 try: 

94 super().__init__(yaml.safe_load(f)) 

95 except yaml.YAMLError as exc: 

96 print(exc) 

97 logger.warning(exc) 

98 elif input_config.endswith("json"): 

99 super().__init__(json.load(f)) 

100 elif input_config.endswith("toml"): 

101 super().__init__(toml.load(f)) 

102 else: 

103 logger.warning("Current handled file formats: YAML, JSON") 

104 elif isinstance(input_config, Config): 

105 super().__init__(input_config.__dict__) 

106 elif isinstance(input_config, dict): 

107 super().__init__(input_config) 

108 else: 

109 raise TypeError("Expecting path string or Config object") 

110 

111 def __str__(self): 

112 return str( 

113 { 

114 k: (v if not isinstance(v, Config) else vars(v)) 

115 for k, v in vars(self).items() 

116 } 

117 ) 

118 

119 # def global_batch_size(self): 

120 # """Compute the global batch size""" 

121 # pp = Dim.PP.from_config(self) 

122 # dp = Dim.DP.from_config(self) 

123 # mbs = Dim.MBS.from_config(self) 

124 # gbs = dp * mbs 

125 # if pp == 1: 

126 # logger.debug("global_batch_size = DP(%d) * MBS(%d)", dp, mbs) 

127 # # gas = Dim.GAS.from_config(self) 

128 # # gbs *= gas 

129 # else: 

130 # mbn = Dim.MBN.from_config(self) 

131 # gbs *= mbn 

132 # logger.debug( 

133 # "global_batch_size = DP(%d) * MB(%d) * MBS(%d)", dp, mbn, mbs 

134 # ) 

135 # return gbs 

136 

137 def set_par(self, parallel_dimensions): 

138 """Set the given parallelization""" 

139 for dim in parallel_dimensions.dims_val: 

140 dim.to_config(self, parallel_dimensions.val(dim)) 

141 

142 # def layer_num(self): 

143 # """Get the layer number""" 

144 # layers = self.model.model_config.num_layers 

145 # if layers and layers is not None: 

146 # return layers 

147 # return self.model.model_config.num_hidden_layers 

148 

149 # def name(self): 

150 # """Get the model name""" 

151 # return self.trainer.model_name