垂直農法の自動化: API を介したロボット制御
ミラノ郊外のカヴェナーゴ・ディ・ブリアンツァにある廃墟となった工場倉庫。内部、9,000メートル 青赤 LED ライトの下でレタス、バジル、ルッコラが育つ正方形の垂直棚 太陽を見ることもなく。温度21度、湿度70%、CO21,200ppmの一定環境。あらゆる植物 パーソナライズされた照明レシピがあり、各栽培ラインは API を介して中央システムと通信し、 各ロボットはリアルタイムでどこに行くべきかを知っています。これは プラネット ファームズ、イタリアの垂直農場 これは 2025 年には屋内自動農業の世界基準の 1 つになりました。
垂直農法はもはや実験室の実験ではありません。 2025 年の世界市場の価値は 96.2億ドル 2033 年までに 392 億人になると予測されています (CAGR 19.3%)。 新世代の垂直農場は、 水を95%削減 尊重する 伝統的な農業を取り入れ、気候に関係なく一年中生産し、農薬を使用しません。 露地よりも最大 100 倍高い平方メートル当たりの収量を達成します。しかし、この結果は これらは、産業グレードのソフトウェアとロボティクス インフラストラクチャによってのみ可能になります。
この記事では、環境センサーから垂直農場のソフトウェア アーキテクチャ全体を構築します。 Python の PID コントローラー、FastAPI を使用した REST API 設計からロボット用の ROS2 統合まで、 デジタルツインから強化学習ベースの最適化パイプラインまで。実用的なコード、 本番環境でテストされたアーキテクチャ、イタリアのエコシステムからの実数。
この記事で学べること
- 垂直農場向けの完全なソフトウェア アーキテクチャ: センサー、コントローラー、SCADA、クラウド、AI
- LED スペクトル管理: PAR、DLI、レタス、バジル、イチゴの照明レシピ
- 温度、CO2、湿度、灌漑用の Python による PID コントローラーの実装
- 作物レシピ管理とアクチュエータ制御のための FastAPI を使用した REST API 設計
- 自動ロボット播種、移植、収穫のための ROS2 統合
- 農場のデジタルツイン: 植物の成長シミュレーションとレイアウトの最適化
- ライトレシピを最適化するための強化学習
- IoT インフラストラクチャ: Modbus RTU、MQTT、OPC-UA、産業用ゲートウェイ
- 経済分析: CAPEX、OPEX、損益分岐点、屋外農業との比較
- ケーススタディ: イタリアの垂直農場大手、プラネット ファームズとアグリコラ モデルナ
フードテックシリーズ - すべての記事
| # | アイテム | レベル | Stato |
|---|---|---|---|
| 1 | Python と MQTT を使用した精密農業用の IoT パイプライン | 高度な | 発行済み |
| 2 | 作物監視のための ML Edge: 圃場でのコンピュータ ビジョン | 高度な | 発行済み |
| 3 | 衛星 API と植生インデックス: Python と Sentinel-2 を使用した NDVI | 中級 | 発行済み |
| 4 | 食品におけるブロックチェーンのトレーサビリティ: 現場からスーパーマーケットまで | 中級 | 発行済み |
| 5 | 食品産業における品質管理のためのコンピュータービジョン | 高度な | 発行済み |
| 6 | FSMA とデジタル コンプライアンス: 規制プロセスの自動化 | 中級 | 発行済み |
| 7 | 垂直農法の自動化: API を介したロボット制御 (ここにいます) | 高度な | 現在 |
| 8 | Prophet と LightGBM を使用した食品小売の需要予測 | 中級 | 近日公開 |
| 9 | ファーム インテリジェンス ダッシュボード: Grafana を使用したリアルタイム分析 | 中級 | 近日公開 |
| 10 | サプライチェーンの食品の最適化: 廃棄物削減のための ML | 中級 | 近日公開 |
2025 年の垂直農業市場: 成長と技術的推進力
垂直農業は 2023 年から 2024 年にかけて統合段階に入り、一部の大手企業が参入する 負担により破産を宣告した北米企業（AeroFarms、AppHarvest、Bowery Farming） 非常に高額な設備投資と爆発的なエネルギーコスト。しかし市場は止まっていない、再構築されている より効率的なモデルを中心に、構築した人々に報いるダーウィンの選択を採用 スケーリング前の確かなユニットエコノミクス。
2025 年、数字は成熟度の高まりを物語ります: 世界市場にはそれだけの価値があります 96.2億ドル 2033 年までに 392 億人に増加すると予想されています。別の推計も (市場調査の最大化) によると、2025 年には 80 億人となり、2025 年までに 397 億人になると予測されています。 2032 年には 25.7% の CAGR で成長します。推定値のばらつきは、何が構成要素かを分類することの難しさを反映しています。 まさに「垂直農法」（屋内の積み重ねトレイのみ？先進的な温室も？）ですが、トレンドは そして明白です。コネクテッド農業ロボット市場は、2025 年には 102 億 3,000 万の価値がある。 2030 年までに 282 億人に増加します。
垂直農法と伝統的農業: パフォーマンスの比較
| パラメータ | 屋外農業 | 温室 | 垂直農場 |
|---|---|---|---|
| 水の消費量（相対） | 100% | 30-40% | 5～10% |
| ㎡あたりの収量（レタス） | ~2 kg/m²/年 | ~15 kg/m²/年 | ~150-200 kg/m²/年 |
| 生産サイクル/年 | 1-3 | 4-8 | 12-18 |
| 農薬 | 必要性が高い | 減少 | ゼロ |
| 気候依存性 | 合計 | 部分的 | なし |
| 土地利用（土地） | 1x | 1x | 0.01～0.05倍 |
| 必要なエネルギー | 低い | 平均 | 高 (LED + HVAC) |
| 生産費（レタス） | 0.5～1ドル/kg | 1.5～3ドル/kg | 4～8ドル/kg |
2025 年の成長の原動力は 4 つあります。まず、高効率 LED のコストが低下しました。 過去 10 年間で 70% 増加し、照明設備投資がはるかに利用しやすくなりました。第二に、私は AI ベースの制御システムにより、最長 3 ～ 4 年間にわたる想像を絶するエネルギーの最適化が可能になります。 前に。第三に、地元産の新鮮で無農薬の製品に対する需要は常に高まっており、 特に都市部では。 4番目は、播種、移植、収穫用のロボットモジュールです。 現在では、実験室用の価格ではなく、工業用価格で入手できるようになりました。
垂直ファームのソフトウェア アーキテクチャ: 完全なスタック
現代の垂直農場は本質的にサイバー物理システムです。あらゆる物理的な決定（オンにする） LED、バルブを開ける、ロボットを動かす）とソフトウェア計算の結果。アーキテクチャ 制御に関してはリアルタイムであり、作物の安全に関しては信頼性があり、拡張可能である必要があります。 何百もの栽培地を管理します。
エンドツーエンドのアーキテクチャ スタック
+-----------------------------------------------------------------------+
| LAYER 1: FIELD DEVICES |
| [Sensori Temp/RH] [CO2 Sensor] [PAR Meter] [Nutrient EC/pH] |
| [Flow Sensor] [Camera RGB-D] [Weight Sensor] [RFID Tray] |
| | | | | |
| +------------------+-------------+--------------+ |
| Modbus RTU / RS-485 / I2C / SPI |
+-----------------------------------------------------------------------+
|
+-----------------------------------------------------------------------+
| LAYER 2: EDGE CONTROLLER |
| [PLC Siemens S7-1500 / Beckhoff CX / Raspberry] |
| - Loop PID per temperatura, CO2, umidita |
| - Scheduling ricette luminose (LED driver DMX/PWM) |
| - Gestione irrigazione NFT/DWC cicli |
| - Buffer locale offline-tolerant |
| - OPC-UA Server / MQTT Publisher |
+-----------------------------------------------------------------------+
|
+-----------------------------------------------------------------------+
| LAYER 3: SCADA / MES |
| [Ignition SCADA / custom Python SCADA] |
| - Supervisione multi-zona real-time |
| - Storicizzazione time-series (InfluxDB/TimescaleDB) |
| - Allarmi e notifiche (temperatura, EC, pH out-of-range) |
| - Ricette colturali e scheduling batch |
+-----------------------------------------------------------------------+
|
+-----------------------------------------------------------------------+
| LAYER 4: CLOUD PLATFORM |
| [FastAPI Backend] [Message Broker MQTT/Kafka] [PostgreSQL] |
| - REST API per integrazione ERP/WMS/retail |
| - Gestione ricette, batch, inventory, ordini |
| - Autenticazione OAuth2, RBAC, audit log |
+-----------------------------------------------------------------------+
|
+-----------------------------------------------------------------------+
| LAYER 5: AI / ANALYTICS |
| [ML Pipeline] [Digital Twin] [Computer Vision] [RL Optimizer] |
| - Ottimizzazione ricette luminose (RL) |
| - Previsione resa e time-to-harvest |
| - Rilevamento anomalie (sensori + visione) |
| - Simulazione crescita (digital twin) |
+-----------------------------------------------------------------------+
産業用 PLC (Siemens S7、Beckhoff) とシングルボード コンピューター (Raspberry Pi 4、 BeagleBone) は、必要な信頼性のレベルによって異なります。収穫量が多い商業農場向け 値、ハードウェア冗長性と正しい選択を備えた IEC 61131-3 認定 PLC。プロトタイプや 実験ファーム、組み込みハードウェア上の Python ソリューション、より柔軟で迅速な開発が可能です。
環境制御システム: LED、HVAC、CO2、灌漑
環境制御は垂直農場の運営の中心です。 4 つのパラメーターが支配的です。 光のスペクトルと強度、気温、CO2濃度、 養液の組成。それぞれに専用の制御ループが必要です。
LED スペクトル管理: PAR、DLI、および照明レシピ
植物はすべての可視光を均等に使用するわけではありません。光合成が活発な範囲 から行く 400～700nm (PAR - 光合成活性放射線)。内部 この範囲では、青 (400 ～ 500 nm) が葉の形態と化合物の合成を制御します。 芳香族化合物。赤 (600 ～ 700 nm) は光合成の主な推進力です。遠赤色 (700-800 nm) フィトクロム システムを介して開花と植物の形状に影響を与えます。
DLI (Daily Light Integral) は、植物が受け取る PAR 光子の総量を測定します。 24 時間で、mol/m²/日で表されます。レシピのサイズを決定するための最も重要な指標 明るい。 2025 年に Nature Scientific Reports に掲載された研究では、LED が最適化されていることが示されています。 垂直農法の場合、彼らは最大で 収量が 32% 増加 スペクトルと比較して 標準で、レタスとバジルの新鮮な重量が大きくなります。
主要作物の光パラメータ
| 文化 | PPFD (μmol/m²/s) | DLI 目標 (mol/m²/日) | 最適なスペクトル | 光周期 |
|---|---|---|---|---|
| レタス（葉） | 150-250 | 午後3時から午後6時まで | R:B = 4:1、+遠赤 5% | 16～18時間の明るさ |
| レタス（頭） | 200-300 | 17-22 | R:B = 3:1、UV 380nm 2% | 16時間ライト |
| バジル | 200-300 | 午後2時から午後5時まで | R:B = 3:1、青 20-25% | 16時間ライト |
| ほうれん草 | 150-200 | 12-17 | R:B = 4:1 | 14～16時間の明るさ |
| イチゴ（植物性） | 200-300 | 午後3時から午後8時まで | R:B = 3:1 | 16時間ライト |
| イチゴ（開花中） | 300-400 | 20-25 | R:B:FR = 3:1:0.5 | 12時間点灯 |
| マイクログリーン | 100-200 | 8-12 | フルスペクトルホワイト | 16時間ライト |
| 芳香のあるハーブ | 200-250 | 14-16 | 青 15%、R 優勢 | 16時間ライト |
LED 管理コントローラーは、これらのレシピを LED ドライバーへの PWM 信号に変換する必要があります。 栽培地域ごとにレシピが異なる場合があり、レシピは年間を通じて変更される場合があります。 栽培サイクル (例: バジルの香りを強めるために、最後の 48 時間でより青みが増します)。 完全な PID コントローラーの Python 実装は次のとおりです。
"""
Vertical Farm Environmental Controller
Controller PID per gestione LED, CO2, temperatura e irrigazione
"""
import time
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
logger = logging.getLogger(__name__)
class CropStage(Enum):
GERMINATION = "germination"
SEEDLING = "seedling"
VEGETATIVE = "vegetative"
MATURATION = "maturation"
HARVEST_READY = "harvest_ready"
@dataclass
class LightRecipe:
"""Ricetta luminosa per una coltura in un determinato stadio"""
crop_name: str
stage: CropStage
ppfd_target: float # µmol/m²/s
dli_target: float # mol/m²/giorno
photoperiod_hours: float # ore di luce al giorno
spectrum_red_pct: float # % canale rosso (620-700nm)
spectrum_blue_pct: float # % canale blu (400-500nm)
spectrum_white_pct: float # % LED bianco full-spectrum
spectrum_farred_pct: float # % far-red (700-800nm)
spectrum_uv_pct: float = 0.0
def validate(self) -> bool:
total = (self.spectrum_red_pct + self.spectrum_blue_pct +
self.spectrum_white_pct + self.spectrum_farred_pct +
self.spectrum_uv_pct)
return abs(total - 100.0) < 0.1
def ppfd_from_dli(self) -> float:
"""Calcola PPFD target dalle ore di fotoperiodo e DLI"""
photoperiod_seconds = self.photoperiod_hours * 3600
return (self.dli_target * 1_000_000) / photoperiod_seconds
@dataclass
class PIDController:
"""
Controller PID generico per parametri ambientali.
Usa anti-windup per prevenire integrator saturation.
"""
kp: float
ki: float
kd: float
setpoint: float
output_min: float = 0.0
output_max: float = 100.0
_integral: float = field(default=0.0, init=False)
_last_error: float = field(default=0.0, init=False)
_last_time: float = field(default_factory=time.time, init=False)
def compute(self, measured_value: float) -> float:
current_time = time.time()
dt = current_time - self._last_time
if dt <= 0:
return self._last_output if hasattr(self, '_last_output') else 0.0
error = self.setpoint - measured_value
# Proporzionale
p_term = self.kp * error
# Integrale con anti-windup (clamping)
self._integral += error * dt
i_term = self.ki * self._integral
# Clamp integrale per prevenire windup
i_max = (self.output_max - self.output_min) / self.ki if self.ki != 0 else 1000
self._integral = max(-i_max, min(i_max, self._integral))
i_term = self.ki * self._integral
# Derivativo
d_term = self.kd * (error - self._last_error) / dt if dt > 0 else 0.0
output = p_term + i_term + d_term
output = max(self.output_min, min(self.output_max, output))
self._last_error = error
self._last_time = current_time
self._last_output = output
return output
class EnvironmentalController:
"""
Controller principale per una zona di coltivazione.
Gestisce temperatura, CO2, umidita e irrigazione via PID.
"""
def __init__(self, zone_id: str, recipe: LightRecipe):
self.zone_id = zone_id
self.recipe = recipe
# PID temperatura: setpoint 21°C, banda ±1°C
self.temp_pid = PIDController(
kp=2.0, ki=0.5, kd=0.1,
setpoint=21.0,
output_min=-100.0, # raffreddamento
output_max=100.0 # riscaldamento
)
# PID CO2: setpoint 1200 ppm per crescita accelerata
self.co2_pid = PIDController(
kp=0.5, ki=0.1, kd=0.05,
setpoint=1200.0,
output_min=0.0,
output_max=100.0 # % apertura valvola CO2
)
# PID umidita relativa: setpoint 70%
self.humidity_pid = PIDController(
kp=1.5, ki=0.3, kd=0.05,
setpoint=70.0,
output_min=0.0,
output_max=100.0
)
def compute_led_pwm(self, current_hour: float) -> dict:
"""
Calcola duty cycle PWM per ogni canale LED
in base all'ora del giorno e alla ricetta.
"""
# Determina se siamo nel fotoperiodo attivo
# Fotoperiodo: ore 6:00 - (6 + photoperiod_hours)
start_hour = 6.0
end_hour = start_hour + self.recipe.photoperiod_hours
if not (start_hour <= current_hour < end_hour):
return {
'red': 0.0, 'blue': 0.0,
'white': 0.0, 'farred': 0.0, 'uv': 0.0
}
# Calcola intensità normalizzata (0-1) dal PPFD target
# Assumendo che 100% PWM = 600 µmol/m²/s
max_ppfd = 600.0
intensity = min(self.recipe.ppfd_target / max_ppfd, 1.0)
return {
'red': round(intensity * self.recipe.spectrum_red_pct / 100, 4),
'blue': round(intensity * self.recipe.spectrum_blue_pct / 100, 4),
'white': round(intensity * self.recipe.spectrum_white_pct / 100, 4),
'farred': round(intensity * self.recipe.spectrum_farred_pct / 100, 4),
'uv': round(intensity * self.recipe.spectrum_uv_pct / 100, 4),
}
async def control_loop(self, sensor_reader, actuator_writer, interval: float = 30.0):
"""
Loop di controllo asincrono: legge sensori, calcola PID, scrive attuatori.
Frequenza default: ogni 30 secondi.
"""
logger.info(f"Avvio loop controllo zona {self.zone_id}")
while True:
try:
# Leggi sensori
sensors = await sensor_reader.read_zone(self.zone_id)
# Calcola output PID
temp_output = self.temp_pid.compute(sensors['temperature'])
co2_output = self.co2_pid.compute(sensors['co2_ppm'])
humidity_output = self.humidity_pid.compute(sensors['humidity_rh'])
# Calcola PWM LED
from datetime import datetime
current_hour = datetime.now().hour + datetime.now().minute / 60.0
led_pwm = self.compute_led_pwm(current_hour)
# Scrivi attuatori
await actuator_writer.set_hvac(self.zone_id, temp_output, humidity_output)
await actuator_writer.set_co2_valve(self.zone_id, co2_output)
await actuator_writer.set_led_channels(self.zone_id, led_pwm)
logger.debug(
f"Zona {self.zone_id} | "
f"T={sensors['temperature']:.1f}°C (PID:{temp_output:.1f}%) | "
f"CO2={sensors['co2_ppm']:.0f}ppm (valve:{co2_output:.1f}%) | "
f"RH={sensors['humidity_rh']:.1f}% | "
f"LED R:{led_pwm['red']:.2f} B:{led_pwm['blue']:.2f}"
)
except Exception as e:
logger.error(f"Errore loop controllo zona {self.zone_id}: {e}")
await asyncio.sleep(interval)
# --- Ricette standard per colture comuni ---
LETTUCE_VEGETATIVE = LightRecipe(
crop_name="Lattuga Lollo",
stage=CropStage.VEGETATIVE,
ppfd_target=200.0,
dli_target=17.0,
photoperiod_hours=16.0,
spectrum_red_pct=65.0,
spectrum_blue_pct=20.0,
spectrum_white_pct=10.0,
spectrum_farred_pct=5.0,
spectrum_uv_pct=0.0,
)
BASIL_VEGETATIVE = LightRecipe(
crop_name="Basilico Genovese",
stage=CropStage.VEGETATIVE,
ppfd_target=250.0,
dli_target=15.0,
photoperiod_hours=16.0,
spectrum_red_pct=60.0,
spectrum_blue_pct=25.0, # blue elevato per aromi
spectrum_white_pct=10.0,
spectrum_farred_pct=5.0,
spectrum_uv_pct=0.0,
)
STRAWBERRY_FLOWERING = LightRecipe(
crop_name="Fragola Elsanta",
stage=CropStage.MATURATION,
ppfd_target=350.0,
dli_target=22.0,
photoperiod_hours=12.0, # fotoperiodo corto per fioritura
spectrum_red_pct=55.0,
spectrum_blue_pct=20.0,
spectrum_white_pct=20.0,
spectrum_farred_pct=5.0,
spectrum_uv_pct=0.0,
)灌漑システム: NFT、DWC、およびエアロポニックス
垂直農場における 3 つの主要な水耕技術には制御アーキテクチャがあります それぞれに監視および制御すべき独自の重要なパラメータがあります。
水耕栽培システムの比較
| システム | 理想的な作物 | 重要なパラメータ | 複雑さの制御 | 水の使用量 |
|---|---|---|---|---|
| NFT (栄養膜技術) | レタス、ハーブ | 流量、流路勾配、EC、pH | 平均 | 最小（再循環） |
| DWC (深海培養) | レタス、ほうれん草 | 酸素化 (DO)、EC、pH、温度 | 低い | ベース |
| エアロポニックス | イチゴ、根っこ、ハーブ | 噴霧サイクル、圧力、EC、pH | 高い | 最小 (90% 対土壌) |
| 基板 (ココナッツ/ロックウール) | トマト、ピーマン | サイクル灌漑、EC、pH、排水 | 平均 | 適度 |
"""
Irrigation Controller per sistema NFT
Gestione pompe, monitoraggio EC/pH, dosaggio nutrienti
"""
import asyncio
from dataclasses import dataclass
from typing import Optional
import logging
logger = logging.getLogger(__name__)
NUTRIENT_TARGETS = {
"lettuce": {"ec_ms_cm": 1.6, "ph": 6.0, "temp_c": 20.0},
"basil": {"ec_ms_cm": 1.8, "ph": 6.0, "temp_c": 21.0},
"spinach": {"ec_ms_cm": 2.0, "ph": 6.2, "temp_c": 20.0},
"strawberry": {"ec_ms_cm": 2.2, "ph": 5.8, "temp_c": 18.0},
"herbs": {"ec_ms_cm": 1.4, "ph": 6.0, "temp_c": 21.0},
}
@dataclass
class NFTController:
zone_id: str
crop_type: str
flow_rate_lpm: float = 1.5 # litri/minuto per canale
channel_slope_pct: float = 2.0 # pendenza canale in %
def get_targets(self) -> dict:
return NUTRIENT_TARGETS.get(self.crop_type, NUTRIENT_TARGETS["lettuce"])
async def check_and_adjust(self, ec_sensor: float, ph_sensor: float,
ec_doser, ph_doser) -> dict:
targets = self.get_targets()
actions = {}
# Controllo EC
ec_delta = targets["ec_ms_cm"] - ec_sensor
if abs(ec_delta) > 0.2:
if ec_delta > 0:
# EC troppo bassa: aggiungi concentrato nutrienti
dose_ml = ec_delta * 50 # ml di concentrato A+B
await ec_doser.dose(zone=self.zone_id, ml=dose_ml, solution="AB")
actions["ec_dosing"] = f"+{dose_ml:.1f}ml AB"
else:
# EC troppo alta: diluisci con acqua RO
await ec_doser.dose_water(zone=self.zone_id, ml=abs(ec_delta) * 100)
actions["ec_dilution"] = f"+{abs(ec_delta)*100:.0f}ml H2O"
# Controllo pH
ph_delta = targets["ph"] - ph_sensor
if abs(ph_delta) > 0.3:
if ph_delta > 0:
# pH troppo basso: aggiungi pH-up (KOH)
dose_ml = abs(ph_delta) * 10
await ph_doser.dose(zone=self.zone_id, ml=dose_ml, solution="ph_up")
actions["ph_adjust"] = f"pH-up +{dose_ml:.1f}ml"
else:
# pH troppo alto: aggiungi pH-down (H3PO4)
dose_ml = abs(ph_delta) * 10
await ph_doser.dose(zone=self.zone_id, ml=dose_ml, solution="ph_down")
actions["ph_adjust"] = f"pH-down +{dose_ml:.1f}ml"
logger.info(
f"NFT zona {self.zone_id} | "
f"EC: {ec_sensor:.2f} (target {targets['ec_ms_cm']:.2f}) | "
f"pH: {ph_sensor:.2f} (target {targets['ph']:.2f}) | "
f"Azioni: {actions}"
)
return actionsロボティクスとオートメーション: 垂直農場における ROS2
ロボット工学は、2025 年の垂直農業において最も変革をもたらす要素です。手動操作 より集中的なのは、播種（トレイに播種）、移植（苗を苗木に移植する）です。 最終ラック）、収穫および梱包。 1人の作業員が約500～700本の植物を移植できます。 今のところ;移植ロボットは、エラー率が低く、毎時 2,000 ～ 3,000 本の植物で動作します。 1%で。 1 日あたり 30,000 トレイのレタスを使用すると (アニデッロのアグリコラ モデルナの場合のように)、 ロボット工学は選択ではなく、必要不可欠なものです。
ROS2 (ロボット オペレーティング システム 2) はロボット プログラミングの事実上の標準になりました 屋内環境で。 ROS1 と比較して、ネイティブのリアルタイム サポート (DDS ミドルウェア) を提供します。 改善されたセキュリティ アーキテクチャ、ネイティブ マルチロボット サポート、管理されたライフサイクル ノード。 ノードとトピックの構造により、計画ロジックを明確に分離できます。 動き、モーター制御、人工視覚、システムとのインターフェース 農場経営のこと。
"""
ROS2 Node per Harvesting Robot in Vertical Farm
Gestisce pianificazione percorso, prelievo e deposito vassoi
"""
import rclpy
from rclpy.node import Node
from rclpy.action import ActionServer
from geometry_msgs.msg import Pose, PoseStamped
from std_msgs.msg import String, Bool
from sensor_msgs.msg import Image
import json
import asyncio
# Messaggi custom per la farm (definiti in farm_interfaces package)
# from farm_interfaces.msg import TrayInfo, HarvestStatus
# from farm_interfaces.action import HarvestTray
# from farm_interfaces.srv import GetZoneLayout
class HarvestingRobotNode(Node):
"""
Nodo ROS2 per robot di raccolta in vertical farm.
Si interfaccia con:
- Sistema SCADA per ricevere job di raccolta
- Controller braccio robotico (MoveIt2)
- Sistema conveyor per deposito vassoi
- Computer vision per verifica maturita
"""
def __init__(self):
super().__init__('harvesting_robot_node')
# Publisher stato robot
self.status_pub = self.create_publisher(
String, '/farm/robot/harvest/status', 10
)
# Subscriber per job di raccolta da SCADA
self.job_sub = self.create_subscription(
String, '/farm/scada/harvest_jobs',
self.on_harvest_job, 10
)
# Subscriber per immagine camera end-effector
self.camera_sub = self.create_subscription(
Image, '/robot/camera/raw',
self.on_camera_frame, 10
)
# Client per servizio layout zona
# self.layout_client = self.create_client(GetZoneLayout, '/farm/zone/layout')
self.current_job: dict = {}
self.is_busy = False
self.get_logger().info('HarvestingRobotNode avviato')
def on_harvest_job(self, msg: String):
"""Riceve job di raccolta dallo SCADA"""
if self.is_busy:
self.get_logger().warn('Robot occupato, job ignorato')
return
try:
job = json.loads(msg.data)
self.get_logger().info(
f"Job ricevuto: zona={job['zone_id']}, "
f"tray={job['tray_id']}, "
f"crop={job['crop_type']}"
)
self.current_job = job
self.is_busy = True
# Avvia sequenza raccolta in thread separato
self.executor.create_task(self.execute_harvest(job))
except (json.JSONDecodeError, KeyError) as e:
self.get_logger().error(f"Job malformato: {e}")
async def execute_harvest(self, job: dict) -> bool:
"""
Sequenza completa raccolta:
1. Naviga verso zona target
2. Verifica maturita con visione artificiale
3. Preleva vassoio con braccio robotico
4. Trasporta a conveyor di uscita
5. Aggiorna SCADA
"""
try:
# Step 1: Navigazione
self.publish_status("NAVIGATING", job)
success = await self.navigate_to_zone(job['zone_id'], job['shelf_row'])
if not success:
self.publish_status("NAV_FAILED", job)
return False
# Step 2: Verifica maturita (computer vision)
maturity_score = await self.check_crop_maturity(job['tray_id'])
if maturity_score < 0.85:
self.get_logger().warn(
f"Vassoio {job['tray_id']}: maturita {maturity_score:.2f} "
f"sotto soglia 0.85, raccolta posticipata"
)
self.publish_status("MATURITY_INSUFFICIENT", job)
self.is_busy = False
return False
# Step 3: Raccolta
self.publish_status("HARVESTING", job)
await self.pick_tray(job['tray_id'], job['shelf_position'])
# Step 4: Deposito su conveyor
self.publish_status("DELIVERING", job)
await self.deliver_to_conveyor(job['destination_line'])
# Step 5: Completamento
self.publish_status("COMPLETED", job)
self.is_busy = False
return True
except Exception as e:
self.get_logger().error(f"Errore harvest job {job.get('tray_id')}: {e}")
self.publish_status("ERROR", job)
self.is_busy = False
return False
def publish_status(self, status: str, job: dict):
msg = String()
msg.data = json.dumps({
"robot_id": self.get_name(),
"status": status,
"tray_id": job.get("tray_id"),
"zone_id": job.get("zone_id"),
"timestamp": self.get_clock().now().to_msg().sec
})
self.status_pub.publish(msg)
async def navigate_to_zone(self, zone_id: str, shelf_row: int) -> bool:
"""Naviga AGV verso la zona target (stub - usa Nav2 in produzione)"""
self.get_logger().info(f"Navigazione verso zona {zone_id} fila {shelf_row}")
await asyncio.sleep(2.0) # simulazione movimento
return True
async def check_crop_maturity(self, tray_id: str) -> float:
"""Analisi visione artificiale per valutazione maturita (stub)"""
# In produzione: inferenza YOLO/custom model su immagine camera
await asyncio.sleep(0.5)
return 0.92 # score maturita 0-1
async def pick_tray(self, tray_id: str, position: dict) -> bool:
"""Controllo braccio robotico per prelievo vassoio via MoveIt2 (stub)"""
await asyncio.sleep(1.5)
return True
async def deliver_to_conveyor(self, destination_line: str) -> bool:
"""Deposita vassoio su conveyor di uscita (stub)"""
await asyncio.sleep(1.0)
return True
def on_camera_frame(self, msg: Image):
"""Callback per frame camera (elaborato async su richiesta)"""
pass
def main(args=None):
rclpy.init(args=args)
node = HarvestingRobotNode()
try:
rclpy.spin(node)
except KeyboardInterrupt:
pass
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()垂直ファームの API 設計: FastAPI REST バックエンド
API レイヤーは、農場の物理制御システムとシステム間の統合ポイントです。 外部世界: 企業 ERP、顧客ポータル、オペレーター モバイル アプリ、WMS システム 物流倉庫。このコンテキストで API が不適切に設計されていると、不整合が発生します。 作物のレシピ、スケジュールミス、潜在的な作物の損失。良いもの それに対して、API はすべてのサブシステムを調整する神経系です。
"""
FastAPI Backend per Gestione Vertical Farm
Endpoints: ricette colturali, zone, batch produzione, attuatori, sensori
"""
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime, date
from enum import Enum
import uuid
app = FastAPI(
title="Vertical Farm Control API",
description="API per gestione vertical farm: ricette, zone, robot, sensori",
version="2.1.0"
)
security = HTTPBearer()
# ============================================================
# MODELLI PYDANTIC
# ============================================================
class CropTypeEnum(str, Enum):
LETTUCE = "lettuce"
BASIL = "basil"
SPINACH = "spinach"
STRAWBERRY = "strawberry"
MICROGREENS = "microgreens"
HERBS = "herbs"
class GrowingSystemEnum(str, Enum):
NFT = "nft"
DWC = "dwc"
AEROPONICS = "aeroponics"
SUBSTRATE = "substrate"
class LightRecipeCreate(BaseModel):
name: str = Field(..., min_length=3, max_length=100)
crop_type: CropTypeEnum
growth_stage: str
ppfd_target: float = Field(..., ge=50, le=800)
dli_target: float = Field(..., ge=5, le=40)
photoperiod_hours: float = Field(..., ge=8, le=24)
spectrum_red_pct: float = Field(..., ge=0, le=100)
spectrum_blue_pct: float = Field(..., ge=0, le=100)
spectrum_white_pct: float = Field(..., ge=0, le=100)
spectrum_farred_pct: float = Field(default=0.0, ge=0, le=20)
spectrum_uv_pct: float = Field(default=0.0, ge=0, le=10)
notes: Optional[str] = None
@validator('spectrum_blue_pct')
def validate_spectrum_sum(cls, v, values):
total = (values.get('spectrum_red_pct', 0) + v +
values.get('spectrum_white_pct', 0))
# Tolleranza +/- 5% per arrotondamenti
if total > 105:
raise ValueError(f"Somma canali spettro {total}% supera 100%")
return v
class BatchCreate(BaseModel):
zone_id: str
recipe_id: str
crop_type: CropTypeEnum
growing_system: GrowingSystemEnum
seeding_date: date
expected_harvest_date: date
tray_count: int = Field(..., ge=1, le=10000)
seeds_per_tray: int = Field(default=50, ge=1, le=500)
client_order_id: Optional[str] = None
@validator('expected_harvest_date')
def harvest_after_seeding(cls, v, values):
seeding = values.get('seeding_date')
if seeding and v <= seeding:
raise ValueError("La data di raccolta deve essere successiva alla semina")
return v
class SensorReading(BaseModel):
zone_id: str
timestamp: datetime
temperature_c: float
humidity_rh: float
co2_ppm: float
ppfd_umol: Optional[float] = None
ec_ms_cm: Optional[float] = None
ph: Optional[float] = None
water_temp_c: Optional[float] = None
class ActuatorCommand(BaseModel):
zone_id: str
command_type: str # "led_update", "co2_valve", "pump_speed", "hvac"
parameters: dict
priority: int = Field(default=5, ge=1, le=10) # 10 = emergenza
# ============================================================
# ROUTES: RICETTE COLTURALI
# ============================================================
@app.post("/api/v1/recipes", status_code=status.HTTP_201_CREATED)
async def create_recipe(
recipe: LightRecipeCreate,
credentials: HTTPAuthorizationCredentials = Depends(security)
):
"""
Crea una nuova ricetta colturale nel sistema.
Le ricette definiscono parametri luminosi, ambientali e irrigazione.
"""
recipe_id = str(uuid.uuid4())
# In produzione: salvataggio su PostgreSQL
return {
"recipe_id": recipe_id,
"name": recipe.name,
"crop_type": recipe.crop_type,
"ppfd_target": recipe.ppfd_target,
"dli_target": recipe.dli_target,
"created_at": datetime.utcnow().isoformat(),
"status": "active"
}
@app.get("/api/v1/recipes/{recipe_id}")
async def get_recipe(recipe_id: str):
"""Recupera ricetta colturale per ID"""
# Stub - in produzione: query PostgreSQL
return {
"recipe_id": recipe_id,
"name": "Lattuga Lollo Rossa - Vegetativo",
"crop_type": "lettuce",
"ppfd_target": 200.0,
"dli_target": 17.0,
"photoperiod_hours": 16.0,
"spectrum": {"red": 65, "blue": 20, "white": 10, "farred": 5},
"env_targets": {"temp_c": 21.0, "humidity_rh": 70.0, "co2_ppm": 1200},
"nutrient_targets": {"ec_ms_cm": 1.6, "ph": 6.0}
}
@app.get("/api/v1/recipes")
async def list_recipes(
crop_type: Optional[CropTypeEnum] = None,
active_only: bool = True,
limit: int = Field(default=50, le=200)
):
"""Lista ricette con filtro per tipo coltura"""
return {
"recipes": [],
"total": 0,
"filters": {"crop_type": crop_type, "active_only": active_only}
}
# ============================================================
# ROUTES: BATCH PRODUZIONE
# ============================================================
@app.post("/api/v1/batches", status_code=status.HTTP_201_CREATED)
async def create_batch(
batch: BatchCreate,
background_tasks: BackgroundTasks,
credentials: HTTPAuthorizationCredentials = Depends(security)
):
"""
Avvia un nuovo batch di produzione.
Associa zona, ricetta, dati di seeding e target harvest.
Background: programma scheduling LED e irrigazione su SCADA.
"""
batch_id = str(uuid.uuid4())
background_tasks.add_task(schedule_batch_on_scada, batch_id, batch)
return {
"batch_id": batch_id,
"zone_id": batch.zone_id,
"crop_type": batch.crop_type,
"seeding_date": batch.seeding_date.isoformat(),
"expected_harvest_date": batch.expected_harvest_date.isoformat(),
"tray_count": batch.tray_count,
"status": "scheduled"
}
async def schedule_batch_on_scada(batch_id: str, batch: BatchCreate):
"""Task background: invia configurazione a SCADA per scheduling"""
# In produzione: chiamata API verso SCADA (Ignition, custom Python SCADA)
pass
# ============================================================
# ROUTES: SENSORI E TELEMETRIA
# ============================================================
@app.post("/api/v1/telemetry")
async def ingest_sensor_data(reading: SensorReading):
"""
Endpoint per ingestion dati sensori da edge controller.
Validazione, allarmi e storicizzazione su InfluxDB/TimescaleDB.
"""
alerts = []
# Allarmi temperatura
if reading.temperature_c > 28.0:
alerts.append({"type": "HIGH_TEMP", "value": reading.temperature_c, "threshold": 28.0})
elif reading.temperature_c < 16.0:
alerts.append({"type": "LOW_TEMP", "value": reading.temperature_c, "threshold": 16.0})
# Allarmi CO2
if reading.co2_ppm > 2000:
alerts.append({"type": "HIGH_CO2", "value": reading.co2_ppm, "threshold": 2000})
# Allarmi pH
if reading.ph is not None and (reading.ph < 5.0 or reading.ph > 7.5):
alerts.append({"type": "PH_OUT_OF_RANGE", "value": reading.ph})
# In produzione: write batch su InfluxDB e publish alert su Kafka/MQTT
return {
"status": "accepted",
"zone_id": reading.zone_id,
"timestamp": reading.timestamp.isoformat(),
"alerts": alerts,
"alert_count": len(alerts)
}
@app.get("/api/v1/zones/{zone_id}/current")
async def get_zone_current_state(zone_id: str):
"""Stato ambientale corrente di una zona (last value da InfluxDB)"""
# Stub
return {
"zone_id": zone_id,
"timestamp": datetime.utcnow().isoformat(),
"sensors": {
"temperature_c": 21.3,
"humidity_rh": 69.8,
"co2_ppm": 1185,
"ppfd_umol": 198.5,
"ec_ms_cm": 1.62,
"ph": 6.05
},
"actuators": {
"led_pwm": {"red": 0.617, "blue": 0.192, "white": 0.098, "farred": 0.049},
"co2_valve_pct": 12.5,
"hvac_cooling_pct": 35.0,
"pump_active": True
},
"active_batch_id": "b-2025-001-lollo",
"days_since_seeding": 18
}
# ============================================================
# ROUTES: CONTROLLO ATTUATORI
# ============================================================
@app.post("/api/v1/actuators/command")
async def send_actuator_command(
command: ActuatorCommand,
credentials: HTTPAuthorizationCredentials = Depends(security)
):
"""
Invia comando manuale a un attuatore di zona.
Usato per override manuali, manutenzione e test.
Richiede autenticazione e viene loggato per audit.
"""
allowed_commands = {"led_update", "co2_valve", "pump_speed", "hvac", "emergency_stop"}
if command.command_type not in allowed_commands:
raise HTTPException(
status_code=400,
detail=f"Tipo comando non valido: {command.command_type}"
)
command_id = str(uuid.uuid4())
# In produzione: publish su MQTT/OPC-UA verso edge controller
return {
"command_id": command_id,
"zone_id": command.zone_id,
"command_type": command.command_type,
"parameters": command.parameters,
"status": "sent",
"sent_at": datetime.utcnow().isoformat()
}ファームデジタルツイン: シミュレーションと最適化
垂直農場のデジタルツインとその動作を再現する計算モデル 予測シミュレーションを可能にするのに十分な精度で農場の物理学を分析します。そうではありません それは視覚的な 3D レプリカ (つまり「視覚化」) ではなく、数学的モデルです 現在の環境パラメータの状態を考慮すると、植物の成長が予測され、 収穫の時期。
制御された環境で最もよく使用される植物成長モデルは、次のアプローチに基づいています。 放射線利用効率（RUE）: 蓄積されたバイオマスは次のように比例します。 遮断光 (PAR) と変換効率は、温度、CO2、 水と栄養の利用可能性。もともとシステム用に開発されたこれらのモデル オープンフィールド（DSSAT、APSIMなど）での収量予測は、環境に合わせて調整されています 実験的に調整されたパラメータを使用した屋内。
"""
Digital Twin - Modello di Crescita Vegetale per Vertical Farm
Basato su Radiation Use Efficiency (RUE) + effetti temperatura/CO2
"""
import numpy as np
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime, timedelta
@dataclass
class PlantGrowthModel:
"""
Modello semplificato di crescita per lattuga in sistema idroponico.
Parametri calibrati su dati sperimentali per Lactuca sativa.
"""
# Parametri biologici della coltura
rue_base: float = 1.8 # g biomassa / MJ PAR intercettato
temp_base: float = 5.0 # temperatura base (°C) - sotto non cresce
temp_opt: float = 22.0 # temperatura ottimale
temp_max: float = 32.0 # temperatura max sopravvivenza
co2_base_ppm: float = 400.0 # CO2 ambient reference
co2_enhancement: float = 0.002 # incremento RUE per ppm CO2 extra
# Stato corrente della pianta
fresh_weight_g: float = 0.5 # peso fresco iniziale (semenzale 5g DW)
dry_weight_g: float = 0.05 # peso secco iniziale
leaf_area_cm2: float = 5.0 # area fogliare iniziale
days_since_seeding: int = 0
# Target harvest
target_fresh_weight_g: float = 150.0 # lattuga da 150g
water_content: float = 0.95 # % acqua rispetto al peso fresco
def temperature_factor(self, temp: float) -> float:
"""
Fattore temperatura (0-1) usando funzione beta.
temp_opt da il massimo rendimento (1.0).
"""
if temp <= self.temp_base or temp >= self.temp_max:
return 0.0
if temp <= self.temp_opt:
return (temp - self.temp_base) / (self.temp_opt - self.temp_base)
else:
return (self.temp_max - temp) / (self.temp_max - self.temp_opt)
def co2_factor(self, co2_ppm: float) -> float:
"""Fattore arricchimento CO2 (1.0 ad ambient, >1 con arricchimento)"""
extra_co2 = max(0, co2_ppm - self.co2_base_ppm)
return 1.0 + (self.co2_enhancement * extra_co2)
def par_intercepted_mj(self, ppfd: float, leaf_area_cm2: float,
photoperiod_h: float) -> float:
"""
Calcola PAR intercettata dalla pianta in MJ/giorno.
ppfd: µmol/m²/s -> conversione a W/m² (1 W/m² ≈ 4.6 µmol/m²/s per LED)
"""
ppfd_wm2 = ppfd / 4.6
par_w = ppfd_wm2 * (leaf_area_cm2 / 10000) # in m²
par_mj_day = par_w * photoperiod_h * 3600 / 1e6
return par_mj_day
def simulate_day(self, temp: float, co2_ppm: float,
ppfd: float, photoperiod_h: float) -> dict:
"""
Simula un giorno di crescita e aggiorna lo stato della pianta.
Restituisce delta giornaliero e stato aggiornato.
"""
# Fattori ambientali
tf = self.temperature_factor(temp)
cf = self.co2_factor(co2_ppm)
par_intercepted = self.par_intercepted_mj(ppfd, self.leaf_area_cm2, photoperiod_h)
# Crescita biomassa secca (RUE model)
delta_dw = self.rue_base * par_intercepted * tf * cf
delta_fw = delta_dw / (1 - self.water_content)
self.dry_weight_g += delta_dw
self.fresh_weight_g += delta_fw
# Aggiornamento area fogliare (SLA - specific leaf area)
sla_cm2_per_g = 350 # cm²/g DW per lattuga
self.leaf_area_cm2 = self.dry_weight_g * sla_cm2_per_g
self.days_since_seeding += 1
# Check harvest readiness
harvest_ready = self.fresh_weight_g >= self.target_fresh_weight_g
return {
"day": self.days_since_seeding,
"fresh_weight_g": round(self.fresh_weight_g, 2),
"dry_weight_g": round(self.dry_weight_g, 3),
"leaf_area_cm2": round(self.leaf_area_cm2, 1),
"delta_fw_g": round(delta_fw, 3),
"temp_factor": round(tf, 3),
"co2_factor": round(cf, 3),
"par_intercepted_mj": round(par_intercepted, 6),
"harvest_ready": harvest_ready,
}
def simulate_full_cycle(self, daily_conditions: List[dict]) -> dict:
"""
Simula l'intero ciclo colturale con condizioni giornaliere variabili.
Restituisce proiezione completa e giorno stimato di raccolta.
"""
days_log = []
harvest_day = None
for day_idx, cond in enumerate(daily_conditions):
day_state = self.simulate_day(
temp=cond.get('temp', 21.0),
co2_ppm=cond.get('co2_ppm', 1200),
ppfd=cond.get('ppfd', 200),
photoperiod_h=cond.get('photoperiod_h', 16)
)
days_log.append(day_state)
if day_state['harvest_ready'] and harvest_day is None:
harvest_day = day_idx + 1
return {
"days_simulated": len(days_log),
"final_fresh_weight_g": self.fresh_weight_g,
"estimated_harvest_day": harvest_day,
"daily_log": days_log,
"achieved_target": self.fresh_weight_g >= self.target_fresh_weight_g
}
# Esempio utilizzo digital twin
def predict_harvest_date(recipe: dict, seeding_date: datetime) -> datetime:
"""
Usa il digital twin per predire la data di raccolta
dato una ricetta ambientale costante.
"""
model = PlantGrowthModel()
# Condizioni giornaliere dalla ricetta (costanti per semplicità)
daily_conditions = [{
'temp': recipe.get('temp_c', 21.0),
'co2_ppm': recipe.get('co2_ppm', 1200),
'ppfd': recipe.get('ppfd_target', 200),
'photoperiod_h': recipe.get('photoperiod_hours', 16)
}] * 40 # massimo 40 giorni di simulazione
result = model.simulate_full_cycle(daily_conditions)
harvest_day = result.get('estimated_harvest_day', 35)
return seeding_date + timedelta(days=harvest_day)最適化のための AI: 明るいレシピのための強化学習
デジタル ツインは、単純な予測よりも強力なものを可能にします。 レシピの最適化 強化学習 (RL) 経由。エージェント RL (実際のファームではなく) デジタル ツインと対話し、何千もの組み合わせを探索します。 ライトパラメータを使用して、消費量を最小限に抑えながら収量を最大化する構成を見つけます。 エネルギッシュ。シミュレーションで最適なレシピが見つかると、それは 1 回で検証されます。 大規模展開前の実際のファームのパイロット エリア。
このアプローチは、次のように述べています。 SIMからリアルへの転送、研究の最前線 AI垂直農業において。シミュレーションと現実の間のギャップ (シミュレーションと現実のギャップ) には、 農場から収集された実際のデータに基づく成長モデルの継続的な校正。
"""
Reinforcement Learning per Ottimizzazione Ricette Luminose
Usa Gymnasium + custom environment basato sul PlantGrowthModel
"""
import gymnasium as gym
import numpy as np
from gymnasium import spaces
from typing import Tuple, Optional
class VerticalFarmEnv(gym.Env):
"""
Ambiente Gymnasium per ottimizzazione ricette luminose.
Observation space: stato ambientale corrente + stato pianta
Action space: aggiustamenti parametri LED (continuo)
Reward: crescita giornaliera / consumo energetico
"""
metadata = {'render_modes': ['human']}
def __init__(self, crop_type: str = "lettuce", episode_days: int = 30):
super().__init__()
self.crop_type = crop_type
self.episode_days = episode_days
self.current_day = 0
# Action space: [delta_ppfd, delta_red_ratio, delta_blue_ratio, delta_photoperiod]
# Valori normalizzati in [-1, 1], scalati internamente
self.action_space = spaces.Box(
low=np.array([-1.0, -1.0, -1.0, -1.0], dtype=np.float32),
high=np.array([1.0, 1.0, 1.0, 1.0], dtype=np.float32)
)
# Observation space: [ppfd, red_ratio, blue_ratio, photoperiod,
# fresh_weight, leaf_area, days, temp, co2]
self.observation_space = spaces.Box(
low=np.array([50, 0, 0, 8, 0, 0, 0, 15, 400], dtype=np.float32),
high=np.array([800, 1, 1, 24, 500, 5000, 45, 30, 2000], dtype=np.float32)
)
# Stato corrente
self.ppfd = 200.0
self.red_ratio = 0.65
self.blue_ratio = 0.20
self.photoperiod = 16.0
self.plant_model = None
def reset(self, seed: Optional[int] = None, **kwargs) -> Tuple[np.ndarray, dict]:
super().reset(seed=seed)
self.current_day = 0
self.ppfd = 200.0
self.red_ratio = 0.65
self.blue_ratio = 0.20
self.photoperiod = 16.0
from digital_twin import PlantGrowthModel # import locale
self.plant_model = PlantGrowthModel()
return self._get_obs(), {}
def step(self, action: np.ndarray) -> Tuple[np.ndarray, float, bool, bool, dict]:
# Applica azione con scaling
self.ppfd = np.clip(self.ppfd + action[0] * 50, 50, 800)
self.red_ratio = np.clip(self.red_ratio + action[1] * 0.1, 0.3, 0.8)
self.blue_ratio = np.clip(self.blue_ratio + action[2] * 0.05, 0.1, 0.35)
self.photoperiod = np.clip(self.photoperiod + action[3] * 1.0, 10, 22)
# Simula giorno con nuove condizioni
day_result = self.plant_model.simulate_day(
temp=21.0, co2_ppm=1200,
ppfd=self.ppfd, photoperiod_h=self.photoperiod
)
# Calcola consumo energetico (kWh/giorno per m² crescita)
energy_kwh = (self.ppfd / 4.6) * (self.photoperiod / 1000) # semplificato
# Reward: crescita / energia (massimizza efficienza)
growth = day_result['delta_fw_g']
reward = growth / max(energy_kwh, 0.001) * 0.01
# Penalita per harvest_ready raggiunto troppo tardi
if self.current_day > 35 and not day_result['harvest_ready']:
reward -= 5.0
# Bonus per harvest_ready raggiunto nei tempi
if day_result['harvest_ready'] and self.current_day <= 28:
reward += 20.0
self.current_day += 1
done = day_result['harvest_ready'] or self.current_day >= self.episode_days
return self._get_obs(), reward, done, False, day_result
def _get_obs(self) -> np.ndarray:
pm = self.plant_model
return np.array([
self.ppfd, self.red_ratio, self.blue_ratio, self.photoperiod,
pm.fresh_weight_g if pm else 0.5,
pm.leaf_area_cm2 if pm else 5.0,
self.current_day, 21.0, 1200.0
], dtype=np.float32)
# Training con Stable-Baselines3
# from stable_baselines3 import PPO
# env = VerticalFarmEnv(crop_type="lettuce")
# model = PPO("MlpPolicy", env, verbose=1, learning_rate=3e-4)
# model.learn(total_timesteps=500_000)
# model.save("optimized_lettuce_recipe_v1")産業用IoTインフラストラクチャ: Modbus、MQTT、OPC-UA
垂直ファームでは、産業用プロトコルのオーバーレイを使用します: Modbus RTU/TCP 従来のセンサーおよびアクチュエーター (温湿度計、CO2 メーター、コントローラー) への通信 LED)、Siemens/Beckhoff PLC および SCADA システムとの通信用の OPC-UA、MQTT 用 データをクラウドに送信します。選択はベンダーのハードウェア、遅延によって異なります。 必要とセキュリティのレベル。
垂直農法における IoT プロトコル: 比較
| プロトコル | レイヤー | レイテンシ | 安全性 | 典型的な使用例 |
|---|---|---|---|---|
| Modbus RTU | フィールドPLC | 10～100ミリ秒 | 不在 (レガシー) | EC/pHセンサー、LEDドライバー |
| Modbus TCP | PLC-SCADA | 5～50ミリ秒 | オプションのTLS | PLCデータ取得 |
| OPC-UA | PLC-SCADA-クラウド | 1～50ミリ秒 | X.509、署名、暗号化 | インダストリー4.0規格 |
| MQTT | エッジクラウド | 10～500ミリ秒 | TLS + 認証 | クラウドへのテレメトリ |
| REST/HTTP | クラウドクラウド | 50～500ミリ秒 | HTTPS、OAuth2 | ERP統合API |
"""
Bridge Modbus -> MQTT per vertical farm
Legge sensori via Modbus RTU e pubblica su MQTT broker
"""
import asyncio
import json
import time
import logging
from pymodbus.client import AsyncModbusSerialClient
import paho.mqtt.client as mqtt
logger = logging.getLogger(__name__)
# Mappa registri Modbus per sensore combo Temp/RH/CO2 (esempio Vaisala HMP60)
MODBUS_REGISTER_MAP = {
"temperature": {"address": 0x0000, "count": 1, "scale": 0.1, "unit": "°C"},
"humidity": {"address": 0x0001, "count": 1, "scale": 0.1, "unit": "%RH"},
"co2_ppm": {"address": 0x0002, "count": 1, "scale": 1.0, "unit": "ppm"},
"ec_ms_cm": {"address": 0x0010, "count": 1, "scale": 0.01, "unit": "mS/cm"},
"ph": {"address": 0x0011, "count": 1, "scale": 0.01, "unit": "pH"},
}
class ModbusMQTTBridge:
def __init__(self, zone_id: str, modbus_port: str,
modbus_address: int, mqtt_broker: str, mqtt_port: int = 1883):
self.zone_id = zone_id
self.modbus_client = AsyncModbusSerialClient(
port=modbus_port, baudrate=9600, timeout=3
)
self.mqtt_client = mqtt.Client(client_id=f"bridge-{zone_id}")
self.mqtt_broker = mqtt_broker
self.mqtt_port = mqtt_port
self.mqtt_topic = f"farm/zones/{zone_id}/telemetry"
async def connect(self):
await self.modbus_client.connect()
self.mqtt_client.connect(self.mqtt_broker, self.mqtt_port, keepalive=60)
self.mqtt_client.loop_start()
logger.info(f"Bridge avviato per zona {self.zone_id}")
async def read_all_sensors(self, device_id: int = 1) -> dict:
readings = {"zone_id": self.zone_id, "timestamp": time.time()}
for sensor_name, reg in MODBUS_REGISTER_MAP.items():
try:
result = await self.modbus_client.read_holding_registers(
address=reg["address"],
count=reg["count"],
slave=device_id
)
if not result.isError():
raw_value = result.registers[0]
readings[sensor_name] = round(raw_value * reg["scale"], 3)
else:
logger.warning(f"Errore lettura {sensor_name} zona {self.zone_id}")
readings[sensor_name] = None
except Exception as e:
logger.error(f"Eccezione Modbus {sensor_name}: {e}")
readings[sensor_name] = None
return readings
async def publish_loop(self, interval_sec: float = 30.0):
while True:
readings = await self.read_all_sensors()
payload = json.dumps(readings)
result = self.mqtt_client.publish(
topic=self.mqtt_topic,
payload=payload,
qos=1
)
if result.rc == mqtt.MQTT_ERR_SUCCESS:
logger.debug(f"Pubblicato su {self.mqtt_topic}: {payload[:80]}...")
else:
logger.error(f"Errore publish MQTT: rc={result.rc}")
await asyncio.sleep(interval_sec)垂直農業の経済学: CAPEX、OPEX、損益分岐点
垂直農業は経済的に持続可能ですか? 2025 年の答えは次のとおりです。 場合によって異なります。 それは作物によって異なります（ハーブやマイクログリーンはレタスよりもはるかに収益性が高くなります）。 規模（5,000㎡以上ではスケールメリットが現れます）、立地（コスト） 地域のエネルギーコストと人件費が重要）、販売チャネル別（直接販売） プレミアム価格の B2C と GDO コモディティの比較）。
経済分析: 農場 1,000 平方メートルの純栽培
| Voce | 価値 | 注意事項 |
|---|---|---|
| CAPEX（初期投資） | ||
| 構造とシステム | 80万ユーロ | 倉庫改修 |
| NFTの棚とチャネル | 300,000ユーロ | 水耕栽培システム |
| LED照明 | 60万ユーロ | 600W/m² 効率 2.8 μmol/J |
| 冷暖房空調設備と気候 | 250,000ユーロ | 冷房+除湿 |
| CO2システム | 50,000ユーロ | ストレージ + 配信 |
| オートメーションとロボット工学 | 40万ユーロ | 播種機、移植機、収穫機 |
| ソフトウェアと統合 | 150,000ユーロ | SCADA、API、デジタルツイン |
| 総設備投資額 | 2,550,000ユーロ | ～2,550ユーロ/㎡ |
| 年間運営費 | ||
| 電気 | 420,000ユーロ | 35% OPEX - 主要な重要エネルギー |
| 作業（オペレーター10名） | 350,000ユーロ | 27% の運用コスト |
| 種子と基質 | 80,000ユーロ | 6% の運用コスト |
| 栄養素とCO2 | 60,000ユーロ | 5% 運用コスト |
| メンテナンス | 90,000ユーロ | 運用コスト 7% |
| 梱包と物流 | 120,000ユーロ | 9%の運用コスト |
| その他（保険等） | 80,000ユーロ | 6% の運用コスト |
| 総運用コスト | 1,200,000ユーロ | €1,200/㎡/年 |
| 収益 | ||
| レタス生産 (18 サイクル x 25kg/m²) | 450kg/m²/年 | 合計450,000kg |
| プレミアムGDO販売価格 | 3.5ユーロ/kg | vs屋外 0.8～1.2ユーロ |
| 総収益 | 1,575,000ユーロ | |
| EBITDA | 375,000ユーロ | マージン 23.8% |
| 設備投資の償却（10年） | 255,000ユーロ | |
| 税引前純利益 | 120,000ユーロ | マージン7.6% |
| 損益分岐点 (年) | 約8～10年 | ハーブを使用した場合：4～5年 |
垂直農法におけるエネルギー問題
垂直農業のエネルギーと実存に関わる挑戦。 2025 年に入手可能な最も効率的な LED 彼らは約に達します 3.0～3.5μmol/J 光子効率のこと。を生産するには 16 時間の光周期で 17 mol/m²/日の DLI には、約 280 Wh/m²/日が必要です。 102kWh/㎡/年 照明のためだけに。面積1,000㎡、費用は エネルギーが 0.15 ユーロ/kWh (イタリアの産業関税 2025 年) の場合、LED の請求額はすでに 153,000 ユーロ/年です。 HVAC (通常 LED エネルギーの 60 ～ 70%)、CO2、ポンプ、自動化を追加すると、効果が得られます。 簡単に年間42万ユーロです。低コストの再生可能エネルギーを利用できる人々（農業など） 100% 再生可能エネルギーを備えた最新の住宅）、または太陽光発電が大部分をカバーしている ニーズの一部ではありますが、すべてではありません。
垂直農法 どのエネルギー価格でも持続可能ではない。エネルギーを持って 0.25 ユーロ/kWh 以上 (2022 年から 2023 年などの危機期に起こり得るシナリオ)、多くの経済モデル 彼らは崩壊します。 LED の効率は向上し続け、再生可能エネルギーは向上すると考えられます。 コストがどんどん下がっていきます。
ケーススタディ: プラネット ファームと現代農業 - イタリア モデル
イタリアにはヨーロッパで最も先進的な垂直農業プロジェクトが 2 つあり、どちらも ミラノの内陸部に拠点を置く。それらのパスは異なりますが、補完的であり、 ヨーロッパの垂直農業における拡張性と収益性の問題に対する 2 つのアプローチ。
プラネット ファームズ - カヴェナーゴ ディ ブリアンツァ
2018 年にルカ トラヴァリーニとマッシミリアーノ ロスキによって設立されたプラネット ファームズは、 カベナゴ（MB）の旧工業地帯に最初の工場を建設。元の植物は、 9,000 平方メートルで、2024 年にはヨーロッパ最大級の規模となりました。 20,000㎡ 成長面。 2023 年 11 月、Planet Farms はラウンドを調達しました 評価額は5億ドルで4,000万ドルで、この分野ではヨーロッパ最大規模の1つです。 2025年にスイス・ライフ・アセット・マネージャーズと提携し、 200ユーロ 何百万もの ヨーロッパ全土で大規模な垂直農場を開発する。
シーメンスとの技術提携はオートメーションの核心です: 制御システム 産業用 Siemens S7-1500 は環境ループを管理し、プラットフォームは Mindsphere (現在は Siemens Industrial Copilot) はデータを収集して分析します。製品 代表的な商品は「リビングハーブ」。 根菜、レタスよりも利益率の高いプレミアムセグメント。
現代農業 - アグナデッロ (CR)
アグリコラ モデルナは、ピエルイジ ジュリアーニとベンジャミン フランケッティによって 2018 年にミラノで設立されました。 2024年9月に新工場がオープン 11,000㎡ 広告 アグナデッロ（クレモナ）はインテサ・サンパオロから1000万ユーロの融資を受けた。 植物が生産するのは、 1日あたりサラダ30,000袋、電源付き 100% 再生可能資源から使用し、R&D チームが開発した内部 AI を使用しています。 レシピを最適化します。
アグリコラ モデルナの技術的差別化とハイパースペクトル イメージング システム (Specim との提携) 植物の栄養状態を評価できるようになります 症状が人間の目に見える前に。再構成用のRGB-Dカメラ キャノピーの 3D、数千の測定点を備えた環境センサー、およびアルゴリズム 社内で開発されたコンピュータ ビジョン システムが独自のテクノロジー スタックを構成しています。
プラネットファームとモダンアグリコラの比較
| パラメータ | プラネット ファームズ | 現代農業 |
|---|---|---|
| 表面 | 20,000㎡ | 11,000㎡ |
| 生産量/日 | N.D.（ハーブフォーカス） | サラダバッグ30,000個 |
| 資金調達 | 4,000万ドルのラウンド + 2億ユーロのJV | 1,000万ユーロ インテーサ・サンパオロ |
| PLC/オートメーション | シーメンス S7-1500 | ベッコフ+カスタム |
| AIプラットフォーム | シーメンスの産業用副操縦士 | 自社開発 |
| コンピュータビジョン | RGB規格 | ハイパースペクトル (スペシム) |
| エネルギー | 部分的に再生可能 | 100%再生可能 |
| ターゲット市場 | プレミアム リテール GDO + 英国での拡大 | イタリアの大規模小売業（IV範囲） |
この分野の課題、限界、現実
垂直農業に関する正直な記事には、利点だけでなく実際の課題も含める必要があります。 このセクターは2022年から2024年にかけて倒産が相次ぎ、幻滅の大きな局面を経験した 数十億ドルの投資を費やしたセンセーショナルなもの。理由を理解することが基本です 持続可能なシステムを構築するために。
垂直農法の本当の課題
- エネルギーコスト: そして第一の殺人者。垂直農場経営者の 60% 主に電気代がかかるため、まだ利益が出ていません。エネルギーがなければ 低コストの再生可能エネルギーであるため、この経済モデルは高級作物にのみ適用されます。
- 限られた作物の品種: 垂直農法は次のような用途に最適です。 葉物野菜（レタス、ほうれん草）、香りのよいハーブ、マイクログリーン。トマトやピーマンなどには、 キュウリのエネルギー収量はまだ競争力がありません。穀類、豆類、根菜類は、 技術的にも経済的にも手の届かないところにある。
- 高い設備投資: 専門的な設置には 2,000 ～ 4,000 ユーロの費用がかかります 栽培面積1平方メートルあたり。レタスの損益分岐点は 8 ～ 10 年です。多くの投資家にとっては多すぎる。 ハーブとマイクログリーンだけが 4 ～ 5 年で損益分岐点に達します。
- LED サプライチェーンの依存関係: 高品質のLEDチップは、 いくつかのサプライヤー (Epistar、Lumileds、Osram) によって製造されています。サプライチェーンの混乱 初期設備投資とメンテナンスの両方に影響します。
- 現実の持続可能性と認識されている持続可能性: 節水（95％）は本物です。 しかし、二酸化炭素排出量はエネルギー構成に大きく依存します。動力付き垂直農場 石炭の二酸化炭素排出量は屋外農業よりも悪い。再生可能エネルギーのみで 残高は明らかにプラスになります。
- 技術的な拡張性: 500 平方メートルのパイロット農場から 20,000㎡の商業スペースは単純な掛け算ではありません。制御システム、 作物管理、社内物流、ロボットシステムが必要 スケールがジャンプするたびに大幅な再設計が行われます。
本番環境への導入: コンテナ化と監視
垂直ファームのソフトウェア コンポーネントは、リアルタイム PID コントローラー ( PLC または Raspberry Pi 上で直接 (コンテナ化されていない)、FastAPI バックエンドと AI パイプライン (Docker コンテナー内で快適に動作)。一般的なクラウド インフラストラクチャ オンプレミスのエッジ レイヤーとクラウド コントロール プレーンを組み合わせます。
# docker-compose.yml per stack vertical farm (dev/staging)
version: '3.9'
services:
# API principale
farm-api:
build: ./farm-api
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://farm:farm@postgres:5432/farmdb
- MQTT_BROKER=emqx
- INFLUXDB_URL=http://influxdb:8086
- REDIS_URL=redis://redis:6379
depends_on:
- postgres
- influxdb
- emqx
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
# MQTT Broker
emqx:
image: emqx/emqx:5.8
ports:
- "1883:1883" # MQTT
- "8883:8883" # MQTT TLS
- "18083:18083" # Dashboard
environment:
- EMQX_NODE__COOKIE=farm-secret-cookie
volumes:
- emqx_data:/opt/emqx/data
# Time-series database per dati sensori
influxdb:
image: influxdb:2.7
ports:
- "8086:8086"
environment:
- DOCKER_INFLUXDB_INIT_MODE=setup
- DOCKER_INFLUXDB_INIT_USERNAME=admin
- DOCKER_INFLUXDB_INIT_PASSWORD=farmpass123
- DOCKER_INFLUXDB_INIT_ORG=verticalfarm
- DOCKER_INFLUXDB_INIT_BUCKET=sensors
volumes:
- influxdb_data:/var/lib/influxdb2
# PostgreSQL per dati applicativi (ricette, batch, inventory)
postgres:
image: postgres:16-alpine
environment:
- POSTGRES_DB=farmdb
- POSTGRES_USER=farm
- POSTGRES_PASSWORD=farm
volumes:
- postgres_data:/var/lib/postgresql/data
# Redis per cache e job queue
redis:
image: redis:7-alpine
command: redis-server --appendonly yes
# Grafana per dashboard real-time
grafana:
image: grafana/grafana:11.0.0
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- grafana_data:/var/lib/grafana
depends_on:
- influxdb
# Worker ML per digital twin e ottimizzazione
ml-worker:
build: ./ml-worker
environment:
- INFLUXDB_URL=http://influxdb:8086
- REDIS_URL=redis://redis:6379
- MODEL_PATH=/models
volumes:
- ml_models:/models
volumes:
emqx_data:
influxdb_data:
postgres_data:
grafana_data:
ml_models:垂直農法におけるトレンドとイノベーション 2025 ～ 2026 年
この分野は、エネルギー効率、電力供給の拡大という 3 つの面で急速に進化しています。 作物、そしてさらに深い AI 統合。最も重要な傾向は次のとおりです。
2025 年の垂直農法における主要なイノベーション
| 革新 | インパクト | 商業的地位 |
|---|---|---|
| LED効率4.0+μmol/J | -20～25% 照明エネルギーコスト | あり（LG、シグニファイ） |
| ハイパースペクトルイメージング | 植物ストレスの早期診断、最適化 | 早期採用 (Specim + Agricola M.) |
| 動的レシピの RL | 同じエネルギーで +15 ～ 25% の収量 | 高度な検索/初期製品化 |
| 高速播種ロボット | 1 時間あたり 3,000 回以上のシード処理と手動の場合の 700 回以上のシード処理 | 商業用 (広い、80 エーカー) |
| きのこ・きのこ栽培 | 作物構成の多様化、高い収益性 | 商業的成長 |
| データセンターの廃熱としての垂直型ファーム | HVAC削減のための熱回収 | パイロット (フィンランド、ドイツ) |
| BESSとの太陽光発電の統合 | 最適なシナリオでのエネルギーコストの -40 ～ 60% | 成長する |
特に興味深い新たなトレンドは、垂直農業と農業の統合です。 データセンター: サーバーは廃熱を生成しますが、これを回収して熱量を削減することができます。 寒い季節の温室の暖房費。エネルギーのあるフィンランドで 費用がかかるため、いくつかのパイロット プロジェクトがこの共生をテストしています。廃熱 40～60℃で冬の温室での栽培温度の維持に最適 追加のエネルギー消費なしで。
結論: ソフトウェアエンジニアリングプロジェクトとしての垂直農業
最新の垂直農場、そして何よりもまずソフトウェア システム。ハードウェア (LED、センサー、 ロボット、棚など）は必要ですが十分ではありません。変換するのはソフトウェア レベルです。 最適化された食品工場の照明付き倉庫。 PIDコントローラーが管理 科学的に調整されたレシピに従った環境。 REST API がファームを接続する 商業エコシステムへ。 ROS2 はロボットを調整します。デジタルツインによりシミュレーションが可能 予測的;強化学習は、設計されたレシピよりも優れたレシピを見つけます 農学者によって手作業で。
2025 年の 96 億から 2033 年の 390 億までの垂直農業市場は、 自動的に保証されます: 問題を解決する業界の能力に依存します エネルギーを増やし、経済的に生産できる作物の範囲を拡大する 持続可能な。プラネット・ファームズとアグリコラ・モデルナを擁するイタリアは正しい立場にある 特にPNRRとインセンティブの後は、この移行の主役になること 5.0 への移行により、技術的な設備投資のコストが削減されます。
この分野への参入を希望する開発者にとって、最も求められるスキルは次のとおりです。 自動化と機械学習には Python、産業用バックエンドには FastAPI、ロボット工学には ROS2、 産業用プロトコル用の MQTT/Modbus/OPC-UA、時系列用の InfluxDB、およびその基本 最適化すべきパラメータを理解するための植物生理学。方法を知る必要はありません ガーデニング、ただし特定の波長で植物がより成長する理由を理解する それが、優れた制御システムと優れた制御システムの違いを生み出します。
詳細を知るためのツールとリソース
- ピモドバス - Modbus RTU/TCP 用の Python ライブラリ、PLC およびセンサーとの通信に最適
- パホ-MQTT - ブローカーと統合するための MQTT Python クライアント (EMQX、HiveMQ、Mosquitto)
- ROS2 ハンブル / ジャジー - ロボティクス フレームワーク、docs.ros.org の公式ドキュメント
- ファストAPI - Python 非同期バックエンド、産業用制御 API に最適
- InfluxDB 2.x - センサー分析用の Flux クエリ言語を使用した時系列データベース
- 安定したベースライン3 - レシピ最適化のための RL 実装 (PPO、SAC、TD3)
- ジム - RL 環境の標準、デジタル ツイン環境で使用
- 点火SCADA - 手頃な価格の OEM ライセンスを取得した産業用 SCADA プラットフォーム
- 垂直農場の毎日 - 垂直分野に関する主なニュース源
- USDA CEA ガイドライン - 環境制御農業の基準パラメータ
シリーズの次回: 食品小売の需要予測
第 8 条では、製品の需要をどのように予測するかという逆の問題を扱います。 生産注文（垂直農場からも）を最適化し、食品を削減する 無駄。季節性とトレンドには Prophet by Meta を、パターンには LightGBM を使用します。 高度な表形式、および機能を備えた完全な予測パイプラインを構築します。 イタリアの食品小売りに特化したエンジニアリング (天候の影響、休日、プロモーション)。
続けて: Prophet と LightGBM を使用した食品小売の需要予測
他のシリーズの関連記事
- IoT パイプライン シリーズ: 精密農業向けの IoT パイプライン - 屋外センサー監視用の MQTT、InfluxDB、Grafana
- MLOps シリーズ: 本番環境での AI モデルの MLOps - 導入方法 RL レシピ最適化モデルは実稼働中です
- コンピュータービジョンシリーズ: 品質管理の履歴書 - テクニック 作物の成熟度と品質を評価するための人工視覚
- データ＆AIビジネスシリーズ： 製造業における AI - 予測 メンテナンスとデジタルツイン、垂直農業に適用可能なツインアーキテクチャ