|
@@ -16,8 +16,10 @@ from app.utils.mapping_utils import MappingUtils, csv_to_raster_workflow
|
|
|
|
|
|
# 配置日志
|
|
# 配置日志
|
|
from app.log.logger import get_logger
|
|
from app.log.logger import get_logger
|
|
|
|
+
|
|
logger = get_logger(__name__)
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
|
|
+
|
|
def get_base_dir():
|
|
def get_base_dir():
|
|
"""获取基础目录路径(与土地数据处理函数一致)"""
|
|
"""获取基础目录路径(与土地数据处理函数一致)"""
|
|
if getattr(sys, 'frozen', False):
|
|
if getattr(sys, 'frozen', False):
|
|
@@ -48,6 +50,7 @@ def get_default_boundary_shp():
|
|
|
|
|
|
return None
|
|
return None
|
|
|
|
|
|
|
|
+
|
|
class FluxCdVisualizationService:
|
|
class FluxCdVisualizationService:
|
|
"""
|
|
"""
|
|
农田Cd通量可视化服务类
|
|
农田Cd通量可视化服务类
|
|
@@ -70,24 +73,81 @@ class FluxCdVisualizationService:
|
|
"""
|
|
"""
|
|
生成输入镉通量(In_Cd)的空间分布图和直方图
|
|
生成输入镉通量(In_Cd)的空间分布图和直方图
|
|
|
|
|
|
|
|
+ @param output_dir: 输出文件目录
|
|
|
|
+ @param boundary_shp: 边界Shapefile文件路径
|
|
|
|
+ @return: 包含输出文件路径的字典
|
|
|
|
+ """
|
|
|
|
+ return self._generate_cd_flux_map(field='in_cd', title_prefix="input",
|
|
|
|
+ output_dir=output_dir, boundary_shp=boundary_shp)
|
|
|
|
+
|
|
|
|
+ def generate_cd_output_flux_map(self, output_dir: str = None, boundary_shp: str = None) -> Dict[str, Any]:
|
|
|
|
+ """
|
|
|
|
+ 生成输出镉通量(Out_Cd)的空间分布图和直方图
|
|
|
|
+
|
|
|
|
+ @param output_dir: 输出文件目录
|
|
|
|
+ @param boundary_shp: 边界Shapefile文件路径
|
|
|
|
+ @return: 包含输出文件路径的字典
|
|
|
|
+ """
|
|
|
|
+ return self._generate_cd_flux_map(field='out_cd', title_prefix="output",
|
|
|
|
+ output_dir=output_dir, boundary_shp=boundary_shp)
|
|
|
|
+
|
|
|
|
+ def generate_cd_net_flux_map(self, output_dir: str = None, boundary_shp: str = None) -> Dict[str, Any]:
|
|
|
|
+ """
|
|
|
|
+ 生成净镉通量(Net_Cd)的空间分布图和直方图
|
|
|
|
+
|
|
|
|
+ @param output_dir: 输出文件目录
|
|
|
|
+ @param boundary_shp: 边界Shapefile文件路径
|
|
|
|
+ @return: 包含输出文件路径的字典
|
|
|
|
+ """
|
|
|
|
+ return self._generate_cd_flux_map(field='net_cd', title_prefix="net",
|
|
|
|
+ output_dir=output_dir, boundary_shp=boundary_shp)
|
|
|
|
+
|
|
|
|
+ def _generate_cd_flux_map(self, field: str, title_prefix: str,
|
|
|
|
+ output_dir: str = None, boundary_shp: str = None) -> Dict[str, Any]:
|
|
|
|
+ """
|
|
|
|
+ 通用的Cd通量地图生成方法
|
|
|
|
+
|
|
|
|
+ @param field: 要可视化的字段('in_cd'、'out_cd'或'net_cd')
|
|
|
|
+ @param title_prefix: 标题前缀('input'、'output'或'net')
|
|
@param output_dir: 输出文件目录
|
|
@param output_dir: 输出文件目录
|
|
@param boundary_shp: 边界Shapefile文件路径
|
|
@param boundary_shp: 边界Shapefile文件路径
|
|
@return: 包含输出文件路径的字典
|
|
@return: 包含输出文件路径的字典
|
|
"""
|
|
"""
|
|
try:
|
|
try:
|
|
|
|
+ # 字段到标题的映射
|
|
|
|
+ field_titles = {
|
|
|
|
+ 'in_cd': "Input",
|
|
|
|
+ 'out_cd': "Output",
|
|
|
|
+ 'net_cd': "Net"
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ # 字段到单位的映射
|
|
|
|
+ field_units = {
|
|
|
|
+ 'in_cd': "g/ha/a",
|
|
|
|
+ 'out_cd': "g/ha/a",
|
|
|
|
+ 'net_cd': "g/ha/a"
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ # 字段到轴标签的映射
|
|
|
|
+ field_labels = {
|
|
|
|
+ 'in_cd': 'input Cd flux',
|
|
|
|
+ 'out_cd': 'output Cd flux',
|
|
|
|
+ 'net_cd': 'net Cd flux'
|
|
|
|
+ }
|
|
|
|
+
|
|
# 如果未提供数据库会话,则创建新的会话
|
|
# 如果未提供数据库会话,则创建新的会话
|
|
db = self.db if self.db else SessionLocal()
|
|
db = self.db if self.db else SessionLocal()
|
|
should_close = self.db is None
|
|
should_close = self.db is None
|
|
|
|
|
|
# 1. 从数据库查询数据
|
|
# 1. 从数据库查询数据
|
|
- data = self._fetch_fluxcd_data(db)
|
|
|
|
|
|
+ data = self._fetch_fluxcd_data(db, field)
|
|
|
|
|
|
if not data:
|
|
if not data:
|
|
if should_close:
|
|
if should_close:
|
|
db.close()
|
|
db.close()
|
|
return {
|
|
return {
|
|
"success": False,
|
|
"success": False,
|
|
- "message": "数据库中未找到任何农田Cd通量数据",
|
|
|
|
|
|
+ "message": f"数据库中未找到任何农田{field_titles[field]}Cd通量数据",
|
|
"data": None
|
|
"data": None
|
|
}
|
|
}
|
|
|
|
|
|
@@ -106,8 +166,9 @@ class FluxCdVisualizationService:
|
|
self.logger.warning("未找到默认边界文件,将不使用边界裁剪")
|
|
self.logger.warning("未找到默认边界文件,将不使用边界裁剪")
|
|
|
|
|
|
# 4. 生成CSV文件
|
|
# 4. 生成CSV文件
|
|
- csv_path = os.path.join(output_dir, "fluxcd_input.csv")
|
|
|
|
- self._generate_csv(data, csv_path)
|
|
|
|
|
|
+ base_name = f"fluxcd_{title_prefix}"
|
|
|
|
+ csv_path = os.path.join(output_dir, f"{base_name}.csv")
|
|
|
|
+ self._generate_csv(data, csv_path, field_names=['lon', 'lan', field])
|
|
|
|
|
|
# 5. 设置模板TIFF路径(使用土地数据处理中的模板)
|
|
# 5. 设置模板TIFF路径(使用土地数据处理中的模板)
|
|
template_tif = os.path.join(static_dir, "meanTemp.tif")
|
|
template_tif = os.path.join(static_dir, "meanTemp.tif")
|
|
@@ -118,7 +179,6 @@ class FluxCdVisualizationService:
|
|
raise FileNotFoundError(f"未找到模板TIFF文件")
|
|
raise FileNotFoundError(f"未找到模板TIFF文件")
|
|
|
|
|
|
# 6. 使用csv_to_raster_workflow将CSV转换为栅格
|
|
# 6. 使用csv_to_raster_workflow将CSV转换为栅格
|
|
- base_name = "fluxcd_input"
|
|
|
|
raster_path = os.path.join(output_dir, f"{base_name}_raster.tif")
|
|
raster_path = os.path.join(output_dir, f"{base_name}_raster.tif")
|
|
|
|
|
|
# 关键修改:确保传递边界文件
|
|
# 关键修改:确保传递边界文件
|
|
@@ -128,7 +188,7 @@ class FluxCdVisualizationService:
|
|
output_dir=output_dir,
|
|
output_dir=output_dir,
|
|
resolution_factor=4.0,
|
|
resolution_factor=4.0,
|
|
interpolation_method='linear',
|
|
interpolation_method='linear',
|
|
- field_name='In_Cd',
|
|
|
|
|
|
+ field_name=field,
|
|
lon_col=0, # CSV中经度列索引
|
|
lon_col=0, # CSV中经度列索引
|
|
lat_col=1, # CSV中纬度列索引
|
|
lat_col=1, # CSV中纬度列索引
|
|
value_col=2, # CSV中数值列索引
|
|
value_col=2, # CSV中数值列索引
|
|
@@ -146,7 +206,7 @@ class FluxCdVisualizationService:
|
|
tif_path=raster_path, # 栅格文件
|
|
tif_path=raster_path, # 栅格文件
|
|
output_path=map_path,
|
|
output_path=map_path,
|
|
colormap='green_yellow_red_purple',
|
|
colormap='green_yellow_red_purple',
|
|
- title="input Cd flux map",
|
|
|
|
|
|
+ title=f"{title_prefix} Cd flux map",
|
|
output_size=12,
|
|
output_size=12,
|
|
dpi=300,
|
|
dpi=300,
|
|
enable_interpolation=False,
|
|
enable_interpolation=False,
|
|
@@ -157,15 +217,15 @@ class FluxCdVisualizationService:
|
|
histogram_path = self.mapper.create_histogram(
|
|
histogram_path = self.mapper.create_histogram(
|
|
raster_path,
|
|
raster_path,
|
|
save_path=os.path.join(output_dir, f"{base_name}_histogram.jpg"),
|
|
save_path=os.path.join(output_dir, f"{base_name}_histogram.jpg"),
|
|
- xlabel='input Cd flux(g/ha/a)',
|
|
|
|
|
|
+ xlabel=f'{field_labels[field]}({field_units[field]})',
|
|
ylabel='frequency',
|
|
ylabel='frequency',
|
|
- title='input Cd flux histogram',
|
|
|
|
|
|
+ title=f'{title_prefix} Cd flux histogram',
|
|
bins=100
|
|
bins=100
|
|
)
|
|
)
|
|
|
|
|
|
result = {
|
|
result = {
|
|
"success": True,
|
|
"success": True,
|
|
- "message": "成功生成Cd通量可视化结果",
|
|
|
|
|
|
+ "message": f"成功生成{field_titles[field]}Cd通量可视化结果",
|
|
"data": {
|
|
"data": {
|
|
"csv": csv_path,
|
|
"csv": csv_path,
|
|
"raster": raster_path,
|
|
"raster": raster_path,
|
|
@@ -194,11 +254,12 @@ class FluxCdVisualizationService:
|
|
update_service = FluxCdUpdateService(db=self.db)
|
|
update_service = FluxCdUpdateService(db=self.db)
|
|
return update_service.update_from_csv(csv_file_path)
|
|
return update_service.update_from_csv(csv_file_path)
|
|
|
|
|
|
- def _fetch_fluxcd_data(self, db: Session) -> list:
|
|
|
|
|
|
+ def _fetch_fluxcd_data(self, db: Session, field: str = 'in_cd') -> list:
|
|
"""
|
|
"""
|
|
从数据库查询需要的数据
|
|
从数据库查询需要的数据
|
|
|
|
|
|
@param db: 数据库会话
|
|
@param db: 数据库会话
|
|
|
|
+ @param field: 要查询的字段('in_cd', 'out_cd'或'net_cd')
|
|
@returns: 查询结果列表
|
|
@returns: 查询结果列表
|
|
"""
|
|
"""
|
|
try:
|
|
try:
|
|
@@ -206,7 +267,7 @@ class FluxCdVisualizationService:
|
|
query = db.query(
|
|
query = db.query(
|
|
FarmlandData.lon,
|
|
FarmlandData.lon,
|
|
FarmlandData.lan,
|
|
FarmlandData.lan,
|
|
- FluxCdOutputData.in_cd
|
|
|
|
|
|
+ getattr(FluxCdOutputData, field)
|
|
).join(
|
|
).join(
|
|
FluxCdOutputData,
|
|
FluxCdOutputData,
|
|
(FarmlandData.farmland_id == FluxCdOutputData.farmland_id) &
|
|
(FarmlandData.farmland_id == FluxCdOutputData.farmland_id) &
|
|
@@ -219,18 +280,22 @@ class FluxCdVisualizationService:
|
|
self.logger.error(f"查询农田Cd通量数据时发生错误: {str(e)}")
|
|
self.logger.error(f"查询农田Cd通量数据时发生错误: {str(e)}")
|
|
return []
|
|
return []
|
|
|
|
|
|
- def _generate_csv(self, data: list, output_path: str):
|
|
|
|
|
|
+ def _generate_csv(self, data: list, output_path: str, field_names: list = None):
|
|
"""
|
|
"""
|
|
将查询结果生成CSV文件
|
|
将查询结果生成CSV文件
|
|
|
|
|
|
@param data: 查询结果列表
|
|
@param data: 查询结果列表
|
|
@param output_path: 输出CSV文件路径
|
|
@param output_path: 输出CSV文件路径
|
|
|
|
+ @param field_names: CSV列名列表
|
|
"""
|
|
"""
|
|
|
|
+ if field_names is None:
|
|
|
|
+ field_names = ['lon', 'lan', 'In_Cd']
|
|
|
|
+
|
|
try:
|
|
try:
|
|
with open(output_path, 'w', newline='', encoding='utf-8') as csvfile:
|
|
with open(output_path, 'w', newline='', encoding='utf-8') as csvfile:
|
|
writer = csv.writer(csvfile)
|
|
writer = csv.writer(csvfile)
|
|
# 写入表头
|
|
# 写入表头
|
|
- writer.writerow(['lon', 'lan', 'In_Cd'])
|
|
|
|
|
|
+ writer.writerow(field_names)
|
|
|
|
|
|
# 写入数据
|
|
# 写入数据
|
|
for row in data:
|
|
for row in data:
|
|
@@ -343,10 +408,10 @@ class FluxCdUpdateService:
|
|
agro_chemicals_input=row['NCP_Cd'],
|
|
agro_chemicals_input=row['NCP_Cd'],
|
|
# 设置合理的默认值
|
|
# 设置合理的默认值
|
|
initial_cd=0.0,
|
|
initial_cd=0.0,
|
|
- groundwater_leaching=0.023,
|
|
|
|
- surface_runoff=0.368,
|
|
|
|
- grain_removal=0.0,
|
|
|
|
- straw_removal=0.0
|
|
|
|
|
|
+ groundwater_leaching=row.get('DX_Cd', 0.023), # 添加默认值处理
|
|
|
|
+ surface_runoff=row.get('DB_Cd', 0.368),
|
|
|
|
+ grain_removal=row.get('ZL_Cd', 0.0),
|
|
|
|
+ straw_removal=row.get('JG_Cd', 0.0)
|
|
)
|
|
)
|
|
self.db.add(input_data)
|
|
self.db.add(input_data)
|
|
self.logger.info("输入通量记录创建成功")
|
|
self.logger.info("输入通量记录创建成功")
|
|
@@ -354,8 +419,11 @@ class FluxCdUpdateService:
|
|
# 计算输入总通量
|
|
# 计算输入总通量
|
|
in_cd = row['DQCJ_Cd'] + row['GGS_Cd'] + row['NCP_Cd']
|
|
in_cd = row['DQCJ_Cd'] + row['GGS_Cd'] + row['NCP_Cd']
|
|
|
|
|
|
- # 计算输出总通量(假设默认值)
|
|
|
|
- out_cd = 0.023 + 0.368 + 0.0 + 0.0
|
|
|
|
|
|
+ # 计算输出总通量
|
|
|
|
+ out_cd = (input_data.groundwater_leaching +
|
|
|
|
+ input_data.surface_runoff +
|
|
|
|
+ input_data.grain_removal +
|
|
|
|
+ input_data.straw_removal)
|
|
|
|
|
|
# 创建新的输出通量记录
|
|
# 创建新的输出通量记录
|
|
output_data = FluxCdOutputData(
|
|
output_data = FluxCdOutputData(
|
|
@@ -388,10 +456,18 @@ class FluxCdUpdateService:
|
|
def _validate_csv(self, df: pd.DataFrame):
|
|
def _validate_csv(self, df: pd.DataFrame):
|
|
"""验证CSV文件包含必要的列"""
|
|
"""验证CSV文件包含必要的列"""
|
|
required_columns = {'lon', 'lan', 'DQCJ_Cd', 'GGS_Cd', 'NCP_Cd'}
|
|
required_columns = {'lon', 'lan', 'DQCJ_Cd', 'GGS_Cd', 'NCP_Cd'}
|
|
|
|
+ # 添加输出通量可选字段
|
|
|
|
+ output_fields = {'DX_Cd', 'DB_Cd', 'ZL_Cd', 'JG_Cd'}
|
|
|
|
+
|
|
if not required_columns.issubset(df.columns):
|
|
if not required_columns.issubset(df.columns):
|
|
missing = required_columns - set(df.columns)
|
|
missing = required_columns - set(df.columns)
|
|
raise ValueError(f"CSV缺少必要列: {', '.join(missing)}")
|
|
raise ValueError(f"CSV缺少必要列: {', '.join(missing)}")
|
|
|
|
|
|
|
|
+ # 检查是否包含输出通量字段(可选)
|
|
|
|
+ if not output_fields.issubset(df.columns):
|
|
|
|
+ missing_output = output_fields - set(df.columns)
|
|
|
|
+ self.logger.warning(f"CSV缺少输出通量字段: {', '.join(missing_output)},将使用默认值")
|
|
|
|
+
|
|
def _find_farmland(self, lon: float, lan: float) -> FarmlandData:
|
|
def _find_farmland(self, lon: float, lan: float) -> FarmlandData:
|
|
"""根据经纬度查找农田样点"""
|
|
"""根据经纬度查找农田样点"""
|
|
# 使用容差匹配(0.001度≈100米)
|
|
# 使用容差匹配(0.001度≈100米)
|
|
@@ -412,7 +488,7 @@ class FluxCdUpdateService:
|
|
self.logger.warning(f"未找到输入数据: Farmland_ID={farmland.farmland_id}, Sample_ID={farmland.sample_id}")
|
|
self.logger.warning(f"未找到输入数据: Farmland_ID={farmland.farmland_id}, Sample_ID={farmland.sample_id}")
|
|
return False
|
|
return False
|
|
|
|
|
|
- # 检查是否需要更新
|
|
|
|
|
|
+ # 检查是否需要更新
|
|
updated = False
|
|
updated = False
|
|
if input_data.atmospheric_deposition != row['DQCJ_Cd']:
|
|
if input_data.atmospheric_deposition != row['DQCJ_Cd']:
|
|
input_data.atmospheric_deposition = row['DQCJ_Cd']
|
|
input_data.atmospheric_deposition = row['DQCJ_Cd']
|
|
@@ -446,11 +522,16 @@ class FluxCdUpdateService:
|
|
FluxCdInputData.sample_id == farmland.sample_id
|
|
FluxCdInputData.sample_id == farmland.sample_id
|
|
).first()
|
|
).first()
|
|
|
|
|
|
- # 重新计算并更新
|
|
|
|
|
|
+ # 计算输出总通量
|
|
|
|
+ out_cd = (input_data.groundwater_leaching + # DX_Cd
|
|
|
|
+ input_data.surface_runoff + # DB_Cd
|
|
|
|
+ input_data.grain_removal + # ZL_Cd
|
|
|
|
+ input_data.straw_removal) # JG_Cd
|
|
|
|
+
|
|
|
|
+ # 更新输出通量记录
|
|
output_data.in_cd = input_data.input_flux()
|
|
output_data.in_cd = input_data.input_flux()
|
|
- # 注意:输出总通量out_cd不会由用户上传的CSV更新,所以我们保持原值
|
|
|
|
- # 重新计算净通量
|
|
|
|
- output_data.net_cd = output_data.in_cd - output_data.out_cd
|
|
|
|
|
|
+ output_data.out_cd = out_cd # 添加输出通量计算
|
|
|
|
+ output_data.net_cd = output_data.in_cd - out_cd # 更新净通量计算
|
|
|
|
|
|
self.logger.info(f"更新输出通量: Farmland_ID={farmland.farmland_id}, Sample_ID={farmland.sample_id}")
|
|
self.logger.info(f"更新输出通量: Farmland_ID={farmland.farmland_id}, Sample_ID={farmland.sample_id}")
|
|
|
|
|
|
@@ -475,6 +556,16 @@ if __name__ == "__main__":
|
|
|
|
|
|
# 测试生成可视化结果
|
|
# 测试生成可视化结果
|
|
print("\n>>> 测试生成Cd输入通量可视化地图")
|
|
print("\n>>> 测试生成Cd输入通量可视化地图")
|
|
|
|
+ input_result = service.generate_cd_input_flux_map()
|
|
|
|
+ print(json.dumps(input_result, indent=2, ensure_ascii=False))
|
|
|
|
+
|
|
|
|
+ print("\n>>> 测试生成Cd输出通量可视化地图")
|
|
|
|
+ output_result = service.generate_cd_output_flux_map()
|
|
|
|
+ print(json.dumps(output_result, indent=2, ensure_ascii=False))
|
|
|
|
+
|
|
|
|
+ print("\n>>> 测试生成Cd净通量可视化地图")
|
|
|
|
+ net_result = service.generate_cd_net_flux_map()
|
|
|
|
+ print(json.dumps(net_result, indent=2, ensure_ascii=False))
|
|
|
|
|
|
# 测试更新服务
|
|
# 测试更新服务
|
|
print("\n>>> 测试从CSV更新Cd通量数据")
|
|
print("\n>>> 测试从CSV更新Cd通量数据")
|
|
@@ -485,9 +576,9 @@ if __name__ == "__main__":
|
|
|
|
|
|
# 创建测试CSV文件
|
|
# 创建测试CSV文件
|
|
test_data = [
|
|
test_data = [
|
|
- "lon,lan,DQCJ_Cd,GGS_Cd,NCP_Cd",
|
|
|
|
- "113.123,25.456,1.24,4.56,7.89",
|
|
|
|
- "113.125,25.457,2.35,5.67,8.90"
|
|
|
|
|
|
+ "lon,lan,DQCJ_Cd,GGS_Cd,NCP_Cd,DX_Cd,DB_Cd,ZL_Cd,JG_Cd",
|
|
|
|
+ "113.123,25.456,1.24,4.56,7.89,0.1,0.2,0.05,0.03",
|
|
|
|
+ "113.125,25.457,2.35,5.67,8.90,0.15,0.25,0.06,0.04"
|
|
]
|
|
]
|
|
|
|
|
|
with open(test_csv_path, 'w', encoding='utf-8') as f:
|
|
with open(test_csv_path, 'w', encoding='utf-8') as f:
|