V1_0.java [src/java/m/crp/smassessmenttool] Revision: default Date:
/*
* $Id$
*
* This file is part of the Cloud Services Integration Platform (CSIP),
* a Model-as-a-Service framework, API, and application suite.
*
* 2012-2019, OMSLab, Colorado State University.
*
* OMSLab licenses this file to you under the MIT license.
* See the LICENSE file in the project root for more information.
*/
package m.crp.smassessmenttool;
import crp.utils.Climate;
import crp.utils.DBResources;
import static crp.utils.DBResources.LOCAL_SQLSERVER;
import crp.utils.DEMSteepness;
import crp.utils.Region;
import crp.utils.SMWEPP;
import crp.utils.SMWEPS;
import crp.utils.ServiceCall;
import crp.utils.ServiceCallData;
import crp.utils.SoilResult;
import crp.utils.StateCounty;
import crp.utils.WEPPSoilInput;
import crp.utils.WEPSFallowRotation;
import crp.utils.WEPSSoilInput;
import crp.utils.WWESoilParams;
import crp.utils.WindGen;
import csip.Config;
import csip.ModelDataService;
import csip.PayloadParameter;
import csip.PayloadResults;
import csip.ServiceException;
import csip.annotations.Description;
import csip.annotations.Name;
import csip.annotations.Polling;
import csip.annotations.Resource;
import csip.utils.JSONUtils;
import data.interpretors.CligenData;
import data.interpretors.IFCFile;
import data.interpretors.SOLFile;
import data.interpretors.SlopeSteepness;
import static data.interpretors.SlopeSteepness.SLOPE_VERSION_4;
import data.interpretors.SlopeSteepness.SlopeData;
import data.interpretors.WindGenData;
import gisobjects.GISObject;
import gisobjects.GISObjectException;
import gisobjects.GISObjectFactory;
import gisobjects.db.GISEngine;
import gisobjects.db.GISEngineFactory;
import gisobjects.vector.GIS_FeatureCollection;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.ws.rs.Path;
import jersey.repackaged.com.google.common.collect.ImmutableMap;
import data.models.WEPPMetaData;
import data.models.WEPSMetaData;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import soils.utils.EvalResult;
/**
*
* @author <a href="mailto:shaun.case@colostate.edu">Shaun Case</a>
*/
@Name("SM CRP Assessment")
@Description("This service consumes a JSON request containing a request identifier "
+ "and CRP Offer geometry and returns sheet/rill water and wind erosion rates "
+ "for a fallow management for up to the three most dominant soil components "
+ "in a CRP Offer area, plus a weighted average for each erosion rate, as "
+ "computed by the WEPP and WEPS surrogate ANN models.")
@Path("m/smcrpassessment/1.0")
@Polling(first = 2000, next = 1000)
@Resource(from = DBResources.class)
public class V1_0 extends ModelDataService {
public static final String SCENARIO_ID = "scenario_id";
public static final String SCENARIO_GEOMETRY = "scenario_geometry";
public static final String USE_DEM_SLOPE_SERVICE = "use_dem_slope";
public static final String SHOW_DEBUG_OUTPUT = "show_debug_output";
public static final String COMPARE_SLOPE_METHODS = "compare_slope_methods";
protected static final int SLEEP_TIMER = 1000; //milliseconds to sleep while polling async model calls
protected static final int INITIAL_POLL_WAIT_TIMER = 1000; //milliseconds to sleep while polling async model calls
protected static final int GET_CALL_WAIT = 150; // milliseconds to sleep between internal get requests in loop of testing status of all calls at once.
protected static final Map<String, String> WEPS_ANNS = ImmutableMap.of("wa_grant", "wa_grant_20200210", "ms_yazoo", "ms_yazoo_20200210");
protected static final Map<String, String> WEPP_ANNS = ImmutableMap.of("wa_grant", "wa_grant_20200123", "ms_yazoo", "ms_yazoo_20200123");
protected static synchronized void getSoilSlopeData(ServiceCallData sData) {
DEMSteepness slopeCall = (DEMSteepness) sData.serviceCall();
SoilResult soilResult = (SoilResult) ((ParallelCallData) sData.data()).callData();
soilResult.slopes = slopeCall.slopes();
}
protected static synchronized void getRegionData(ServiceCallData sData) {
Region regionCall = (Region) sData.serviceCall();
V1_0 callInstance = ((ParallelCallData) sData.data()).instance();
callInstance.regionLength = regionCall.getLength();
callInstance.regionWidth = regionCall.getWidth();
callInstance.regionOrientation = regionCall.getOrientation();
}
protected static synchronized void getWindData(ServiceCallData sData) {
WindGen windCall = (WindGen) sData.serviceCall();
V1_0 callInstance = ((ParallelCallData) sData.data()).instance();
callInstance.windgenData = new WindGenData(windCall.data(), callInstance.wepsDaysWithPrecipAndWind);
}
protected static synchronized void getWeppClimateData(ServiceCallData sData) throws ServiceException {
Climate cliCall = (Climate) sData.serviceCall();
V1_0 callInstance = ((ParallelCallData) sData.data()).instance();
try {
callInstance.weppCligenData = new CligenData(cliCall.data(), callInstance.weppDaysWithPrecipAndWind);
} catch (IOException ex) {
//System.err.println("Throwing exception inside getWeppClimateData: " + ex.getMessage());
throw new ServiceException("Cannot get climate call results: " + ex.getMessage(), ex);
}
}
protected static synchronized void getWepsClimateData(ServiceCallData sData) throws ServiceException {
Climate cliCall = (Climate) sData.serviceCall();
V1_0 callInstance = ((ParallelCallData) sData.data()).instance();
try {
callInstance.wepsCligenData = new CligenData(cliCall.data(), callInstance.wepsDaysWithPrecipAndWind);
} catch (IOException ex) {
//System.err.println("Throwing exception inside getWepsClimateData: " + ex.getMessage());
throw new ServiceException("Cannot get climate call results: " + ex.getMessage(), ex);
}
}
protected static synchronized void getWEPSSoilData(ServiceCallData sData) {
WEPSSoilInput soilCall = (WEPSSoilInput) sData.serviceCall();
SoilResult soilResult = (SoilResult) ((ParallelCallData) sData.data()).callData();
soilResult.ifcFile = soilCall.getIFCFileData();
}
protected static synchronized void getWEPPSoilData(ServiceCallData sData) {
WEPPSoilInput soilCall = (WEPPSoilInput) sData.serviceCall();
SoilResult soilResult = (SoilResult) ((ParallelCallData) sData.data()).callData();
soilResult.solFile = soilCall.getSOLFileData();
}
protected static synchronized void getSMWEPSData(ServiceCallData sData) {
SMWEPS smWepsCall = (SMWEPS) sData.serviceCall();
SoilResult soilResult = (SoilResult) ((ParallelCallData) sData.data()).callData();
soilResult.windErosion = smWepsCall.windErosion();
}
protected static synchronized void getSMWEPPData(ServiceCallData sData) {
SMWEPP smWeppCall = (SMWEPP) sData.serviceCall();
SoilResult soilResult = (SoilResult) ((ParallelCallData) sData.data()).callData();
soilResult.waterErosion = smWeppCall.waterErosion();
}
protected static synchronized void getSoilsData(ServiceCallData sData) throws ServiceException {
WWESoilParams wweCall = (WWESoilParams) sData.serviceCall();
V1_0 callInstance = ((ParallelCallData) sData.data()).instance();
try {
callInstance.topThreeComponents = wweCall.getTopThree();
} catch (JSONException ex) {
//System.err.println("Throwing exception in getSoilsData: " + ex.getMessage());
throw new ServiceException("Cannot get soils call results: " + ex.getMessage(), ex);
}
}
protected JSONObject scenario_geometry;
protected String scenario_id;
protected GISObject aoa_geometry;
protected JSONObject dem_featureCollection;
protected GISEngine gEngine;
protected String regionURI = "http://csip.engr.colostate.edu:8083/csip-weps/m/region/2.0";
protected String soilsURI = "http://csip.engr.colostate.edu:8083/csip-soils/d/wwesoilparams/2.1";
protected String climateURI = "http://csip.engr.colostate.edu:8083/csip-climate/m/cligen_prism/2.0";
protected String windURI = "http://csip.engr.colostate.edu:8083/csip-climate/m/windgen/2.0";
protected String demSteepnessURI = "http://csip.engr.colostate.edu:8087/csip-watershed/m/average_slope/4.0";
protected String weppSoilInputURI = "http://csip.engr.colostate.edu:8083/csip-soils/d/weppsoilinput/2.0";
protected String wepsSoilInputURI = "http://csip.engr.colostate.edu:8083/csip-soils/d/wepssoilinput/3.0";
protected String weppURI = "http://csip.engr.colostate.edu:8083/csip-crp/m/smwepp/1.0";
protected String wepsURI = "http://csip.engr.colostate.edu:8083/csip-crp/m/smweps/1.0";
protected String stateCountyURI = "http://csip.engr.colostate.edu:8083/csip-misc/d/stateCounty/2.0";
protected String outputFile;
protected boolean streamOutputFile = false;
protected volatile double regionLength, regionWidth, regionOrientation;
protected ArrayList<SoilResult> topThreeComponents;
protected ArrayList<SoilResult> alternateSlopeTopThreeComponents;
protected double latitude, longitude;
protected JSONObject rotation;
protected volatile double waterErosionRate = 0.0;
protected volatile double windErosionRate = 0.0;
protected SlopeSteepness slopes;
protected boolean useDemSlope = true;
protected boolean debugOutput = false;
protected boolean compareSlopes = true;
protected volatile int[][][] wepsDaysWithPrecipAndWind = new int[100][12][31];
protected volatile int[][][] weppDaysWithPrecipAndWind = new int[100][12][31];
protected volatile CligenData weppCligenData;
protected volatile CligenData wepsCligenData;
protected volatile WindGenData windgenData;
protected String countyCode, county, stateAbbrev;
protected ArrayList<AsyncCalls> weppCalls = new ArrayList<>();
protected ArrayList<AsyncCalls> wepsCalls = new ArrayList<>();
protected ArrayList<AsyncCalls> asyncCalls = new ArrayList<>();
protected volatile int wepsDayCountPrecipAndWind = 0;
protected volatile int wepsYearsSimulated;
protected String weps_ann_name;
protected String wepp_ann_name;
@Override
protected void preProcess() throws Exception {
PayloadParameter params = parameter();
soilsURI = Config.getString("crp.soils", soilsURI);
regionURI = Config.getString("crp.region", regionURI);
weppURI = Config.getString("crp.sm.wepp", weppURI);
wepsURI = Config.getString("crp.sm.weps", wepsURI);
weppSoilInputURI = Config.getString("crp.wepp.soilinput", weppSoilInputURI);
wepsSoilInputURI = Config.getString("crp.weps.soilinput", wepsSoilInputURI);
climateURI = Config.getString("crp.cligen", climateURI);
windURI = Config.getString("crp.windgen", windURI);
stateCountyURI = Config.getString("crp.statecounty", stateCountyURI);
demSteepnessURI = Config.getString("crp.dem.steepness", demSteepnessURI);
// Keep in this order.
compareSlopes = params.getBoolean(COMPARE_SLOPE_METHODS, false);
debugOutput = params.getBoolean(SHOW_DEBUG_OUTPUT, false) | compareSlopes;
if (params.has(SCENARIO_ID)) {
scenario_id = params.getString(SCENARIO_ID);
} else {
throw new ServiceException("A required input 'scenario_id' was missing. Please specify a scenario_id value.");
}
if (params.has(USE_DEM_SLOPE_SERVICE)) {
useDemSlope = params.getBoolean(USE_DEM_SLOPE_SERVICE);
}
if (params.has(SCENARIO_GEOMETRY)) {
scenario_geometry = params.getParamJSON(SCENARIO_GEOMETRY);
try (Connection connection = resources().getJDBC(LOCAL_SQLSERVER);) {
gEngine = GISEngineFactory.createGISEngine(connection);
aoa_geometry = GISObjectFactory.createGISObject(scenario_geometry, gEngine);
if (aoa_geometry.getType() == GISObject.GISObjectType.featurecollection) { //May it also be a GeometryCollection??
//For now disable this for the SM models
throw new ServiceException("This SM service requires a polygon or multipolygon as an AoI geometry.");
// This is okay, just need to step through each feature, and also need
// to submit the feature collection to watershed/m/averageslope (DEM service), making
// sure that each has a field in properties with which to index it.
// validateFeatureCollection((GIS_FeatureCollection) aoa_geometry);
// dem_featureCollection = scenario_geometry;
} else {
// If this is a polygon/multipolygon object, need to convert
// it into a featurecollection for the averageslope (DEM service)
if ((aoa_geometry.getType() == GISObject.GISObjectType.polygon
|| aoa_geometry.getType() == GISObject.GISObjectType.multipolygon)) {
dem_featureCollection = convertPolyToFeatureCollection(aoa_geometry);
} else {
throw new ServiceException("The input geometry must be a Polygon, Multipolygon, or FeatureCollection.");
}
}
// Shortcut to getting the centroid of a shape...Built in to the GISObject source for getLat/Lon if shape is not a point.
latitude = aoa_geometry.getLatitude();
longitude = aoa_geometry.getLongitude();
getApplicableANNs();
}
} else {
throw new ServiceException("A required input " + SCENARIO_GEOMETRY + " was missing. Please specify a scenario_geometry value.");
}
}
protected void getApplicableANNs() throws ServiceException {
// Get region data for this AoI first before running or using any other resources.
// If this AoI cannot be modeled by existing ANN's currently available error-out before using any other resources.
StateCounty stateCountyCall = new StateCounty(metainfo().toString(), latitude, longitude, stateCountyURI);
stateCountyCall.setAsync(false);
stateCountyCall.call();
countyCode = stateCountyCall.countyCode();
county = stateCountyCall.county();
stateAbbrev = stateCountyCall.stateAbbrev();
String[] tCounty = county.trim().split("\\s+");
if (tCounty.length <= 0) {
throw new ServiceException("Cannot determne which county this AoI is in for the State of: " + stateAbbrev);
}
String shortCounty = tCounty[0];
weps_ann_name = getWEPSANN(stateAbbrev, shortCounty);
wepp_ann_name = getWEPPANN(stateAbbrev, shortCounty);
if (wepp_ann_name.isEmpty() || weps_ann_name.isEmpty()) {
String extendedMessage = "";
int count = 0;
if (wepp_ann_name.isEmpty()) {
extendedMessage += "\nThe WEPP model requested was for " + shortCounty + " County, " + stateAbbrev + ".\n\tThe Available WEPP ANN models are for:";
for (String model : WEPP_ANNS.keySet()) {
extendedMessage += ((count > 0) ? "," : "") + "\n\t\t\t" + model;
count++;
}
}
if (weps_ann_name.isEmpty()) {
extendedMessage += "\nThe WEPS model requested was for " + shortCounty + " County, " + stateAbbrev + ".\n\tThe Available WEPS ANN models are for:";
count = 0;
for (String model : WEPP_ANNS.keySet()) {
extendedMessage += ((count > 0) ? "," : "") + "\n\t\t\t" + model;
count++;
}
}
throw new ServiceException("The AoI geometry provided does not correspond to any WEPP/WEPS SM ANN models currently present on this system. " + extendedMessage);
}
}
protected void addNewAsyncCall(ServiceCall service, Object data) {
ParallelCallData pData = new ParallelCallData(this, data);
service.setCallData(pData);
asyncCalls.add(new AsyncCalls(pData, service));
}
@Override
protected void doProcess() throws Exception {
// Build queue of 5 service calls that can be all simultaneiously called.
// Get all soils information for this AoI.
addNewAsyncCall(new WWESoilParams(metainfo().toString(), scenario_geometry, soilsURI, V1_0::getSoilsData), null);
///////////////////////////////////
// Get Cligen data for this AoI
// WEPP Cligen Call
addNewAsyncCall(new Climate(metainfo().toString(), aoa_geometry, climateURI, 100, V1_0::getWeppClimateData), null);
// WEPS Cligen call
addNewAsyncCall(new Climate(metainfo().toString(), aoa_geometry, climateURI, 50, V1_0::getWepsClimateData), null);
// End Cligen calls
//////////////////////////////////
// Get Windgen data for this AoI
addNewAsyncCall(new WindGen(metainfo().toString(), aoa_geometry, windURI, V1_0::getWindData), null);
// Get region data for this AoI
addNewAsyncCall(new Region(metainfo().toString(), scenario_geometry, regionURI, V1_0::getRegionData), null);
// We must wait for WWESoilPArams to finish before we can continue so that we
// get all the cokeys we need and fill in the top three list.
// Wait for above services to return results, and clear list of pending calls when done.
waitForParallelCalls(asyncCalls, true);
// Create sol and ifc files for the selected three components.
// Now get information for each of the top three soils chosen.
for (SoilResult soilResult : topThreeComponents) {
addNewAsyncCall(new WEPPSoilInput(metainfo().toString(), soilResult.cokey, weppSoilInputURI, V1_0::getWEPPSoilData), soilResult);
addNewAsyncCall(new WEPSSoilInput(metainfo().toString(), soilResult.cokey, wepsSoilInputURI, V1_0::getWEPSSoilData), soilResult);
if (useDemSlope || compareSlopes) {
addNewAsyncCall(new DEMSteepness(metainfo().toString(), soilResult.polygonList, demSteepnessURI, V1_0::getSoilSlopeData, SLOPE_VERSION_4), soilResult);
}
}
// We must wait for all WEPP/WEPSSoilInputs to finish before we can continue so that we
// get all the IFC and SOL file data that we need to run the models.
// Wait for above WEPS/WEPP SoilInput services to return results, and clear list of pending calls when done.
waitForParallelCalls(asyncCalls, true);
// Get DEM slope information, if requested.
if (useDemSlope || compareSlopes) {
getDEMSlopes(topThreeComponents);
}
// Copy these soilResults to new instances for use in comparison model runs.
// No need to re-run WEPPSoilInputs on these since we take the slope steepness
// value from the soilResult structure not the IFC file.
if (compareSlopes) {
alternateSlopeTopThreeComponents = new ArrayList<>();
for (SoilResult soilResult : topThreeComponents) {
SoilResult tSoilResult = new SoilResult(soilResult);
alternateSlopeTopThreeComponents.add(tSoilResult);
}
}
//TODO: Currently rotations are the default fallow operation...final version of this service requires user to provide rotation.
double totalArea = 0.0;
adjustWEPSYears();
for (SoilResult soilResult : topThreeComponents) {
WEPSMetaData wepsMetaData = new WEPSMetaData();
WEPPMetaData weppMetaData = new WEPPMetaData();
// Set which soil slope to use before creating the service.
soilResult.usedDEM = useDemSlope;
fillMetaDataFields(soilResult, wepsMetaData, weppMetaData);
addNewAsyncCall(new SMWEPP(metainfo().toString(), wepp_ann_name, weppMetaData, weppURI, V1_0::getSMWEPPData), soilResult);
addNewAsyncCall(new SMWEPS(metainfo().toString(), weps_ann_name, wepsMetaData, wepsURI, V1_0::getSMWEPSData), soilResult);
totalArea += soilResult.area;
}
if (compareSlopes) {
for (SoilResult soilResult : alternateSlopeTopThreeComponents) {
WEPSMetaData wepsMetaData = new WEPSMetaData();
WEPPMetaData weppMetaData = new WEPPMetaData();
// Set which soil slope to use before creating the service.
soilResult.usedDEM = !useDemSlope;
fillMetaDataFields(soilResult, wepsMetaData, weppMetaData);
addNewAsyncCall(new SMWEPP(metainfo().toString(), wepp_ann_name, weppMetaData, weppURI, V1_0::getSMWEPPData), soilResult);
}
}
if (!asyncCalls.isEmpty()) {
// Wait for the simultaneous calls to all finish and set their appropriate data, may not clear the list when done. We'll use this list in writing output data when in debug mode, only.
waitForParallelCalls(asyncCalls, !debugOutput);
} else {
throw new ServiceException("No WEPP/WEPS model runs were created. Cannot proceed.");
}
for (SoilResult soilResult : topThreeComponents) {
if (Double.isNaN(soilResult.waterErosion) || Double.isNaN(soilResult.windErosion)) {
throw new ServiceException("Not a number value encountered while trying to average wind or water erosion values.");
}
waterErosionRate += soilResult.waterErosion * (soilResult.area / totalArea);
windErosionRate += soilResult.windErosion * (soilResult.area / totalArea);
}
}
@Override
protected void postProcess() throws Exception {
PayloadResults results = results();
JSONArray soilsResult = new JSONArray();
for (SoilResult soilResult : topThreeComponents) {
JSONArray soilData = new JSONArray();
if (debugOutput) {
soilData.put(JSONUtils.data("mukey", soilResult.mukey));
soilData.put(JSONUtils.data("cokey", soilResult.cokey));
soilData.put(JSONUtils.data("soilName", soilResult.soilName));
soilData.put(JSONUtils.data("slope_method", (!soilResult.usedDEM ? "SDM slope_r value" : "DEM Slope Service")));
soilData.put(JSONUtils.data("slope_r", soilResult.slope_r));
soilData.put(JSONUtils.data("dem_slope", ((!Double.isNaN(soilResult.slopeDEM)) ? soilResult.slopeDEM * 100.0 : "NONE")));
soilData.put(JSONUtils.data("length", soilResult.length));
soilData.put(JSONUtils.data("soilName", soilResult.soilName));
soilData.put(JSONUtils.data("soilLongName", soilResult.soilLongName));
}
soilData.put(JSONUtils.data("soilPtr", soilResult.cokey));
soilData.put(JSONUtils.data("soilName", soilResult.soilName));
soilData.put(JSONUtils.data("area_pct", soilResult.area_pct, "The percentage of the component of the mapunit intersection with the CRP Offer Area", "Percent"));
soilData.put(JSONUtils.data("area", soilResult.area, "Soil Component Area (Acres) in the CRP Offer area", "Acres"));
soilData.put(JSONUtils.data("WaterSoilLoss", soilResult.waterErosion, "Average Annual Soil Loss by Water", "ton/ac/yr"));
soilData.put(JSONUtils.data("WindSoilLoss", soilResult.windErosion, "Average Annual Soil Loss by Wind", "ton/ac/yr"));
soilsResult.put(soilData);
}
results.put("top_3_soils", soilsResult);
results.put("WtAvgWaterSoilLoss", waterErosionRate, "Weighted Average Annual Soil Loss by Water", "ton/ac/yr");
results.put("WtAvgWindSoilLoss", windErosionRate, "Weighted Average Annual Soil Loss by Wind", "ton/ac/yr");
results.put("WtAvgTotalSoilLoss", waterErosionRate + windErosionRate, "Weighted Average Annual Soil Loss by Water and Wind", "ton/ac/yr");
if (debugOutput) {
JSONArray soilsDebugResults = new JSONArray();
JSONArray subCalls = new JSONArray();
for (AsyncCalls aCall : asyncCalls) {
JSONArray allModelData = new JSONArray();
SoilResult soilResult = (SoilResult) aCall.callData().callData();
switch (aCall.service().getClass().getName()) {
case "crp.utils.SMWEPP":
allModelData.put(JSONUtils.data("mukey", soilResult.mukey));
allModelData.put(JSONUtils.data("cokey", soilResult.cokey));
allModelData.put(JSONUtils.data("soilName", soilResult.soilName));
allModelData.put(JSONUtils.data("slope_method", (!soilResult.usedDEM ? "SDM slope_r value" : "DEM Slope Service")));
allModelData.put(JSONUtils.data("slope_r", soilResult.slope_r));
allModelData.put(JSONUtils.data("dem_slope", ((!Double.isNaN(soilResult.slopeDEM)) ? soilResult.slopeDEM * 100.0 : "NONE")));
allModelData.put(JSONUtils.data("length", soilResult.length));
allModelData.put(JSONUtils.data("soilName", soilResult.soilName));
allModelData.put(JSONUtils.data("soilLongName", soilResult.soilLongName));
allModelData.put(JSONUtils.data("area_pct", soilResult.area_pct, "The percentage of the component of the mapunit intersection with the CRP Offer Area", "Percent"));
allModelData.put(JSONUtils.data("area", soilResult.area, "Soil Component Area (Acres) in the CRP Offer area", "Acres"));
allModelData.put(JSONUtils.data("WaterSoilLoss", EvalResult.writeDouble(soilResult.waterErosion), "Average Annual Soil Loss by Water", "ton/ac/yr"));
allModelData.put(JSONUtils.data("WindSoilLoss", EvalResult.writeDouble(soilResult.windErosion), "Average Annual Soil Loss by Wind", "ton/ac/yr"));
allModelData.put(JSONUtils.data("wepp_request", aCall.service().requestJSON(), "Request JSON sent to the WEPP model"));
allModelData.put(JSONUtils.data("wepp_result", aCall.service().getResults(), "Result JSON returned from the WEPP model"));
subCalls.put(allModelData);
break;
}
}
soilsDebugResults.put(JSONUtils.data("WEPP_CALLS", subCalls));
subCalls = new JSONArray();
for (AsyncCalls aCall : asyncCalls) {
JSONArray allModelData = new JSONArray();
SoilResult soilResult = (SoilResult) aCall.callData().callData();
switch (aCall.service().getClass().getName()) {
case "crp.utils.SMWEPS":
allModelData.put(JSONUtils.data("mukey", soilResult.mukey));
allModelData.put(JSONUtils.data("cokey", soilResult.cokey));
allModelData.put(JSONUtils.data("soilName", soilResult.soilName));
allModelData.put(JSONUtils.data("slope_method", (!soilResult.usedDEM ? "SDM slope_r value" : "DEM Slope Service")));
allModelData.put(JSONUtils.data("slope_r", soilResult.slope_r));
allModelData.put(JSONUtils.data("dem_slope", ((!Double.isNaN(soilResult.slopeDEM)) ? soilResult.slopeDEM * 100.0 : "NONE")));
allModelData.put(JSONUtils.data("length", soilResult.length));
allModelData.put(JSONUtils.data("soilName", soilResult.soilName));
allModelData.put(JSONUtils.data("soilLongName", soilResult.soilLongName));
allModelData.put(JSONUtils.data("area_pct", soilResult.area_pct, "The percentage of the component of the mapunit intersection with the CRP Offer Area", "Percent"));
allModelData.put(JSONUtils.data("area", soilResult.area, "Soil Component Area (Acres) in the CRP Offer area", "Acres"));
allModelData.put(JSONUtils.data("WaterSoilLoss", EvalResult.writeDouble(soilResult.waterErosion), "Average Annual Soil Loss by Water", "ton/ac/yr"));
allModelData.put(JSONUtils.data("WindSoilLoss", EvalResult.writeDouble(soilResult.windErosion), "Average Annual Soil Loss by Wind", "ton/ac/yr"));
allModelData.put(JSONUtils.data("weps_request", aCall.service().requestJSON(), "Request JSON sent to the WEPS model"));
allModelData.put(JSONUtils.data("weps_result", aCall.service().getResults(), "Result JSON returned from the WEPS model"));
subCalls.put(allModelData);
break;
}
}
soilsDebugResults.put(JSONUtils.data("WEPS_CALLS", subCalls));
results.put("DEBUGGING_DATA", soilsDebugResults);
}
}
protected String getWEPSANN(String stateAbbrev, String county) {
return WEPS_ANNS.getOrDefault((stateAbbrev + "_" + county).toLowerCase(), "");
}
protected String getWEPPANN(String stateAbbrev, String county) {
return WEPP_ANNS.getOrDefault((stateAbbrev + "_" + county).toLowerCase(), "");
}
protected void waitForParallelCalls(ArrayList<AsyncCalls> asyncCalls, boolean clearCalls) throws ServiceException {
if (!asyncCalls.isEmpty()) {
List<Future<String>> taskList = new ArrayList<>();
//ArrayList<ServiceCall> tasks = new ArrayList<>();
ExecutorService executor = Executors.newCachedThreadPool();
// for (AsyncCalls asyncCall : asyncCalls) {
// tasks.add(asyncCall.service());
// }
try {
// Using this per task method instead of invokeAll as testing with invokeAll()
// seems to overload the web server front end of this same context, under some circumstances.
// A 150ms wait between them is not much of a loss in time, but can prevent
// unfair loading of the services frontends, and allows other clients to interleave their calls here too.
for (AsyncCalls asyncCall : asyncCalls) {
taskList.add(executor.submit(asyncCall.service()));
Thread.sleep(150);
}
//taskList = executor.invokeAll(tasks);
} catch (Exception ex) {
for (int i = 0; i < taskList.size(); i++) {
Future<String> task = taskList.get(i);
if (!task.isDone()) {
task.cancel(true); // Attempt to force a stop of this task, may fail.
// try {
// // Force a cancel operation on the running service call, csip-core
// // allows for this, take advantage of it. This should also cause the running thread to finish.
// asyncCalls.get(i).service().cancel();
// } catch (ServiceException e) {
// Logger.getLogger(V1_0.class.getName()).log(Level.SEVERE, "Couldn't cancel this service call: " + e.getMessage(), e);
// }
}
}
// Not using async calls, so cannot take advantage of csip-core cancel operation, must just force stop on running threads of sync calls.
// Cleanup any previously canceled tasks that did not stop.
executor.shutdownNow();
throw new ServiceException("Forced to cancel pending service requests due to a timeout.", ex);
}
int count = 0;
for (Future<String> task : taskList) {
try {
// Effectively the same as wait or wait for CountDownLatch, as this pauses
// this thread until the task is "Done" and the return value is available.
// We are not concerned about doing anything with these service call threads'
// data here because each thread had a CallBack function to call in order to
// have its data used by this class already. Here we are just waiting for
// them to all finish and then cleaning up the pool.
if (task.get().equalsIgnoreCase("finished")) {
count++;
}
} catch (InterruptedException | ExecutionException ex) {
// Any interrupt at this point means this service needs to fail.
executor.shutdownNow(); //don't wait, force all existing, outstanding calls, if any, to stop.
throw new ServiceException("Cannot continue with the CRP Assessment. A task get() function for an external service call was interrupted: " + ex.getMessage(), ex);
}
}
// Clean up the list if desired. (Some calling code reuses this list, so this is an option here.
if (clearCalls) {
asyncCalls.clear();
}
// If we didn't get all of our parallel calls completed, then we've failed.
if (count != taskList.size()) {
// Should be all done if we got here, but if we're in this block, then
// something is still pending and hung for some reason. ShutdownNow will force these to stop.
// A plain "shutdown()" won't work here as it allows a thread to finish and we want to force a stop.
executor.shutdownNow();
throw new ServiceException("Not all calls finished, some have failed. Check other logs for error messages.");
}
} else {
throw new ServiceException("No asynchronous calls were created. Cannot proceed with wait.");
}
}
protected void getDEMSlopes(ArrayList<SoilResult> soils) throws ServiceException, SQLException, GISObjectException, JSONException, IOException {
try (Connection connection = resources().getJDBC(LOCAL_SQLSERVER);) {
gEngine = GISEngineFactory.createGISEngine(connection);
for (SoilResult soilResult : soils) {
double tSlope = Double.NaN;
//slopes = getDEMSteepness(soilResult.polygonList);
slopes = soilResult.slopes;
if (null != slopes) {
if (!slopes.badSlopeData() && slopes.slopeDataMessages().isEmpty()) {
double wAvgSlope = 0.0;
double totalAcres = 0.0;
//Get overall average slope for these intersected polygons from the result file.
for (int i = 0; i < slopes.numSlopes(); i++) {
SlopeData tSlopeData = slopes.slope(i + 1);
if (null != tSlopeData) {
double tAvg = tSlopeData.mean();
double tAcres = 0.0;
JSONObject polyJSON = soilResult.polygonList.get(i);
GISObject tVector = GISObjectFactory.createGISObject(polyJSON, gEngine);
tAcres = tVector.areaInAcres();
wAvgSlope += tAcres * tAvg;
totalAcres += tAcres;
}
}
if (0.0 != totalAcres) {
tSlope = (wAvgSlope / totalAcres);
if (tSlope < 0.01) {
tSlope = 0.01;
}
} else {
tSlope = 0.0;
Logger.getLogger(ServiceCall.class.getName()).log(Level.WARNING, "Total acres for this cokey's," + soilResult.cokey + " mapunit intersections was zero. Setting DEM slope to 0 .");
}
} else {
throw new ServiceException("The DEM returned slopes had bad data in the returned file. " + ((!slopes.slopeDataMessages().isEmpty()) ? " Bad data message: " + slopes.slopeDataMessages() : ""));
}
} else {
throw new ServiceException("No data was returned from the DEM slope servcie for this cokey, " + soilResult.cokey + " and intersected mapunit polygons");
}
if (Double.isNaN(tSlope)) {
throw new ServiceException("DEM slope calculation requested, but no DEM slope could be found for this soil: " + soilResult.cokey);
}
soilResult.slopeDEM = tSlope;
}
}
}
protected void adjustWEPSYears() {
////////////////////////////
// WEPS adjustments
wepsDayCountPrecipAndWind = 0;
wepsYearsSimulated = windgenData.yearsInFile();
if (wepsYearsSimulated > wepsCligenData.yearsSimulated()) {
wepsYearsSimulated = wepsCligenData.yearsInFile();
}
for (int i = 0; i < wepsYearsSimulated; i++) {
for (int j = 0; j < 12; j++) {
for (int k = 0; k < 31; k++) {
if (wepsDaysWithPrecipAndWind[i][j][k] >= 2) {
wepsDayCountPrecipAndWind++;
}
}
}
}
// End WEPS Adjustments
////////////////////////////////
}
protected void fillMetaDataFields(SoilResult soilResult, WEPSMetaData wepsMetaData, WEPPMetaData weppMetaData) throws IOException {
calcWEPSMetaData(soilResult, wepsMetaData);
calcWEPPMetaData(soilResult, weppMetaData);
}
protected void calcWEPSMetaData(SoilResult soilResult, WEPSMetaData wepsMetaData) throws IOException {
IFCFile ifcFile = new IFCFile(soilResult.ifcFile);
//Average the number of days where both precipitation amount and wind energy exceeded the minimum thresholds for each. Round up??
wepsMetaData.daysWithPrecipAndErosiveWind(((wepsYearsSimulated != 0) ? (int) Math.ceil((double) wepsDayCountPrecipAndWind / (double) wepsYearsSimulated) : 0));
wepsMetaData.cokey(soilResult.cokey);
wepsMetaData.annualPrecip(wepsCligenData.annualAvgPrecip());
wepsMetaData.avgAnnualTMin(wepsCligenData.avgAnnualTMin());
wepsMetaData.avgAnnualTMax(wepsCligenData.avgAnnualTMax());
wepsMetaData.annualErosiveWindEnergy(windgenData.annualErosiveWindEnergy());
// HOW to calculate this one?? --> wepsMetaData.daysWithPrecipAndErosiveWind(cligenData.daysWithPrecipAndErosiveWind());
wepsMetaData.windEnergy(windgenData.simulationAverage());
wepsMetaData.componentName(ifcFile.componentName);
wepsMetaData.fractionSand(ifcFile.fractionSand);
wepsMetaData.fractionSilt(ifcFile.fractionSilt);
wepsMetaData.fractionClay(ifcFile.fractionClay);
wepsMetaData.crustStability(ifcFile.crustStability);
wepsMetaData.surfRockFrag(ifcFile.surfaceFragmentCover);
wepsMetaData.albedo(ifcFile.surfaceAlbedo);
wepsMetaData.num_layers(ifcFile.layerThickness.length);
wepsMetaData.surface_thickness(((int) ifcFile.layerThickness[0]));
wepsMetaData.slope_gradient(ifcFile.surfaceSlope);
//TODO: Do we need to measure the layers before setting these?? WEPSSoilInput already filters/sorts layers, etc...
wepsMetaData.aggStability(ifcFile.aggregateStability[0]);
wepsMetaData.soilWiltPoint(ifcFile.wiltingPointSWC[0]);
wepsMetaData.aggGeomDiam(ifcFile.aggregateMeanDiameter[0]);
}
protected void calcWEPPMetaData(SoilResult soilResult, WEPPMetaData weppMetaData) throws IOException {
SOLFile solFile = new SOLFile(soilResult.solFile);
weppMetaData.cokey(soilResult.cokey);
weppMetaData.annualPrecip(weppCligenData.annualAvgPrecip());
weppMetaData.avgAnnualPrecipEventDuration(weppCligenData.nonZeroDurationAverage());
weppMetaData.interrillErodibility(solFile.ki);
weppMetaData.rillErodibility(solFile.kr);
double tClaySum = 0.0;
double tSandSum = 0.0;
double tDepth = 0.0;
for (int i = 0; i < solFile.numLayers; i++) {
tClaySum += solFile.clay[i] * solFile.solthk[i];
tSandSum += solFile.sand[i] * solFile.solthk[i];
tDepth += solFile.solthk[i];
}
weppMetaData.wAvgSoilClayFraction(tClaySum / tDepth);
weppMetaData.wAvgSoilSandFraction(tSandSum / tDepth);
weppMetaData.effectiveSurfaceConductivity(solFile.avke);
weppMetaData.slopeSteepness((soilResult.usedDEM ? soilResult.slopeDEM : (soilResult.slope_r / 100.0)));
weppMetaData.componentName(solFile.soilName);
weppMetaData.num_layers(solFile.numLayers);
weppMetaData.surface_thickness(((int) solFile.solthk[0]));
}
//TODO finish when Jack is ready.
protected JSONObject getRotation(int length) throws JSONException {
return new WEPSFallowRotation(length).rotation();
}
protected SlopeSteepness getDEMSteepness() throws ServiceException {
if (!demSteepnessURI.equals("NONE")) {
DEMSteepness slopeCall = new DEMSteepness(metainfo().toString(), dem_featureCollection, demSteepnessURI);
slopeCall.call();
return slopeCall.slopes();
}
return null;
}
protected SlopeSteepness getDEMSteepness(ArrayList<JSONObject> intersectedPolygons) throws ServiceException {
if (!demSteepnessURI.equals("NONE")) {
DEMSteepness slopeCall = new DEMSteepness(metainfo().toString(), intersectedPolygons, demSteepnessURI);
slopeCall.call();
return slopeCall.slopes();
}
return null;
}
protected void validateFeatureCollection(GIS_FeatureCollection featureCollection) throws ServiceException {
int featureCount = featureCollection.getFeatureCount();
if (0 < featureCount) {
String id;
try {
for (int i = 0; i < featureCount; i++) {
if (null != (id = featureCollection.getFeatureAttribute(i, "id"))) {
if (id.isEmpty()) {
throw new ServiceException("Feature " + (i + 1) + " in the input JSON FeatureCollection does not have a property value for 'id'. Cannot proceed without this value.");
}
} else {
throw new ServiceException("Feature " + (i + 1) + " in the input JSON FeatureCollection does not have a property 'id'. Cannot proceed without it.");
}
}
} catch (GISObjectException ex) {
throw new ServiceException("Could not find the 'id' attribute within the first feature in this input JSON FeatureCollection. Cannot proceed: " + ex.getMessage(), ex);
}
} else {
throw new ServiceException("No features were found in the input JSON FeatureCollection.");
}
}
protected JSONObject convertPolyToFeatureCollection(GISObject polygon_object) throws ServiceException, IOException, JSONException {
return new JSONObject(
" {\n"
+ " \"name\": \"boundary\",\n"
+ " \"value\": [{\n"
+ " \"type\": \"FeatureCollection\",\n"
+ " \"features\": [\n"
+ " {\n"
+ " \"id\": \"1\",\n"
+ " \"type\": \"Feature\",\n"
+ " \"properties\": {},\n"
+ " \"geometry\":" + polygon_object.toJSON()
+ " }\n"
+ " ]\n"
+ " }]"
+ "}\n"
);
}
protected void writeOutputFile(String data, String filename) throws ServiceException {
byte[] outBytes = data.getBytes();
FileOutputStream fileOut = null;
BufferedOutputStream bufferOut;
File outFile = workspace().getFile(filename);
try {
fileOut = new FileOutputStream(outFile);
} catch (FileNotFoundException ex) {
throw new ServiceException("Cannot open that file: " + filename, ex);
}
bufferOut = new BufferedOutputStream(fileOut);
try {
bufferOut.write(outBytes, 0, outBytes.length);
bufferOut.flush();
bufferOut.close();
} catch (IOException ex) {
throw new ServiceException("Cannot write to that file: " + filename, ex);
}
}
public class ParallelCallData {
private V1_0 instance = null;
private Object callData = null;
private CountDownLatch latch;
public ParallelCallData(V1_0 instance, Object data) {
this.instance = instance;
callData = data;
}
public V1_0 instance() {
return instance;
}
public Object callData() {
return callData;
}
public void setLatch(CountDownLatch latch) {
this.latch = latch;
}
}
public class AsyncCalls {
private FutureTask<String> task;
private ParallelCallData callData;
private ServiceCall service;
public AsyncCalls(ParallelCallData callData, ServiceCall service) {
this.task = new FutureTask<>(service);
this.callData = callData;
this.service = service;
}
public FutureTask<String> task() {
return task;
}
public ParallelCallData callData() {
return callData;
}
public ServiceCall service() {
return service;
}
}
}