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

1from enum import Enum 

2from pathlib import Path 

3from typing import List, Optional, Union 

4 

5import pydantic 

6from agentlib.core.agent import AgentConfig 

7from agentlib.core.errors import ConfigurationError 

8from agentlib_mpc.data_structures.mpc_datamodels import MPCVariable 

9from pydantic import ConfigDict, model_validator 

10 

11from agentlib_flexquant.data_structures.mpcs import ( 

12 BaselineMPCData, 

13 NFMPCData, 

14 PFMPCData 

15) 

16 

17 

18class ForcedOffers(Enum): 

19 positive = "positive" 

20 negative = "negative" 

21 

22 

23class ShadowMPCConfigGeneratorConfig(pydantic.BaseModel): 

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

25 model_config = ConfigDict( 

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

27 extra='forbid' 

28 ) 

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

30 default=[], 

31 description="Name and value of weights", 

32 ) 

33 pos_flex: PFMPCData = pydantic.Field( 

34 default=None, 

35 description="Data for PF-MPC" 

36 ) 

37 neg_flex: NFMPCData = pydantic.Field( 

38 default=None, 

39 description="Data for NF-MPC" 

40 ) 

41 @model_validator(mode="after") 

42 def assign_weights_to_flex(self): 

43 if self.pos_flex is None: 

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

45 if self.neg_flex is None: 

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

47 if self.weights: 

48 self.pos_flex.weights = self.weights 

49 self.neg_flex.weights = self.weights 

50 return self 

51 

52 

53class FlexibilityMarketConfig(pydantic.BaseModel): 

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

55 model_config = ConfigDict( 

56 extra='forbid' 

57 ) 

58 agent_config: AgentConfig 

59 name_of_created_file: str = pydantic.Field( 

60 default="flexibility_market.json", 

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

62 ) 

63 

64 

65class FlexibilityIndicatorConfig(pydantic.BaseModel): 

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

67 model_config = ConfigDict( 

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

69 extra='forbid' 

70 ) 

71 agent_config: AgentConfig 

72 name_of_created_file: str = pydantic.Field( 

73 default="indicator.json", 

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

75 ) 

76 @model_validator(mode="after") 

77 def check_file_extension(self): 

78 if self.name_of_created_file: 

79 file_path = Path(self.name_of_created_file) 

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

81 raise ConfigurationError( 

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

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

84 ) 

85 return self 

86 

87 

88class FlexQuantConfig(pydantic.BaseModel): 

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

90 model_config = ConfigDict( 

91 json_encoders={Path: str}, 

92 extra='forbid' 

93 ) 

94 prep_time: int = pydantic.Field( 

95 default=1800, 

96 ge=0, 

97 unit="s", 

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

99 ) 

100 flex_event_duration: int = pydantic.Field( 

101 default=7200, 

102 ge=0, 

103 unit="s", 

104 description="Flexibility event duration", 

105 ) 

106 market_time: int = pydantic.Field( 

107 default=900, 

108 ge=0, 

109 unit="s", 

110 description="Time for market interaction", 

111 ) 

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

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

114 ) 

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

116 default=None, 

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

118 ) 

119 baseline_config_generator_data: BaselineMPCData = pydantic.Field( 

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

121 ) 

122 shadow_mpc_config_generator_data: ShadowMPCConfigGeneratorConfig = pydantic.Field( 

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

124 ) 

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

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

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

128 ) 

129 flex_files_directory: Path = pydantic.Field( 

130 default=Path("created_flex_files"), 

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

132 ) 

133 results_directory: Path = pydantic.Field( 

134 default=Path("results"), 

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

136 ) 

137 delete_files: bool = pydantic.Field( 

138 default=True, 

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

140 ) 

141 overwrite_files: bool = pydantic.Field( 

142 default=False, 

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

144 ) 

145 

146 @model_validator(mode="after") 

147 def check_config_file_extension(self): 

148 """ 

149 Validates that the indicator and market config file paths have a '.json' extension. 

150  

151 Raises: 

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

153 """ 

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

155 raise ValueError( 

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

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

158 ) 

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

160 raise ValueError( 

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

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

163 ) 

164 return self 

165 

166 @model_validator(mode="after") 

167 def adapt_paths_and_create_directory(self): 

168 """ 

169 Adjusts and ensures the directory structure for flex file generation and results storage. 

170 

171 This method: 

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

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

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

175 """ 

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

177 self.flex_files_directory = ( 

178 self.flex_base_directory_path 

179 / self.flex_files_directory.name 

180 ) 

181 self.results_directory = ( 

182 self.flex_base_directory_path 

183 / self.results_directory.name 

184 ) 

185 # create directories if not already existing 

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

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

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

189 return self