手机全程托管种地系统
README.md
#
项目简介
基于Python开发的智慧农业全流程自动化托管平台，集成物联网感知、AI决策、自动化执行等技术，实现"一键托管，全程无忧"的现代农业管理模式。
##
🚀
项目结构
smartfarm/
├──
main.py
使用说明
详见下方"使用说明"章节
📋
使用说明
场景描述
痛点分析：
1.
劳动力短缺：农村青壮年外出务工，留守老人难以承担高强度农事操作
2.
技术门槛高：智慧农业设备操作复杂，农民缺乏专业知识
3.
管理粗放：传统种植凭经验，水肥药投入不精准，造成浪费和污染
4.
监控滞后：无法实时掌握田间状况，错过最佳农时
5.
决策困难：面对突发天气、病虫害等情况，缺乏科学应对方案
解决方案：
本系统通过"一键托管"模式，将复杂的农业技术封装为简单操作，实现：
-
手机端一键启动，系统自动完成从整地到收获的全流程
-
AI根据土壤、气候、作物品种自动制定最优种植方案
-
每日推送详细进度报告，包含环境数据、操作记录、生长状态
核心逻辑流程图
用户手机APP
发送"一键托管"指令
任务管理器接收
↓
计划生成器分析(土壤+气候+作物)
生成7阶段种植计划
↓
任务执行引擎按日分解任务
调度各设备控制器
↓
传感器实时采集数据
核心代码实现
1.
(config.py)
"""
智能农机装备前沿技术
系统配置文件
包含设备连接、作物参数、系统设置等配置项
"""
from
dataclasses
os
@dataclass
class
DeviceConfig:
"""智能农机设备配置类"""
#
灌溉系统
irrigation_controller_ip:
str
"192.168.1.100"
irrigation_port:
int
field(default_factory=lambda:
[1,
"192.168.1.101"
fertilizer_tank_capacity:
float
field(default_factory=lambda:
0.3,
氮肥比例
"phosphorus":
0.2,
磷肥比例
"potassium":
0.3,
"192.168.1.102"
tank_volume:
float
环境传感器
sensor_mqtt_broker:
str
"192.168.1.50"
sensor_mqtt_port:
int
"farm/sensors/soil/+"
weather_topic:
str
"farm/sensors/weather"
@dataclass
class
CropConfig:
"""作物种植参数配置"""
name:
str
生育期配置（天）
growth_stages:
Dict[str,
field(default_factory=lambda:
15,
分蘖拔节期
"jointing":
25,
灌浆结实期
"maturity":
环境参数阈值
optimal_temp_range:
tuple
适宜温度范围℃
optimal_humidity_range:
tuple
适宜湿度范围%
optimal_soil_moisture:
tuple
适宜土壤含水量%
optimal_ph_range:
tuple
水肥需求参数
daily_water_requirement:
Dict[str,
field(default_factory=lambda:
15.0,
25.0,
"jointing":
30.0,
"maturity":
20.0
})
nitrogen_application:
Dict[str,
field(default_factory=lambda:
150.0,
kg/ha
"top_dressing_1":
80.0,
第一次追肥
"top_dressing_2":
60.0
第二次追肥
})
@dataclass
class
SystemConfig:
"""系统运行配置"""
app_name:
str
"智慧农业全程托管系统"
version:
str
每日报告生成时间
data_sync_interval:
int
"data/smartfarm.db"
log_path:
str
"logs/system.log"
#
Dict[str,
field(default_factory=lambda:
{
"soil_moisture_low":
50.0,
土壤含水量过低报警值%
"soil_moisture_high":
85.0,
土壤含水量过高报警值%
"temperature_high":
35.0,
高温报警值℃
"temperature_low":
5.0,
低温报警值℃
"pest_detection_threshold":
0.3
DeviceConfig()
crop_config
=
CropConfig()
system_config
=
[system_config.db_path.rsplit('/',
1)[0],
system_config.log_path.rsplit('/',
1)[0]]:
if
os.path.exists(path):
os.makedirs(path)
2.
数据模型
(models/task.py)
"""
任务数据模型模块
定义托管任务、种植计划、执行记录等核心数据结构
"""
from
dataclasses
TaskStatus(Enum):
"""任务状态枚举"""
PENDING
=
OperationType(Enum):
"""农事操作类型枚举"""
LAND_PREPARATION
=
监测
@dataclass
class
OperationRecord:
"""
农事操作记录类
记录单次操作的详细信息
"""
operation_id:
str
field(default_factory=lambda:
str(uuid.uuid4()))
operation_type:
OperationType
field(default_factory=dict)
result:
Dict[str,
field(default_factory=dict)
notes:
str
field(default_factory=list)
@dataclass
class
DailyProgress:
"""
每日进度类
记录单日种植进度和关键指标
"""
date:
datetime
field(default_factory=datetime.now)
stage:
str
当前生育期
operations_completed:
=
field(default_factory=list)
operations_pending:
=
field(default_factory=list)
sensor_data:
Dict[str,
field(default_factory=dict)
growth_metrics:
Dict[str,
field(default_factory=dict)
alerts:
List[str]
field(default_factory=list)
summary:
str
""
@dataclass
class
PlantingPlan:
"""
种植计划类
包含完整的作物种植周期计划
"""
plan_id:
str
field(default_factory=lambda:
str(uuid.uuid4()))
crop_config:
'CropConfig'
None
total_duration_days:
int
field(default_factory=list)
daily_tasks:
Dict[int,
field(default_factory=dict)
created_at:
datetime
field(default_factory=datetime.now)
updated_at:
datetime
field(default_factory=datetime.now)
@dataclass
class
ManagedTask:
"""
托管任务类
核心业务对象
表示一个完整的"一键托管"种植任务
"""
task_id:
str
field(default_factory=lambda:
str(uuid.uuid4()))
user_id:
str
""
farm_location:
Dict[str,
field(default_factory=dict)
lat,
"preparation"
current_day:
int
TaskStatus.PENDING
progress_history:
List[DailyProgress]
field(default_factory=list)
created_at:
datetime
field(default_factory=datetime.now)
started_at:
Optional[datetime]
float:
"""计算总体进度百分比"""
if
self.status
TaskStatus.COMPLETED:
return
=
sum(
stage["duration"]
for
stage
self.planting_plan.stages
)
self.planting_plan
List[OperationRecord]:
"""获取当日待执行操作"""
return
self.planting_plan.daily_tasks.get(self.current_day,
[])
def
Any]:
"""转换为字典格式，用于API返回"""
return
self.task_id,
"status":
self.status.value,
"current_stage":
self.current_stage,
"current_day":
self.current_day,
"progress":
f"{self.get_progress_percentage():.1f}%",
"total_stages":
self.planting_plan
self.created_at.isoformat(),
"started_at":
self.started_at
None,
"completed_at":
self.completed_at
(core/plan_generator.py)
"""
种植计划生成器模块
基于智能农机装备前沿技术，结合土壤、气候、作物品种等因素
自动生成个性化种植方案
"""
import
random
from
OperationRecord,
OperationType,
TaskStatus
)
@dataclass
class
EnvironmentalData:
"""环境数据类，存储土壤和气象数据"""
soil_moisture:
float
field(default_factory=list)
7天温度预报
class
PlanGenerator:
"""
智能种植计划生成器
核心技术点：
1.
多因子决策分析
针对极端天气的应对策略
"""
def
__init__(self,
device_config):
self.crop_config
=
crop_config
self.device_config
=
device_config
self.env_data
=
EnvironmentalData:
"""
分析环境传感器数据
Args:
sensor_data:
来自物联网传感器的原始数据
Returns:
EnvironmentalData:
处理后的环境数据对象
"""
self.env_data.soil_moisture
=
sensor_data.get("soil_moisture",
60.0)
self.env_data.soil_temperature
=
sensor_data.get("soil_temperature",
20.0)
self.env_data.soil_ph
=
sensor_data.get("soil_ph",
6.5)
self.env_data.soil_nitrogen
=
sensor_data.get("nitrogen",
100.0)
self.env_data.soil_phosphorus
=
sensor_data.get("phosphorus",
30.0)
self.env_data.soil_potassium
=
sensor_data.get("potassium",
80.0)
self.env_data.air_temperature
=
sensor_data.get("air_temp",
25.0)
self.env_data.air_humidity
=
sensor_data.get("air_humidity",
70.0)
self.env_data.rainfall_24h
=
sensor_data.get("rainfall_24h",
0.0)
return
calculate_stage_adjustments(self)
->
int]:
"""
根据环境数据计算各生育期调整系数
算法说明：
-
温度影响：每偏离最适温度1℃，调整5%时间
-
水分影响：土壤含水量每低于最适10%，苗期延长2天
-
pH影响：pH每偏离最适0.5，延长适应期3天
"""
adjustments
=
self.crop_config.optimal_temp_range
avg_temp
=
self.env_data.air_temperature
if
avg_temp
opt_max
else:
temp_factor
=
self.crop_config.optimal_soil_moisture
moist
=
self.env_data.soil_moisture
if
moist
opt_moist_min:
moisture_deficit
=
opt_moist_min
else:
moisture_deficit
=
6.75为最适中点
else:
ph_factor
=
self.crop_config.growth_stages.copy()
for
stage,
base_stages.items():
adjustment
=
"tillering"]:
adjustment
+=
"jointing":
adjustment
+=
generate_irrigation_schedule(self)
->
Any]]:
"""
生成智能灌溉计划
采用ET0（参考作物蒸散量）方法计算需水量
结合土壤含水量、降雨预测进行动态调整
"""
schedule
=
self.crop_config.growth_stages
water_req
=
self.crop_config.daily_water_requirement
day_counter
=
stages.items():
daily_water
=
仅添加有意义的灌溉任务
schedule.append({
"day":
day_counter,
"operation":
"irrigation",
"water_amount":
1),
"duration_minutes":
2),
2分钟/毫米估算
"priority":
"high"
"normal"
})
day_counter
+=
generate_fertilization_schedule(self)
->
Any]]:
"""
生成精准施肥计划
基于土壤养分检测和作物需求，计算差异化施肥方案
"""
schedule
=
self.crop_config.growth_stages
#
=
self.env_data.soil_nitrogen
p_level
=
self.env_data.soil_phosphorus
k_level
=
self.env_data.soil_potassium
#
=
假设标准150mg/kg
p_deficiency
=
self.crop_config.nitrogen_application["base"]
#
=
0.5)
schedule.append({
"day":
day_counter,
"operation":
"fertilization",
"type":
"base_fertilizer",
"amounts":
round(actual_n,
1),
"phosphorus":
round(actual_p,
1)
},
"method":
"deep_placement",
"priority":
"critical"
})
elif
stage
第一次追肥（攻蘖肥）
top_n
=
self.crop_config.nitrogen_application["top_dressing_1"]
actual_n
=
0.3)
schedule.append({
"day":
day_counter,
"operation":
"fertilization",
"type":
"top_dressing_1",
"amounts":
{"nitrogen":
"side_dressing",
"priority":
"high"
})
elif
stage
第二次追肥（攻穗肥）
top_n
=
self.crop_config.nitrogen_application["top_dressing_2"]
actual_n
=
0.3)
schedule.append({
"day":
day_counter,
"operation":
"fertilization",
"type":
"top_dressing_2",
"amounts":
{"nitrogen":
"root_irrigation",
"priority":
"critical"
})
day_counter
+=
generate_pest_control_schedule(self)
->
Any]]:
"""
生成植保作业计划
基于作物生育期和气象条件预测病虫害发生风险
"""
schedule
=
self.crop_config.growth_stages
#
病虫害高发期识别
high_risk_periods
=
苗期中后期
("jointing",
15,
0.2:
schedule.append({
"day":
day_counter,
"operation":
"pest_control",
"type":
"disease_prevention",
"target":
"fungal_diseases",
"method":
"spraying",
"concentration":
"low"
"medium",
"priority":
"medium"
})
#
抽穗期蚜虫防治
if
"jointi
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
