Coverage for agentlib_flexquant/data_structures/flexquant.py: 93%

1import pydantic 

2from enum import Enum 

3from pathlib import Path 

4from typing import List, Optional, Union 

5from pydantic import ConfigDict, model_validator 

6from agentlib.core.agent import AgentConfig 

7from agentlib.core.errors import ConfigurationError 

8from agentlib_mpc.data_structures.mpc_datamodels import MPCVariable 

9from agentlib_flexquant.data_structures.mpcs import BaselineMPCData, NFMPCData, PFMPCData 

10 

11 

12class ForcedOffers(Enum): 

13 positive = "positive" 

14 negative = "negative" 

15 

16 

17class ShadowMPCConfigGeneratorConfig(pydantic.BaseModel): 

18 """Class defining the options to initialize the shadow mpc config generation.""" 

19 model_config = ConfigDict( 

20 json_encoders={MPCVariable: lambda v: v.dict()}, 

21 extra='forbid' 

22 ) 

23 weights: List[MPCVariable] = pydantic.Field( 

24 default=[], 

25 description="Name and value of weights", 

26 ) 

27 pos_flex: PFMPCData = pydantic.Field( 

28 default=None, 

29 description="Data for PF-MPC" 

30 ) 

31 neg_flex: NFMPCData = pydantic.Field( 

32 default=None, 

33 description="Data for NF-MPC" 

34 ) 

35 @model_validator(mode="after") 

36 def assign_weights_to_flex(self): 

37 if self.pos_flex is None: 

38 raise ValueError("Missing required field: 'pos_flex' specifying the pos flex cost function.") 

39 if self.neg_flex is None: 

40 raise ValueError("Missing required field: 'neg_flex' specifying the neg flex cost function.") 

41 if self.weights: 

42 self.pos_flex.weights = self.weights 

43 self.neg_flex.weights = self.weights 

44 return self 

45 

46 

47class FlexibilityMarketConfig(pydantic.BaseModel): 

48 """Class defining the options to initialize the market.""" 

49 model_config = ConfigDict( 

50 extra='forbid' 

51 ) 

52 agent_config: AgentConfig 

53 name_of_created_file: str = pydantic.Field( 

54 default="flexibility_market.json", 

55 description="Name of the config that is created by the generator", 

56 ) 

57 

58 

59class FlexibilityIndicatorConfig(pydantic.BaseModel): 

60 """Class defining the options for the flexibility indicators.""" 

61 model_config = ConfigDict( 

62 json_encoders={Path: str, AgentConfig: lambda v: v.model_dump()}, 

63 extra='forbid' 

64 ) 

65 agent_config: AgentConfig 

66 name_of_created_file: str = pydantic.Field( 

67 default="indicator.json", 

68 description="Name of the config that is created by the generator", 

69 ) 

70 @model_validator(mode="after") 

71 def check_file_extension(self): 

72 if self.name_of_created_file: 

73 file_path = Path(self.name_of_created_file) 

74 if file_path.suffix != ".json": 

75 raise ConfigurationError( 

76 f"Invalid file extension for name_of_created_file: '{self.name_of_created_file}'. " 

77 f"Expected a '.json' file." 

78 ) 

79 return self 

80 

81 

82class FlexQuantConfig(pydantic.BaseModel): 

83 """Class defining the options to initialize the FlexQuant generation.""" 

84 model_config = ConfigDict( 

85 json_encoders={Path: str}, 

86 extra='forbid' 

87 ) 

88 prep_time: int = pydantic.Field( 

89 default=1800, 

90 ge=0, 

91 unit="s", 

92 description="Preparation time before the flexibility event", 

93 ) 

94 flex_event_duration: int = pydantic.Field( 

95 default=7200, 

96 ge=0, 

97 unit="s", 

98 description="Flexibility event duration", 

99 ) 

100 market_time: int = pydantic.Field( 

101 default=900, 

102 ge=0, 

103 unit="s", 

104 description="Time for market interaction", 

105 ) 

106 indicator_config: Union[FlexibilityIndicatorConfig, Path] = pydantic.Field( 

107 description="Path to the file or dict of flexibility indicator config", 

108 ) 

109 market_config: Optional[Union[FlexibilityMarketConfig, Path]] = pydantic.Field( 

110 default=None, 

111 description="Path to the file or dict of market config", 

112 ) 

113 baseline_config_generator_data: BaselineMPCData = pydantic.Field( 

114 description="Baseline generator data config file or dict", 

115 ) 

116 shadow_mpc_config_generator_data: ShadowMPCConfigGeneratorConfig = pydantic.Field( 

117 description="Shadow mpc generator data config file or dict", 

118 ) 

119 flex_base_directory_path: Optional[Path] = pydantic.Field( 

120 default_factory=lambda: Path.cwd() / "flex_output_data", 

121 description="Base path where generated flex data is stored", 

122 ) 

123 flex_files_directory: Path = pydantic.Field( 

124 default=Path("created_flex_files"), 

125 description="Directory where generated files (jsons) should be stored", 

126 ) 

127 results_directory: Path = pydantic.Field( 

128 default=Path("results"), 

129 description="Directory where generated result files (CSVs) should be stored", 

130 ) 

131 delete_files: bool = pydantic.Field( 

132 default=True, 

133 description="If generated files should be deleted afterwards", 

134 ) 

135 overwrite_files: bool = pydantic.Field( 

136 default=False, 

137 description="If generated files should be overwritten by new files", 

138 ) 

139 

140 @model_validator(mode="after") 

141 def check_config_file_extension(self): 

142 """Validate that the indicator and market config file paths have a '.json' extension. 

143  

144 Raises: 

145 ValueError: If either file does not have the expected '.json' extension. 

146 

147 """ 

148 if isinstance(self.indicator_config, Path) and self.indicator_config.suffix != ".json": 

149 raise ValueError( 

150 f"Invalid file extension for indicator config: '{self.indicator_config}'. " 

151 f"Expected a '.json' file." 

152 ) 

153 if isinstance(self.market_config, Path) and self.market_config.suffix != ".json": 

154 raise ValueError( 

155 f"Invalid file extension for market config: '{self.market_config}'. " 

156 f"Expected a '.json' file." 

157 ) 

158 return self 

159 

160 @model_validator(mode="after") 

161 def adapt_paths_and_create_directory(self): 

162 """Adjust and ensure the directory structure for flex file generation and results storage. 

163 

164 This method: 

165 - Updates `flex_files_directory` and `results_directory` paths, so they are relative to 

166 the base flex directory, using only the directory names (ignoring any user-supplied paths). 

167 - Creates the base, flex files, and results directories if they do not already exist. 

168 

169 """ 

170 # adapt paths and use only names for user supplied data 

171 self.flex_files_directory = ( 

172 self.flex_base_directory_path 

173 / self.flex_files_directory.name 

174 ) 

175 self.results_directory = ( 

176 self.flex_base_directory_path 

177 / self.results_directory.name 

178 ) 

179 # create directories if not already existing 

180 self.flex_base_directory_path.mkdir(parents=True, exist_ok=True) 

181 self.flex_files_directory.mkdir(parents=True, exist_ok=True) 

182 self.results_directory.mkdir(parents=True, exist_ok=True) 

183 return self