""" 农业投入Cd通量计算服务 @description: 根据Parameters表中的数据计算各个地区的农业投入输入Cd通量预测 @author: AcidMap Team @version: 1.0.0 """ import logging from typing import Dict, Any, List, Optional from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from ..database import SessionLocal, engine from ..models.parameters import Parameters class AgriculturalInputService: """ 农业投入Cd通量计算服务类 @description: 提供基于Parameters表数据的农业投入输入Cd通量预测计算功能 """ def __init__(self): """ 初始化农业投入服务 """ self.logger = logging.getLogger(__name__) def calculate_cd_flux_by_area(self, area: str) -> Dict[str, Any]: """ 根据地区计算农业投入输入Cd通量 @param area: 地区名称 @returns: 计算结果 """ try: with SessionLocal() as db: # 查询指定地区的参数 parameter = db.query(Parameters).filter(Parameters.area == area).first() if not parameter: return { "success": False, "message": f"未找到地区 '{area}' 的参数数据", "data": None } # 计算农业投入输入Cd通量 # 公式：F3*NF + F4*PF + F5*KF + F6*CF + F7*OF + F8*P + F9*FF + F10*AF cd_flux = ( parameter.f3 * parameter.nf + # 氮肥 parameter.f4 * parameter.pf + # 磷肥 parameter.f5 * parameter.kf + # 钾肥 parameter.f6 * parameter.cf + # 复合肥 parameter.f7 * parameter.of + # 有机肥 parameter.f8 * parameter.p + # 农药 parameter.f9 * parameter.ff + # 农家肥 parameter.f10 * parameter.af # 农膜 ) # 计算各项明细 details = { "nitrogen_fertilizer": parameter.f3 * parameter.nf, # 氮肥贡献 "phosphorus_fertilizer": parameter.f4 * parameter.pf, # 磷肥贡献 "potassium_fertilizer": parameter.f5 * parameter.kf, # 钾肥贡献 "compound_fertilizer": parameter.f6 * parameter.cf, # 复合肥贡献 "organic_fertilizer": parameter.f7 * parameter.of, # 有机肥贡献 "pesticide": parameter.f8 * parameter.p, # 农药贡献 "farmyard_manure": parameter.f9 * parameter.ff, # 农家肥贡献 "agricultural_film": parameter.f10 * parameter.af # 农膜贡献 } return { "success": True, "message": f"成功计算地区 '{area}' 的农业投入输入Cd通量", "data": { "area": area, "total_cd_flux": round(cd_flux, 6), "unit": "g/ha/a", "details": {key: round(value, 6) for key, value in details.items()}, "parameters_used": { "f3_nitrogen_cd_content": parameter.f3, "f4_phosphorus_cd_content": parameter.f4, "f5_potassium_cd_content": parameter.f5, "f6_compound_cd_content": parameter.f6, "f7_organic_cd_content": parameter.f7, "f8_pesticide_cd_content": parameter.f8, "f9_farmyard_cd_content": parameter.f9, "f10_film_cd_content": parameter.f10, "nf_nitrogen_usage": parameter.nf, "pf_phosphorus_usage": parameter.pf, "kf_potassium_usage": parameter.kf, "cf_compound_usage": parameter.cf, "of_organic_usage": parameter.of, "p_pesticide_usage": parameter.p, "ff_farmyard_usage": parameter.ff, "af_film_usage": parameter.af } } } except Exception as e: self.logger.error(f"计算地区 '{area}' 的Cd通量时发生错误: {str(e)}") return { "success": False, "message": f"计算失败: {str(e)}", "data": None } def calculate_all_areas_cd_flux(self) -> Dict[str, Any]: """ 计算所有地区的农业投入输入Cd通量 @returns: 所有地区的计算结果 """ try: with SessionLocal() as db: # 查询所有参数记录 parameters = db.query(Parameters).all() if not parameters: return { "success": False, "message": "数据库中未找到任何参数数据", "data": None } results = [] for param in parameters: # 计算每个地区的Cd通量 cd_flux = ( param.f3 * param.nf + param.f4 * param.pf + param.f5 * param.kf + param.f6 * param.cf + param.f7 * param.of + param.f8 * param.p + param.f9 * param.ff + param.f10 * param.af ) details = { "nitrogen_fertilizer": param.f3 * param.nf, "phosphorus_fertilizer": param.f4 * param.pf, "potassium_fertilizer": param.f5 * param.kf, "compound_fertilizer": param.f6 * param.cf, "organic_fertilizer": param.f7 * param.of, "pesticide": param.f8 * param.p, "farmyard_manure": param.f9 * param.ff, "agricultural_film": param.f10 * param.af } result_item = { "area": param.area, "total_cd_flux": round(cd_flux, 6), "details": {key: round(value, 6) for key, value in details.items()} } results.append(result_item) # 按总通量排序 results.sort(key=lambda x: x["total_cd_flux"], reverse=True) return { "success": True, "message": f"成功计算所有 {len(results)} 个地区的农业投入输入Cd通量", "data": { "total_areas": len(results), "unit": "g/ha/a", "results": results, "summary": { "max_cd_flux": max(results, key=lambda x: x["total_cd_flux"]) if results else None, "min_cd_flux": min(results, key=lambda x: x["total_cd_flux"]) if results else None, "average_cd_flux": round(sum(r["total_cd_flux"] for r in results) / len(results), 6) if results else 0 } } } except Exception as e: self.logger.error(f"计算所有地区Cd通量时发生错误: {str(e)}") return { "success": False, "message": f"计算失败: {str(e)}", "data": None } def get_available_areas(self) -> Dict[str, Any]: """ 获取数据库中所有可用的地区列表 @returns: 可用地区列表 """ try: with SessionLocal() as db: # 查询所有不重复的地区 areas = db.query(Parameters.area).distinct().all() area_list = [area[0] for area in areas if area[0]] return { "success": True, "message": f"成功获取 {len(area_list)} 个可用地区", "data": { "areas": sorted(area_list), "total_count": len(area_list) } } except Exception as e: self.logger.error(f"获取可用地区列表时发生错误: {str(e)}") return { "success": False, "message": f"获取失败: {str(e)}", "data": None } def calculate_cd_flux_with_custom_data(self, request) -> Dict[str, Any]: """ 使用用户提供的自定义数据计算农业投入输入Cd通量 @param request: 包含计算参数的请求对象 @returns: 计算结果 """ try: # 进行参数验证 if not self._validate_calculation_parameters(request): return { "success": False, "message": "提供的参数数据不完整或无效", "data": None } # 计算农业投入输入Cd通量 # 公式：F3*NF + F4*PF + F5*KF + F6*CF + F7*OF + F8*P + F9*FF + F10*AF cd_flux = ( request.f3_nitrogen_cd_content * request.nf_nitrogen_usage + # 氮肥 request.f4_phosphorus_cd_content * request.pf_phosphorus_usage + # 磷肥 request.f5_potassium_cd_content * request.kf_potassium_usage + # 钾肥 request.f6_compound_cd_content * request.cf_compound_usage + # 复合肥 request.f7_organic_cd_content * request.of_organic_usage + # 有机肥 request.f8_pesticide_cd_content * request.p_pesticide_usage + # 农药 request.f9_farmyard_cd_content * request.ff_farmyard_usage + # 农家肥 request.f10_film_cd_content * request.af_film_usage # 农膜 ) # 计算各项明细贡献 details = { "nitrogen_fertilizer": request.f3_nitrogen_cd_content * request.nf_nitrogen_usage, "phosphorus_fertilizer": request.f4_phosphorus_cd_content * request.pf_phosphorus_usage, "potassium_fertilizer": request.f5_potassium_cd_content * request.kf_potassium_usage, "compound_fertilizer": request.f6_compound_cd_content * request.cf_compound_usage, "organic_fertilizer": request.f7_organic_cd_content * request.of_organic_usage, "pesticide": request.f8_pesticide_cd_content * request.p_pesticide_usage, "farmyard_manure": request.f9_farmyard_cd_content * request.ff_farmyard_usage, "agricultural_film": request.f10_film_cd_content * request.af_film_usage } # 构建返回结果 result_data = { "total_cd_flux": round(cd_flux, 6), "unit": "g/ha/a", "details": {key: round(value, 6) for key, value in details.items()}, "input_parameters": { "cd_contents": { "f3_nitrogen_cd_content": request.f3_nitrogen_cd_content, "f4_phosphorus_cd_content": request.f4_phosphorus_cd_content, "f5_potassium_cd_content": request.f5_potassium_cd_content, "f6_compound_cd_content": request.f6_compound_cd_content, "f7_organic_cd_content": request.f7_organic_cd_content, "f8_pesticide_cd_content": request.f8_pesticide_cd_content, "f9_farmyard_cd_content": request.f9_farmyard_cd_content, "f10_film_cd_content": request.f10_film_cd_content }, "usage_amounts": { "nf_nitrogen_usage": request.nf_nitrogen_usage, "pf_phosphorus_usage": request.pf_phosphorus_usage, "kf_potassium_usage": request.kf_potassium_usage, "cf_compound_usage": request.cf_compound_usage, "of_organic_usage": request.of_organic_usage, "p_pesticide_usage": request.p_pesticide_usage, "ff_farmyard_usage": request.ff_farmyard_usage, "af_film_usage": request.af_film_usage } } } # 如果有描述信息，添加到结果中 if request.description: result_data["description"] = request.description return { "success": True, "message": "成功计算农业投入输入Cd通量", "data": result_data } except Exception as e: self.logger.error(f"使用自定义数据计算Cd通量时发生错误: {str(e)}") return { "success": False, "message": f"计算失败: {str(e)}", "data": None } def _validate_calculation_parameters(self, request) -> bool: """ 验证计算参数的有效性 @param request: 请求对象 @returns: 验证是否通过 """ try: # 检查所有必需的参数是否存在且为非负数 required_attrs = [ 'f3_nitrogen_cd_content', 'f4_phosphorus_cd_content', 'f5_potassium_cd_content', 'f6_compound_cd_content', 'f7_organic_cd_content', 'f8_pesticide_cd_content', 'f9_farmyard_cd_content', 'f10_film_cd_content', 'nf_nitrogen_usage', 'pf_phosphorus_usage', 'kf_potassium_usage', 'cf_compound_usage', 'of_organic_usage', 'p_pesticide_usage', 'ff_farmyard_usage', 'af_film_usage' ] for attr in required_attrs: if not hasattr(request, attr): self.logger.warning(f"缺少必需参数: {attr}") return False value = getattr(request, attr) if value is None or value < 0: self.logger.warning(f"参数 {attr} 无效: {value}") return False return True except Exception as e: self.logger.error(f"验证参数时发生错误: {str(e)}") return False