Coverage for src/distopf/cim_importer/processors/base_processor.py: 82%

1from abc import ABC, abstractmethod 

2import numpy as np 

3import cimgraph.data_profile.cimhub_2023 as cim 

4from cimgraph.models import FeederModel 

5 

6 

7class BaseProcessor(ABC): 

8 """Base class for all component processors.""" 

9 

10 def __init__(self, s_base: float = 1e6): 

11 self.s_base = s_base 

12 

13 @abstractmethod 

14 def process(self, network: FeederModel) -> list[dict]: 

15 """Process components and return list of dictionaries.""" 

16 pass 

17 

18 def process_branch(self, network: FeederModel) -> list[dict]: 

19 """Process branch data. Used by regulator_processor to produce branch data entries.""" 

20 return self.process(network) 

21 

22 def _create_base_branch_dict(self) -> dict: 

23 """Create base dictionary structure for branch data.""" 

24 return { 

25 "name": None, 

26 "fb": None, 

27 "tb": None, 

28 "from_name": None, 

29 "to_name": None, 

30 "raa": 0.0, 

31 "rab": 0.0, 

32 "rac": 0.0, 

33 "rbb": 0.0, 

34 "rbc": 0.0, 

35 "rcc": 0.0, 

36 "xaa": 0.0, 

37 "xab": 0.0, 

38 "xac": 0.0, 

39 "xbb": 0.0, 

40 "xbc": 0.0, 

41 "xcc": 0.0, 

42 "type": None, 

43 "status": None, 

44 "s_base": self.s_base, 

45 "v_ln_base": None, 

46 "z_base": None, 

47 "phases": None, 

48 "length": None, 

49 } 

50 

51 def _get_terminals_info(self, equipment) -> tuple[str, str]: 

52 """Get from and to bus names from equipment terminals.""" 

53 terminals = equipment.Terminals 

54 if len(terminals) < 2: 

55 raise ValueError(f"Equipment {equipment.name} has insufficient terminals") 

56 

57 from_bus = terminals[0].ConnectivityNode.name 

58 to_bus = terminals[1].ConnectivityNode.name 

59 return from_bus, to_bus 

60 

61 def _get_bus_voltage_base(self, node: cim.ConnectivityNode) -> float: 

62 """Get voltage base for bus from connected equipment.""" 

63 v_base = None 

64 for terminal in node.Terminals: 

65 if ( 

66 hasattr(terminal, "ConductingEquipment") 

67 and terminal.ConductingEquipment 

68 ): 

69 baseVoltage = terminal.ConductingEquipment.BaseVoltage 

70 if baseVoltage is not None: 

71 assert baseVoltage is not None 

72 v_base = float(baseVoltage.nominalVoltage) 

73 if v_base is None and hasattr(terminal, "TransformerEnd"): 

74 if len(terminal.TransformerEnd) == 1: 

75 v_base = float( 

76 terminal.TransformerEnd[0].BaseVoltage.nominalVoltage 

77 ) 

78 if hasattr(terminal, "BaseVoltage") and terminal.BaseVoltage: 

79 if ( 

80 hasattr(terminal.BaseVoltage, "nominalVoltage") 

81 and terminal.BaseVoltage.nominalVoltage 

82 ): 

83 v_base = float(terminal.BaseVoltage.nominalVoltage) 

84 if v_base is None: 

85 v_base = 0 

86 v_ln_base = v_base / np.sqrt(3) 

87 return v_ln_base