64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
import importlib.util
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
from collections import namedtuple
|
|
from datetime import date, timedelta
|
|
from unittest import skipUnless
|
|
|
|
from django.test import TestCase
|
|
|
|
from crop_simulation.services import CropSimulationService, PcseSimulationManager
|
|
|
|
|
|
@skipUnless(
|
|
importlib.util.find_spec("pcse") is not None,
|
|
"pcse must be installed to run the real WOFOST test.",
|
|
)
|
|
class CropSimulationSingleRunTest(TestCase):
|
|
def test_single_simulation_prints_response(self):
|
|
os.environ["HOME"] = tempfile.mkdtemp(prefix="pcse-home-", dir="/tmp")
|
|
from pcse import settings as pcse_settings
|
|
from pcse.tests.db_input import (
|
|
AgroManagementDataProvider,
|
|
GridWeatherDataProvider,
|
|
fetch_cropdata,
|
|
fetch_sitedata,
|
|
fetch_soildata,
|
|
)
|
|
|
|
def namedtuple_factory(cursor, row):
|
|
fields = [column[0] for column in cursor.description]
|
|
cls = namedtuple("Row", fields)
|
|
return cls._make(row)
|
|
|
|
db_path = os.path.join(pcse_settings.PCSE_USER_HOME, "pcse.db")
|
|
connection = sqlite3.connect(db_path)
|
|
connection.row_factory = namedtuple_factory
|
|
|
|
grid = int(os.environ.get("PCSE_TEST_GRID", "31031"))
|
|
crop_no = int(os.environ.get("PCSE_TEST_CROP_NO", "1"))
|
|
year = int(os.environ.get("PCSE_TEST_YEAR", "2000"))
|
|
|
|
weather = GridWeatherDataProvider(connection, grid_no=grid).export()
|
|
soil = fetch_soildata(connection, grid)
|
|
site = fetch_sitedata(connection, grid, year)
|
|
crop_parameters = fetch_cropdata(connection, grid, year, crop_no)
|
|
agromanagement = AgroManagementDataProvider(connection, grid, crop_no, year)
|
|
|
|
response = CropSimulationService(
|
|
manager=PcseSimulationManager(model_name="Wofost72_WLP_CWB")
|
|
).run_single_simulation(
|
|
weather=weather,
|
|
soil=soil,
|
|
crop_parameters=crop_parameters,
|
|
agromanagement=agromanagement,
|
|
site_parameters=site,
|
|
name="single real wofost run",
|
|
)
|
|
|
|
connection.close()
|
|
print("\nCrop Simulation Response:\n", response)
|
|
self.assertEqual(response["result"]["engine"], "pcse")
|
|
self.assertIn("yield_estimate", response["result"]["metrics"])
|