重要前提
安装AI Skills的关键前提是:必须科学上网,且开启TUN模式,这一点至关重要,直接决定安装能否顺利完成,在此郑重提醒三遍:科学上网,科学上网,科学上网。查看完整安装教程 →
farming-expert by personamanagmentlayer/pcl
npx skills add https://github.com/personamanagmentlayer/pcl --skill farming-expert为精准农业、农场管理系统、作物监测、物联网传感器和农业技术提供专业指导。
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
from enum import Enum
class CropType(Enum):
WHEAT = "wheat"
CORN = "corn"
SOYBEANS = "soybeans"
RICE = "rice"
VEGETABLES = "vegetables"
class GrowthStage(Enum):
PLANTED = "planted"
GERMINATION = "germination"
VEGETATIVE = "vegetative"
FLOWERING = "flowering"
HARVEST_READY = "harvest_ready"
HARVESTED = "harvested"
@dataclass
class Field:
field_id: str
name: str
area_hectares: float
soil_type: str
coordinates: List[tuple] # GPS polygon
irrigation_system: str
drainage_quality: str
@dataclass
class CropCycle:
cycle_id: str
field_id: str
crop_type: CropType
variety: str
planting_date: datetime
expected_harvest_date: datetime
growth_stage: GrowthStage
seed_rate: float
fertilizer_applied: List[dict]
pesticides_applied: List[dict]
irrigation_schedule: List[dict]
class FarmManagementSystem:
"""农场管理与作物追踪"""
def __init__(self, db):
self.db = db
def plan_crop_rotation(self, field_id, years=3):
"""生成作物轮作计划"""
field = self.db.get_field(field_id)
history = self.db.get_crop_history(field_id, years=10)
# 分析土壤养分和先前作物
rotation_plan = []
# 轮作规则:
# - 交替种植固氮作物和需氮作物
# - 避免连续种植同一科作物
# - 考虑土壤健康和病虫害管理
for year in range(years):
recommended_crop = self.recommend_next_crop(field, history, year)
rotation_plan.append({
'year': datetime.now().year + year,
'crop': recommended_crop,
'reason': self.explain_recommendation(recommended_crop, history)
})
return rotation_plan
def monitor_crop_health(self, field_id):
"""使用传感器数据监测作物健康状况"""
field = self.db.get_field(field_id)
current_crop = self.db.get_current_crop(field_id)
# 收集传感器数据
soil_moisture = self.get_soil_moisture_data(field_id)
weather_data = self.get_weather_data(field.coordinates)
ndvi_data = self.get_ndvi_from_satellite(field.coordinates)
# 分析健康指标
health_score = self.calculate_health_score(
soil_moisture,
weather_data,
ndvi_data,
current_crop
)
alerts = []
if soil_moisture < current_crop.optimal_moisture_min:
alerts.append({
'type': 'irrigation_needed',
'severity': 'high',
'message': '土壤湿度低于最佳水平'
})
if ndvi_data < 0.6: # 植被健康阈值
alerts.append({
'type': 'crop_stress',
'severity': 'medium',
'message': 'NDVI 指示可能存在作物胁迫'
})
return {
'field_id': field_id,
'health_score': health_score,
'soil_moisture': soil_moisture,
'ndvi': ndvi_data,
'alerts': alerts,
'recommendations': self.generate_recommendations(alerts)
}
def predict_yield(self, field_id):
"""使用机器学习预测作物产量"""
field = self.db.get_field(field_id)
current_crop = self.db.get_current_crop(field_id)
# 预测特征
features = {
'field_area': field.area_hectares,
'soil_type': field.soil_type,
'crop_variety': current_crop.variety,
'days_since_planting': (datetime.now() - current_crop.planting_date).days,
'total_rainfall': self.get_accumulated_rainfall(field_id),
'avg_temperature': self.get_avg_temperature(field_id),
'fertilizer_amount': sum(f['amount'] for f in current_crop.fertilizer_applied),
'ndvi_avg': self.get_avg_ndvi(field_id)
}
# 使用训练好的模型预测产量
predicted_yield_per_hectare = self.yield_model.predict([features])[0]
total_yield = predicted_yield_per_hectare * field.area_hectares
return {
'field_id': field_id,
'predicted_yield_kg': total_yield,
'yield_per_hectare': predicted_yield_per_hectare,
'confidence': 0.85,
'expected_harvest_date': current_crop.expected_harvest_date
}
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
class AgricultureIoT:
"""物联网传感器数据收集与分析"""
def process_soil_sensor_data(self, sensor_id):
"""处理土壤传感器读数"""
readings = self.db.get_recent_readings(sensor_id, hours=24)
analysis = {
'sensor_id': sensor_id,
'avg_moisture': np.mean([r['moisture'] for r in readings]),
'avg_temperature': np.mean([r['temperature'] for r in readings]),
'avg_ph': np.mean([r['ph'] for r in readings]),
'avg_ec': np.mean([r['ec'] for r in readings]), # 电导率
'nitrogen_level': np.mean([r['nitrogen'] for r in readings]),
'phosphorus_level': np.mean([r['phosphorus'] for r in readings]),
'potassium_level': np.mean([r['potassium'] for r in readings])
}
# 检测异常
anomalies = []
if analysis['avg_moisture'] < 20:
anomalies.append('土壤湿度低 - 建议灌溉')
if analysis['avg_ph'] < 5.5 or analysis['avg_ph'] > 7.5:
anomalies.append(f'土壤 pH 超出最佳范围: {analysis["avg_ph"]:.1f}')
analysis['anomalies'] = anomalies
return analysis
def automate_irrigation(self, field_id):
"""自动化灌溉控制"""
field = self.db.get_field(field_id)
soil_moisture = self.get_soil_moisture_data(field_id)
weather_forecast = self.get_weather_forecast(field.coordinates, days=3)
# 决策逻辑
should_irrigate = False
duration_minutes = 0
# 检查是否需要灌溉
if soil_moisture < field.moisture_threshold:
# 检查是否预期有降雨
expected_rainfall = sum(day['rainfall_mm'] for day in weather_forecast)
if expected_rainfall < 10: # 预期降雨少于 10 毫米
should_irrigate = True
# 计算灌溉时长
moisture_deficit = field.moisture_threshold - soil_moisture
duration_minutes = int(moisture_deficit * field.area_hectares * 60 / field.irrigation_rate)
if should_irrigate:
self.activate_irrigation(field_id, duration_minutes)
return {
'field_id': field_id,
'irrigation_activated': should_irrigate,
'duration_minutes': duration_minutes,
'reason': '土壤湿度低于阈值' if should_irrigate else '无需灌溉'
}
class WeatherAnalytics:
"""基于天气的农业决策"""
def analyze_growing_conditions(self, field_id, date_range):
"""分析天气对作物的适宜性"""
weather_data = self.get_historical_weather(field_id, date_range)
# 计算生长度日数
gdd = sum([
max(0, (day['temp_max'] + day['temp_min']) / 2 - 10)
for day in weather_data
])
# 分析霜冻风险
frost_days = len([d for d in weather_data if d['temp_min'] < 0])
# 水平衡
total_rainfall = sum(d['rainfall_mm'] for d in weather_data)
total_evapotranspiration = sum(d['et_mm'] for d in weather_data)
water_deficit = total_evapotranspiration - total_rainfall
return {
'growing_degree_days': gdd,
'frost_days': frost_days,
'total_rainfall_mm': total_rainfall,
'water_deficit_mm': water_deficit,
'avg_temperature': np.mean([d['temp_avg'] for d in weather_data]),
'suitability_score': self.calculate_suitability_score(gdd, frost_days, water_deficit)
}
def optimize_planting_date(self, field_id, crop_type):
"""确定最佳播种日期"""
historical_weather = self.get_historical_weather(field_id, years=10)
crop_requirements = self.get_crop_requirements(crop_type)
# 寻找满足以下条件的窗口:
# - 霜冻风险最小
# - 土壤温度适宜
# - 降雨分布良好
optimal_dates = []
for year_data in historical_weather:
for date, conditions in year_data.items():
score = self.score_planting_conditions(
conditions,
crop_requirements
)
optimal_dates.append((date, score))
# 返回最佳播种窗口
best_dates = sorted(optimal_dates, key=lambda x: x[1], reverse=True)[:10]
return {
'recommended_planting_window': {
'start': best_dates[-1][0],
'end': best_dates[0][0]
},
'confidence': np.mean([d[1] for d in best_dates])
}
class PestManagement:
"""病虫害监测与管理"""
def detect_pest_risk(self, field_id):
"""预测病虫害压力"""
weather_data = self.get_recent_weather(field_id, days=14)
crop = self.db.get_current_crop(field_id)
# 影响病虫害的环境因素
avg_temp = np.mean([d['temperature'] for d in weather_data])
avg_humidity = np.mean([d['humidity'] for d in weather_data])
rainfall = sum(d['rainfall_mm'] for d in weather_data)
risk_factors = {
'temperature_risk': self.assess_temp_risk(avg_temp, crop.crop_type),
'humidity_risk': self.assess_humidity_risk(avg_humidity),
'rainfall_risk': self.assess_rainfall_risk(rainfall)
}
overall_risk = sum(risk_factors.values()) / len(risk_factors)
recommendations = []
if overall_risk > 0.7:
recommendations.append('巡查田地寻找病虫害活动')
recommendations.append('考虑预防性处理')
elif overall_risk > 0.5:
recommendations.append('增加监测频率')
return {
'field_id': field_id,
'overall_risk_score': overall_risk,
'risk_level': 'high' if overall_risk > 0.7 else 'medium' if overall_risk > 0.4 else 'low',
'risk_factors': risk_factors,
'recommendations': recommendations
}
def analyze_crop_image(self, image_data):
"""使用机器学习从作物图像中检测病害"""
# 将使用在作物病害上训练的计算机视觉模型
# 返回检测到的病害和置信度分数
predictions = self.disease_detection_model.predict(image_data)
return {
'detected_diseases': predictions['diseases'],
'confidence': predictions['confidence'],
'affected_area_percent': predictions['affected_area'],
'treatment_recommendations': self.get_treatment_plan(predictions['diseases'])
}
❌ 过量施用投入品 ❌ 忽视土壤健康 ❌ 不进行作物轮作 ❌ 仅进行手动数据收集 ❌ 忽视天气预报 ❌ 被动管理而非主动管理 ❌ 不进行产量分析
每周安装量
66
代码仓库
GitHub 星标数
13
首次出现
2026年1月24日
安全审计
安装于
opencode57
codex56
gemini-cli54
cursor52
github-copilot50
kimi-cli48
Expert guidance for precision agriculture, farm management systems, crop monitoring, IoT sensors, and agricultural technology.
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
from enum import Enum
class CropType(Enum):
WHEAT = "wheat"
CORN = "corn"
SOYBEANS = "soybeans"
RICE = "rice"
VEGETABLES = "vegetables"
class GrowthStage(Enum):
PLANTED = "planted"
GERMINATION = "germination"
VEGETATIVE = "vegetative"
FLOWERING = "flowering"
HARVEST_READY = "harvest_ready"
HARVESTED = "harvested"
@dataclass
class Field:
field_id: str
name: str
area_hectares: float
soil_type: str
coordinates: List[tuple] # GPS polygon
irrigation_system: str
drainage_quality: str
@dataclass
class CropCycle:
cycle_id: str
field_id: str
crop_type: CropType
variety: str
planting_date: datetime
expected_harvest_date: datetime
growth_stage: GrowthStage
seed_rate: float
fertilizer_applied: List[dict]
pesticides_applied: List[dict]
irrigation_schedule: List[dict]
class FarmManagementSystem:
"""Farm management and crop tracking"""
def __init__(self, db):
self.db = db
def plan_crop_rotation(self, field_id, years=3):
"""Generate crop rotation plan"""
field = self.db.get_field(field_id)
history = self.db.get_crop_history(field_id, years=10)
# Analyze soil nutrients and previous crops
rotation_plan = []
# Rules for rotation:
# - Alternate nitrogen-fixing and nitrogen-demanding crops
# - Avoid same crop family consecutively
# - Consider soil health and pest management
for year in range(years):
recommended_crop = self.recommend_next_crop(field, history, year)
rotation_plan.append({
'year': datetime.now().year + year,
'crop': recommended_crop,
'reason': self.explain_recommendation(recommended_crop, history)
})
return rotation_plan
def monitor_crop_health(self, field_id):
"""Monitor crop health using sensor data"""
field = self.db.get_field(field_id)
current_crop = self.db.get_current_crop(field_id)
# Collect sensor data
soil_moisture = self.get_soil_moisture_data(field_id)
weather_data = self.get_weather_data(field.coordinates)
ndvi_data = self.get_ndvi_from_satellite(field.coordinates)
# Analyze health indicators
health_score = self.calculate_health_score(
soil_moisture,
weather_data,
ndvi_data,
current_crop
)
alerts = []
if soil_moisture < current_crop.optimal_moisture_min:
alerts.append({
'type': 'irrigation_needed',
'severity': 'high',
'message': 'Soil moisture below optimal level'
})
if ndvi_data < 0.6: # Vegetation health threshold
alerts.append({
'type': 'crop_stress',
'severity': 'medium',
'message': 'NDVI indicates possible crop stress'
})
return {
'field_id': field_id,
'health_score': health_score,
'soil_moisture': soil_moisture,
'ndvi': ndvi_data,
'alerts': alerts,
'recommendations': self.generate_recommendations(alerts)
}
def predict_yield(self, field_id):
"""Predict crop yield using ML"""
field = self.db.get_field(field_id)
current_crop = self.db.get_current_crop(field_id)
# Features for prediction
features = {
'field_area': field.area_hectares,
'soil_type': field.soil_type,
'crop_variety': current_crop.variety,
'days_since_planting': (datetime.now() - current_crop.planting_date).days,
'total_rainfall': self.get_accumulated_rainfall(field_id),
'avg_temperature': self.get_avg_temperature(field_id),
'fertilizer_amount': sum(f['amount'] for f in current_crop.fertilizer_applied),
'ndvi_avg': self.get_avg_ndvi(field_id)
}
# Use trained model to predict yield
predicted_yield_per_hectare = self.yield_model.predict([features])[0]
total_yield = predicted_yield_per_hectare * field.area_hectares
return {
'field_id': field_id,
'predicted_yield_kg': total_yield,
'yield_per_hectare': predicted_yield_per_hectare,
'confidence': 0.85,
'expected_harvest_date': current_crop.expected_harvest_date
}
class AgricultureIoT:
"""IoT sensor data collection and analysis"""
def process_soil_sensor_data(self, sensor_id):
"""Process soil sensor readings"""
readings = self.db.get_recent_readings(sensor_id, hours=24)
analysis = {
'sensor_id': sensor_id,
'avg_moisture': np.mean([r['moisture'] for r in readings]),
'avg_temperature': np.mean([r['temperature'] for r in readings]),
'avg_ph': np.mean([r['ph'] for r in readings]),
'avg_ec': np.mean([r['ec'] for r in readings]), # Electrical conductivity
'nitrogen_level': np.mean([r['nitrogen'] for r in readings]),
'phosphorus_level': np.mean([r['phosphorus'] for r in readings]),
'potassium_level': np.mean([r['potassium'] for r in readings])
}
# Detect anomalies
anomalies = []
if analysis['avg_moisture'] < 20:
anomalies.append('Low soil moisture - irrigation recommended')
if analysis['avg_ph'] < 5.5 or analysis['avg_ph'] > 7.5:
anomalies.append(f'Soil pH out of optimal range: {analysis["avg_ph"]:.1f}')
analysis['anomalies'] = anomalies
return analysis
def automate_irrigation(self, field_id):
"""Automated irrigation control"""
field = self.db.get_field(field_id)
soil_moisture = self.get_soil_moisture_data(field_id)
weather_forecast = self.get_weather_forecast(field.coordinates, days=3)
# Decision logic
should_irrigate = False
duration_minutes = 0
# Check if irrigation is needed
if soil_moisture < field.moisture_threshold:
# Check if rain is expected
expected_rainfall = sum(day['rainfall_mm'] for day in weather_forecast)
if expected_rainfall < 10: # Less than 10mm expected
should_irrigate = True
# Calculate irrigation duration
moisture_deficit = field.moisture_threshold - soil_moisture
duration_minutes = int(moisture_deficit * field.area_hectares * 60 / field.irrigation_rate)
if should_irrigate:
self.activate_irrigation(field_id, duration_minutes)
return {
'field_id': field_id,
'irrigation_activated': should_irrigate,
'duration_minutes': duration_minutes,
'reason': 'Soil moisture below threshold' if should_irrigate else 'No irrigation needed'
}
class WeatherAnalytics:
"""Weather-based agricultural decisions"""
def analyze_growing_conditions(self, field_id, date_range):
"""Analyze weather suitability for crops"""
weather_data = self.get_historical_weather(field_id, date_range)
# Calculate growing degree days (GDD)
gdd = sum([
max(0, (day['temp_max'] + day['temp_min']) / 2 - 10)
for day in weather_data
])
# Analyze frost risk
frost_days = len([d for d in weather_data if d['temp_min'] < 0])
# Water balance
total_rainfall = sum(d['rainfall_mm'] for d in weather_data)
total_evapotranspiration = sum(d['et_mm'] for d in weather_data)
water_deficit = total_evapotranspiration - total_rainfall
return {
'growing_degree_days': gdd,
'frost_days': frost_days,
'total_rainfall_mm': total_rainfall,
'water_deficit_mm': water_deficit,
'avg_temperature': np.mean([d['temp_avg'] for d in weather_data]),
'suitability_score': self.calculate_suitability_score(gdd, frost_days, water_deficit)
}
def optimize_planting_date(self, field_id, crop_type):
"""Determine optimal planting date"""
historical_weather = self.get_historical_weather(field_id, years=10)
crop_requirements = self.get_crop_requirements(crop_type)
# Find window with:
# - Minimal frost risk
# - Adequate soil temperature
# - Good rainfall distribution
optimal_dates = []
for year_data in historical_weather:
for date, conditions in year_data.items():
score = self.score_planting_conditions(
conditions,
crop_requirements
)
optimal_dates.append((date, score))
# Return best planting window
best_dates = sorted(optimal_dates, key=lambda x: x[1], reverse=True)[:10]
return {
'recommended_planting_window': {
'start': best_dates[-1][0],
'end': best_dates[0][0]
},
'confidence': np.mean([d[1] for d in best_dates])
}
class PestManagement:
"""Pest and disease monitoring and management"""
def detect_pest_risk(self, field_id):
"""Predict pest pressure"""
weather_data = self.get_recent_weather(field_id, days=14)
crop = self.db.get_current_crop(field_id)
# Environmental factors affecting pests
avg_temp = np.mean([d['temperature'] for d in weather_data])
avg_humidity = np.mean([d['humidity'] for d in weather_data])
rainfall = sum(d['rainfall_mm'] for d in weather_data)
risk_factors = {
'temperature_risk': self.assess_temp_risk(avg_temp, crop.crop_type),
'humidity_risk': self.assess_humidity_risk(avg_humidity),
'rainfall_risk': self.assess_rainfall_risk(rainfall)
}
overall_risk = sum(risk_factors.values()) / len(risk_factors)
recommendations = []
if overall_risk > 0.7:
recommendations.append('Scout fields for pest activity')
recommendations.append('Consider preventive treatment')
elif overall_risk > 0.5:
recommendations.append('Increase monitoring frequency')
return {
'field_id': field_id,
'overall_risk_score': overall_risk,
'risk_level': 'high' if overall_risk > 0.7 else 'medium' if overall_risk > 0.4 else 'low',
'risk_factors': risk_factors,
'recommendations': recommendations
}
def analyze_crop_image(self, image_data):
"""Detect diseases from crop images using ML"""
# Would use computer vision model (CNN) trained on crop diseases
# Returns detected diseases and confidence scores
predictions = self.disease_detection_model.predict(image_data)
return {
'detected_diseases': predictions['diseases'],
'confidence': predictions['confidence'],
'affected_area_percent': predictions['affected_area'],
'treatment_recommendations': self.get_treatment_plan(predictions['diseases'])
}
❌ Over-application of inputs ❌ Ignoring soil health ❌ No crop rotation ❌ Manual data collection only ❌ Ignoring weather forecasts ❌ Reactive instead of proactive management ❌ No yield analysis
Weekly Installs
66
Repository
GitHub Stars
13
First Seen
Jan 24, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
opencode57
codex56
gemini-cli54
cursor52
github-copilot50
kimi-cli48
前端代码审计工具 - 自动化检测可访问性、性能、响应式设计、主题化与反模式
49,600 周安装