Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / distributed / encoder_parallel / config.py: 82%

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"""User configuration for the visual encoder parallel runtime.""" 

16 

17from __future__ import annotations 

18 

19from dataclasses import dataclass, fields 

20from typing import Any, Mapping 

21 

22 

23@dataclass(frozen=True) 

24class EncoderParallelConfig: 

25 """Configuration of a derived Encoder-DP/Encoder-CP domain.""" 

26 

27 enabled: bool = False 

28 encoder_cp_size: int = 1 

29 

30 def __post_init__(self) -> None: 

31 if not isinstance(self.enabled, bool): 

32 raise ValueError("enabled must be a boolean") 

33 if (isinstance(self.encoder_cp_size, bool) or not isinstance(self.encoder_cp_size, int) 

34 or self.encoder_cp_size < 1): 

35 raise ValueError( 

36 "accelerator.encoder_parallel.encoder_cp_size " 

37 f"must be a positive integer, got {self.encoder_cp_size!r}" 

38 ) 

39 

40 @classmethod 

41 def from_mapping(cls, raw: Mapping[str, Any] | None) -> "EncoderParallelConfig": 

42 """Parse a mapping while rejecting unknown configuration keys. 

43 

44 Args: 

45 raw: User's optional encoder-parallel configuration mapping. 

46 """ 

47 if raw is None: 

48 return cls() 

49 if not isinstance(raw, Mapping): 

50 raise ValueError("encoder_parallel must be a mapping") 

51 unknown = set(raw) - {field.name for field in fields(cls)} 

52 if unknown: 

53 raise ValueError(f"unknown encoder_parallel keys: {sorted(unknown)}") 

54 return cls(**raw)