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

1""" 

2Pydantic data models for FlexQuant configuration and validation. 

3""" 

4# from enum import Enum 

5from pathlib import Path 

6from typing import Optional, Union 

7 

8import pydantic 

9from pydantic import field_validator, ConfigDict, model_validator, Field, BaseModel 

10from agentlib.core.agent import AgentConfig 

11from agentlib.core.errors import ConfigurationError 

12from agentlib_mpc.data_structures.mpc_datamodels import MPCVariable 

13 

14from agentlib_flexquant.data_structures.mpcs import ( 

15 BaselineMPCData, 

16 NFMPCData, 

17 PFMPCData, 

18) 

19 

20 

21# class ForcedOffers(Enum): 

22# positive = "positive" 

23# negative = "negative" 

24 

25 

26class ShadowMPCConfigGeneratorConfig(BaseModel): 

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

28 

29 model_config = ConfigDict( 

30 json_encoders={MPCVariable: lambda v: v.dict()}, extra="forbid" 

31 ) 

32 weights: list[MPCVariable] = Field( 

33 default=[], description="Name and value of weights", 

34 ) 

35 pos_flex: PFMPCData = Field(default=None, description="Data for PF-MPC") 

36 neg_flex: NFMPCData = Field(default=None, description="Data for NF-MPC") 

37 

38 @model_validator(mode="after") 

39 def assign_weights_to_flex(self): 

40 """Validate flexibility cost function fields and assign weights to them.""" 

41 if self.pos_flex is None: 

42 raise ValueError( 

43 "Missing required field: 'pos_flex' specifying the pos flex cost function." 

44 ) 

45 if self.neg_flex is None: 

46 raise ValueError( 

47 "Missing required field: 'neg_flex' specifying the neg flex cost function." 

48 ) 

49 if self.weights: 

50 self.pos_flex.weights = self.weights 

51 self.neg_flex.weights = self.weights 

52 return self 

53 

54 

55class FlexibilityMarketConfig(BaseModel): 

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

57 

58 model_config = ConfigDict(extra="forbid") 

59 agent_config: AgentConfig 

60 name_of_created_file: str = Field( 

61 default="flexibility_market.json", 

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

63 ) 

64 

65 

66class FlexibilityIndicatorConfig(BaseModel): 

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

68 

69 model_config = ConfigDict( 

70 json_encoders={Path: str, AgentConfig: lambda v: v.model_dump()}, extra="forbid" 

71 ) 

72 agent_config: AgentConfig 

73 name_of_created_file: str = Field( 

74 default="indicator.json", 

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

76 ) 

77 

78 @model_validator(mode="after") 

79 def check_file_extension(self): 

80 """Validate that name_of_created_file has a .json extension.""" 

81 if self.name_of_created_file: 

82 file_path = Path(self.name_of_created_file) 

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

84 raise ConfigurationError( 

85 f"Invalid file extension for " 

86 f"name_of_created_file: '{self.name_of_created_file}'. " 

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

88 ) 

89 return self 

90 

91 

92class FlexQuantConfig(BaseModel): 

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

94 

95 model_config = ConfigDict(json_encoders={Path: str}, extra="forbid") 

96 prep_time: int = Field( 

97 default=1800, 

98 ge=0, 

99 unit="s", 

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

101 ) 

102 flex_event_duration: int = Field( 

103 default=7200, ge=0, unit="s", description="Flexibility event duration", 

104 ) 

105 market_time: int = Field( 

106 default=900, ge=0, unit="s", description="Time for market interaction", 

107 ) 

108 indicator_config: Union[FlexibilityIndicatorConfig, Path] = Field( 

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

110 ) 

111 market_config: Optional[Union[FlexibilityMarketConfig, Path]] = Field( 

112 default=None, description="Path to the file or dict of market config", 

113 ) 

114 baseline_config_generator_data: BaselineMPCData = Field( 

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

116 ) 

117 shadow_mpc_config_generator_data: ShadowMPCConfigGeneratorConfig = Field( 

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

119 ) 

120 casadi_sim_time_step: int = Field( 

121 default=0, 

122 description="Simulate over the prediction horizon with a defined resolution using Casadi " 

123 "simulator. " 

124 "Only use it when the power depends on the states. Don't use it when power " 

125 "itself is the control variable." 

126 "Set to 0 to skip simulation", 

127 ) 

128 flex_base_directory_path: Optional[Path] = Field( 

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

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

131 ) 

132 flex_files_directory: Path = Field( 

133 default=Path("created_flex_files"), 

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

135 ) 

136 results_directory: Path = Field( 

137 default=Path("results"), 

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

139 ) 

140 delete_files: bool = Field( 

141 default=True, description="If generated files should be deleted afterwards", 

142 ) 

143 overwrite_files: bool = Field( 

144 default=False, 

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

146 ) 

147 

148 @model_validator(mode="after") 

149 def check_config_file_extension(self): 

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

151 

152 Raises: 

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

154 

155 """ 

156 if ( 

157 isinstance(self.indicator_config, Path) 

158 and self.indicator_config.suffix != ".json" 

159 ): 

160 raise ValueError( 

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

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

163 ) 

164 if ( 

165 isinstance(self.market_config, Path) 

166 and self.market_config.suffix != ".json" 

167 ): 

168 raise ValueError( 

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

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

171 ) 

172 return self 

173 

174 @field_validator('casadi_sim_time_step', mode='after') 

175 @classmethod 

176 def is_none_negative_integer(cls, value: int) -> int: 

177 if value < 0: 

178 raise ValueError(f'{value} is not a non-negative integer') 

179 return value 

180 

181 @model_validator(mode="after") 

182 def adapt_paths_and_create_directory(self): 

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

184 

185 This method: 

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

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

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

189 

190 """ 

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

192 self.flex_files_directory = ( 

193 self.flex_base_directory_path / self.flex_files_directory.name 

194 ) 

195 self.results_directory = ( 

196 self.flex_base_directory_path / self.results_directory.name 

197 ) 

198 # create directories if not already existing 

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

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

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

202 return self