import traceback import json from algorithmContract.dataContractType import DataContractType from algorithmContract.customDataContract import DataContract from algorithmContract.dataSource import DataSource from algorithmContract.dataFilter import DataFilter from algorithmContract.customFilter import CustomFilter from algorithmContract.configAnalysis import ConfigAnalysis from algorithmContract.graphSet import GraphSet class Contract: def __init__(self, contractType: DataContractType, dataContract: DataContract): self.contractType = contractType self.dataContract = dataContract def to_dict(self): return { "dataContractType": self.contractType.to_dict(), "dataContract": self.dataContract.to_dict() } def LoadAnalysisInput(jsonString: str): # Load the JSON string into a Python dictionary try: jsonData = json.loads(jsonString) jsonDataType = jsonData["dataContractType"] contractType = DataContractType( jsonDataType["type"], jsonDataType["version"]) jsonDataContract = jsonData["dataContract"] autoOrManual = jsonDataContract["autoOrManual"] # Initialize the entity classes using the JSON data # data_source = DataSource(jsonDataContract["dataSource"]["scada"]) custom_filters = {} for key, value in jsonDataContract["dataFilter"]["customFilter"].items(): custom_filter = CustomFilter(value.get("min"), value.get("max")) custom_filters[key] = custom_filter if (not jsonDataContract["dataFilter"]["turbines"] is None) and (not isinstance(jsonDataContract["dataFilter"]["turbines"],list)): raise Exception("输入参数中机组数据类型不合法.") data_filter = DataFilter( jsonDataContract["dataFilter"]["powerFarmID"], jsonDataContract["dataFilter"]["turbines"], jsonDataContract["dataFilter"]["dataBatchNum"], jsonDataContract["dataFilter"]["beginTime"], jsonDataContract["dataFilter"]["endTime"], jsonDataContract["dataFilter"]["excludingMonths"], custom_filters ) config_analysis = [] for item in jsonDataContract["configAnalysis"]: analysis = ConfigAnalysis( item["package"], item["className"], item["methodName"], item["scada"]) config_analysis.append(analysis) graph_sets = {} for key, value in jsonDataContract["graphSets"].items(): graph_set = GraphSet(value["step"], value["min"], value["max"]) graph_sets[key] = graph_set data_contract = DataContract(autoOrManual, data_filter, config_analysis, graph_sets) contract = Contract(contractType, data_contract) except Exception as e: print("exception: {}".format(e)) traceback.print_exc() # 打印异常的堆栈跟踪 raise e return contract # JSON string representing the data contract jsonString = ''' { "dataContractType": { "type": "analysisExecuteOrder", "version": "1.2.0" }, "dataContract": { "autoOrManual":"automatic", "dataSource": { "scada": "second" }, "dataFilter": { "powerFarmID": "010-00001", "turbines": [ "010-00001-0001", "010-00001-0002" ], "dataBatchNum": "B2024042211-0", "beginTime": "2023-01-01 00:00:00", "endTime": "2023-12-31 23:59:59", "excludingMonths": [ "2023-12", "2023-09" ], "customFilter": { "valueWindSpeed": { "min": 3.0, "max": 25.0 }, "valuePitchAngle": { "min": 2, "max": null }, "valueActivePower": { "min": 10, "max": 2500 }, "valueGeneratorSpeed": { "min": 10, "max": 2500 } } }, "configAnalysis": [ { "package": "algorithm.powerCurveAnalyst", "className": "PowerCurveAnalyst", "methodName": "executeAnalysis" }, { "package": "algorithm.powerScatter2DAnalyst", "className": "PowerScatter2DAnayst", "methodName": "executeAnalysis" }, { "package": "algorithm.powerScatterAnalyst", "className": "PowerScatterAnalyst", "methodName": "executeAnalysis" }, { "package": "algorithm.windSpeedFrequencyAnalyst", "className": "WindSpeedFrequencyAnalyst", "methodName": "executeAnalysis" }, { "package": "algorithm.generatorSpeedPowerAnalyst", "className": "GeneratorSpeedPowerAnalyst", "methodName": "executeAnalysis" } ], "graphSets": { "generatorSpeed": { "step": 200, "min": 1000, "max": 2000 }, "generatorTorque": { "step": 2000, "min": 0, "max": 12000 }, "cp": { "step": 0.5, "min": 0, "max": 2 }, "tsr": { "step": 5, "min": 0, "max": 30 }, "pitchAngle": { "step": 1, "min": -1, "max": 20 }, "activePower": { "step": 250, "min": 0, "max": 2000 } } } } ''' def load(): return LoadAnalysisInput(jsonString) def Analysis(contract: Contract): jsonString = "" try: contractDict = contract.to_dict() # 先转换为字典 jsonString = json.dumps(contractDict) except: traceback.print_exc() return jsonString