Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_candidate_model_json_none_and_nan_values():
eemeter_warning = EEMeterWarning(
qualified_name="qualified_name", description="description", data={}
)
candidate_model = CalTRACKUsagePerDayCandidateModel(
model_type="model_type",
formula="formula",
status="status",
warnings=[eemeter_warning],
r_squared_adj=None,
)
assert candidate_model.json()["r_squared_adj"] is None
candidate_model.r_squared_adj = np.nan
assert candidate_model.json()["r_squared_adj"] is None
def test_candidate_model_json_with_warning():
eemeter_warning = EEMeterWarning(
qualified_name="qualified_name", description="description", data={}
)
candidate_model = CalTRACKUsagePerDayCandidateModel(
model_type="model_type",
formula="formula",
status="status",
warnings=[eemeter_warning],
)
assert candidate_model.json() == {
"formula": "formula",
"model_params": {},
"model_type": "model_type",
"r_squared_adj": None,
"status": "status",
"warnings": [
{
def test_model_results_json_with_objects():
candidate_model = CalTRACKUsagePerDayCandidateModel(
model_type="model_type", formula="formula", status="status"
)
eemeter_warning = EEMeterWarning(
qualified_name="qualified_name", description="description", data={}
)
model_results = CalTRACKUsagePerDayModelResults(
status="status",
method_name="method_name",
model=candidate_model,
candidates=[candidate_model],
warnings=[eemeter_warning],
)
assert model_results.json(with_candidates=True) == {
"candidates": [
{
"formula": "formula",
"model_params": {},
"model_type": "model_type",
"r_squared_adj": None,
period_days : :any:`pandas.Series`
A series of containing day counts.
minimum_total : :any:`float`
Minimum allowable total sum of degree day values.
Returns
-------
warnings : :any:`list` of :any:`eemeter.EEMeterWarning`
Empty list or list of single warning.
"""
warnings = []
total_degree_days = (avg_degree_days * period_days).sum()
if total_degree_days < minimum_total:
warnings.append(
EEMeterWarning(
qualified_name=(
"eemeter.caltrack_daily.{model_type}.total_{degree_day_type}_too_low".format(
model_type=model_type, degree_day_type=degree_day_type
)
),
description=(
"Total {degree_day_type} below accepted minimum."
" Candidate fit not attempted.".format(
degree_day_type=degree_day_type.upper()
)
),
data={
"total_{degree_day_type}".format(
degree_day_type=degree_day_type
): total_degree_days,
"total_{degree_day_type}_minimum".format(
model_type : :any:`str`
Model type (e.g., ``'cdd_hdd'``).
model_params : :any:`dict`
Parameters as stored in :any:`eemeter.CalTRACKUsagePerDayCandidateModel.model_params`.
parameter : :any:`str`
The name of the parameter, e.g., ``'intercept'``.
Returns
-------
warnings : :any:`list` of :any:`eemeter.EEMeterWarning`
Empty list or list of single warning.
"""
warnings = []
if model_params.get(parameter, 0) < 0:
warnings.append(
EEMeterWarning(
qualified_name=(
"eemeter.caltrack_daily.{model_type}.{parameter}_negative".format(
model_type=model_type, parameter=parameter
)
),
description=(
"Model fit {parameter} parameter is negative. Candidate model rejected.".format(
parameter=parameter
)
),
data=model_params,
)
)
return warnings
Parameters
----------
model_type : :any:`str`
Model type (e.g., ``'cdd_hdd'``).
formula : :any:`float`
The candidate model formula.
Returns
-------
candidate_model : :any:`eemeter.CalTRACKUsagePerDayCandidateModel`
Candidate model instance with status ``'ERROR'``, and warning with
traceback.
"""
warnings = [
EEMeterWarning(
qualified_name="eemeter.caltrack_daily.{}.model_results".format(model_type),
description=(
"Error encountered in statsmodels.formula.api.ols method. (Empty data?)"
),
data={"traceback": traceback.format_exc()},
)
]
return CalTRACKUsagePerDayCandidateModel(
model_type=model_type, formula=formula, status="ERROR", warnings=warnings
)
def _make_reporting_warnings(
end_inf, start_inf, data_start, data_end, start_limit, end_limit
):
warnings = []
# warn if there is a gap at end
if not end_inf and data_end < end_limit:
warnings.append(
EEMeterWarning(
qualified_name="eemeter.get_reporting_data.gap_at_reporting_end",
description=(
"Data does not have coverage at requested reporting end date."
),
data={
"requested_end": end_limit.isoformat(),
"data_end": data_end.isoformat(),
},
)
)
# warn if there is a gap at start
if not start_inf and start_limit < data_start:
warnings.append(
EEMeterWarning(
qualified_name="eemeter.get_reporting_data.gap_at_reporting_start",
description=(
observed = observed_input.to_frame().dropna()
predicted = predicted_input.to_frame().dropna()
observed.columns = ["observed"]
predicted.columns = ["predicted"]
self.observed_length = observed.shape[0]
self.predicted_length = predicted.shape[0]
# Do an inner join on the two input series to make sure that we only
# use observations with the same time stamps.
combined = observed.merge(predicted, left_index=True, right_index=True)
self.merged_length = len(combined)
if self.observed_length != self.predicted_length:
self.warnings.append(
EEMeterWarning(
qualified_name="eemeter.metrics.input_series_are_of_different_lengths",
description="Input series are of different lengths.",
data={
"observed_input_length": len(observed_input),
"predicted_input_length": len(predicted_input),
"observed_length_without_nan": self.observed_length,
"predicted_length_without_nan": self.predicted_length,
"merged_length": self.merged_length,
},
)
)
# Calculate residuals because these are an input for most of the metrics.
combined["residuals"] = combined.predicted - combined.observed
self.num_parameters = num_parameters
warnings.append(
EEMeterWarning(
qualified_name="eemeter.get_reporting_data.gap_at_reporting_end",
description=(
"Data does not have coverage at requested reporting end date."
),
data={
"requested_end": end_limit.isoformat(),
"data_end": data_end.isoformat(),
},
)
)
# warn if there is a gap at start
if not start_inf and start_limit < data_start:
warnings.append(
EEMeterWarning(
qualified_name="eemeter.get_reporting_data.gap_at_reporting_start",
description=(
"Data does not have coverage at requested reporting start date."
),
data={
"requested_start": start_limit.isoformat(),
"data_start": data_start.isoformat(),
},
)
)
return warnings
The maximum allowable p-value of the parameter.
Returns
-------
warnings : :any:`list` of :any:`eemeter.EEMeterWarning`
Empty list or list of single warning.
"""
warnings = []
if p_value > maximum_p_value:
data = {
"{}_p_value".format(parameter): p_value,
"{}_maximum_p_value".format(parameter): maximum_p_value,
}
data.update(model_params)
warnings.append(
EEMeterWarning(
qualified_name=(
"eemeter.caltrack_daily.{model_type}.{parameter}_p_value_too_high".format(
model_type=model_type, parameter=parameter
)
),
description=(
"Model fit {parameter} p-value is too high. Candidate model rejected.".format(
parameter=parameter
)
),
data=data,
)
)
return warnings