Simple but lots of cleanups

glhe/aggregation/agg_factory.py

@@ -9,7 +9,7 @@ def make_agg_method(inputs: dict, ip: InputProcessor):
     """
     Factory method for creating load aggregation objects
 
-    :param inputs: load aggregation inputs
+    :param inputs: load aggregation input dictionary
     :param ip: input processor instance
     :return: load aggregation object
     """

glhe/aggregation/dynamic.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 import numpy as np
 
 from glhe.aggregation.agg_types import AggregationTypes
@@ -92,7 +90,7 @@ def aggregate(self, time: int, energy: float):
         # update time
         self.prev_update_time = time
 
-    def calc_temporal_superposition(self, time_step: int, flow_rate: float = None) -> Union[float, tuple]:
+    def calc_temporal_superposition(self, time_step: int, flow_rate: float = None) -> float | tuple[float, float]:
 
         # compute temporal superposition
         # this includes all thermal history before the present time

glhe/aggregation/no_agg.py

@@ -6,12 +6,12 @@
 
 class NoAgg(BaseAgg):
     """
-    No aggregation. Just keep all of the values.
+    No aggregation. Just keep all the values.
     """
 
     Type = AggregationTypes.NO_AGG
 
-    def __init__(self, inputs):
+    def __init__(self, inputs: dict):
         BaseAgg.__init__(self, inputs)
 
     def aggregate(self, time: int, energy: float):
@@ -27,7 +27,7 @@ def aggregate(self, time: int, energy: float):
         # update time
         self.prev_update_time = time
 
-    def calc_temporal_superposition(self, time_step: int) -> float:
+    def calc_temporal_superposition(self, time_step: int, _: float = None) -> float:
         # compute temporal superposition
         # this includes all thermal history before the present time
         q = self.energy / self.dts
@@ -51,7 +51,7 @@ def calc_temporal_superposition(self, time_step: int) -> float:
     def get_g_value(self, time_step: int) -> float:
         pass  # pragma: no cover
 
-    def get_g_b_value(self, time_step: int) -> float:
+    def get_g_b_value(self, time_step: int, _: float = None) -> float:
         pass  # pragma: no cover
 
     def get_q_prev(self) -> float:

glhe/aggregation/static.py

@@ -67,12 +67,25 @@ def aggregate(self, time: int, energy: float):
             return
         else:
             # aggregate
-            dts_flip = np.flipud(self.dts)
-            vals, idxs, cnts = np.unique(dts_flip, return_counts=True, return_index=True)
+            # TODO: The next two lines are showing as unused in Pycharm, is this IF block needed then?
+            # dts_flip = np.flipud(self.dts)
+            # vals, idxs, cnts = np.unique(dts_flip, return_counts=True, return_index=True)
             self.energy[-1] += e_1
 
             # numpy split will do a lot of work too
 
         # update times
         self.prev_update_time = time
         self.prev_update_time_hr = int(time / SEC_IN_HOUR)
+
+    def calc_temporal_superposition(self, time_step: int, flow_rate: float = None) -> float:
+        pass
+
+    def get_g_value(self, time_step: int) -> float:
+        pass
+
+    def get_g_b_value(self, time_step: int, flow_rate: float = None) -> float:
+        pass
+
+    def get_q_prev(self) -> float:
+        pass

glhe/g_function/flow_fraction.py

@@ -3,7 +3,7 @@
 from glhe.utilities.constants import gamma_const
 
 
-class FlowFraction(object):
+class FlowFraction:
 
     def __init__(self, inputs):
 
@@ -20,20 +20,20 @@ def __init__(self, inputs):
 
         self.alpha_s = self.k_s / self.c_s
 
-    def calc_flow_fraction(self, sim_time, vol_flow_rate, bh_int_resist, bh_ave_resist):
+    def calc_flow_fraction(self, sim_time: float, vol_flow: float, bh_int_resist: float, bh_ave_resist: float) -> float:
         """
         Computes the flow fraction based on the method outlined in:
 
         Beier, R.A., M.S. Mitchell, J.D. Spitler, S. Javed. 2018. 'Validation of borehole heat
         exchanger models against multi-flow rate thermal response tests.' Geothermics 71, 55-68.
 
-        :return flow fraction
+        :return: flow fraction
         """
 
         # Define base variables
         t_i = sim_time
 
-        w = vol_flow_rate
+        w = vol_flow
 
         # Transit time
         t_tr = self.v_f / w
@@ -62,16 +62,16 @@ def calc_flow_fraction(self, sim_time, vol_flow_rate, bh_int_resist, bh_ave_resi
 
         # Equations 11
         if 0.2 < psi <= 1.2:
-            tdsf_over_cd = -8.0554 * phi ** 3 + 3.8111 * phi ** 2 - 3.2585 * phi + 2.8004  # pragma: no cover
+            td_sf_over_cd = -8.0554 * phi ** 3 + 3.8111 * phi ** 2 - 3.2585 * phi + 2.8004  # pragma: no cover
         elif 1.2 < phi <= 160:
-            tdsf_over_cd = -0.2662 * phi ** 4 + 3.5589 * phi ** 3 - 18.311 * phi ** 2 + 57.93 * phi - 6.1661
+            td_sf_over_cd = -0.2662 * phi ** 4 + 3.5589 * phi ** 3 - 18.311 * phi ** 2 + 57.93 * phi - 6.1661
         elif 160 < phi <= 2E5:  # pragma: no cover
-            tdsf_over_cd = 12.506 * phi + 45.051  # pragma: no cover
+            td_sf_over_cd = 12.506 * phi + 45.051  # pragma: no cover
         else:
             raise ValueError  # pragma: no cover
 
         # Equation 12
-        t_sf_num = tdsf_over_cd * self.c_f * self.v_f
+        t_sf_num = td_sf_over_cd * self.c_f * self.v_f
         t_sf_den = 2 * pi * self.l_bh * self.k_s
         t_sf = t_sf_num / t_sf_den + self.t_i_minus_1
 
@@ -127,7 +127,16 @@ def calc_flow_fraction(self, sim_time, vol_flow_rate, bh_int_resist, bh_ave_resi
         return f
 
     @staticmethod
-    def calc_soil_resist(sim_time, bh_resist, soil_cond, soil_diff):
+    def calc_soil_resist(sim_time: float, bh_resist: float, soil_cond: float, soil_diff: float) -> float:
+        """
+        Calculate soil resistance
+
+        :param sim_time: simulation time, in units
+        :param bh_resist: comment
+        :param soil_cond: comment
+        :param soil_diff: comment
+        :return: soil resistance
+        """
         part_1 = 2 / (4 * pi * soil_cond)
         part_2_num = 4 * soil_diff * sim_time
         if part_2_num == 0:

glhe/ground_temps/ground_temp_factory.py

@@ -3,7 +3,7 @@
 from glhe.ground_temps.two_harmonic import TwoHarmonic
 
 
-def make_ground_temp_model(inputs):
+def make_ground_temp_model(inputs: dict):
     """
     Factory method to make ground temperature model objects
 
@@ -19,4 +19,4 @@ def make_ground_temp_model(inputs):
     elif gtm_type == 'two-harmonic':
         return TwoHarmonic(inputs)
     else:
-        raise ValueError("Ground temperature model '{}' is not valid.".format(gtm_type))
+        raise ValueError(f"Ground temperature model '{gtm_type}' is not valid.")

glhe/ground_temps/single_harmonic.py

@@ -12,14 +12,14 @@ class SingleHarmonic(BaseGroundTemp):
     stations in the United States. ASHRAE Transactions 71(1): 61-74.
     """
 
-    def __init__(self, inputs: dict) -> float:
+    def __init__(self, inputs: dict):
         self.ave_ground_temp = inputs['average-temperature']
         self.amplitude = inputs['amplitude']
         self.phase_shift = inputs['phase-shift']
         self.soil_diffusivity = inputs['soil-diffusivity']
 
     def get_temp(self, time: int, depth: float) -> float:
         term1 = -depth * sqrt(pi / (SEC_IN_YEAR * self.soil_diffusivity))
-        coeff = (2 * pi / SEC_IN_YEAR)
-        term2 = coeff * (time - self.phase_shift - (depth / 2) * sqrt(SEC_IN_YEAR / (pi * self.soil_diffusivity)))
+        c = (2 * pi / SEC_IN_YEAR)
+        term2 = c * (time - self.phase_shift - (depth / 2) * sqrt(SEC_IN_YEAR / (pi * self.soil_diffusivity)))
         return self.ave_ground_temp - self.amplitude * exp(term1) * cos(term2)

glhe/input_processor/component_types.py

@@ -1,6 +1,6 @@
 class ComponentTypes(object):
     """
-    All simulatable component types
+    All simulate-able component types
     """
     BoreholeSingleUTubeGrouted = 'SingleUTubeBHGrouted'
     ConstantFlow = 'ConstantFlow'

glhe/input_processor/input_processor.py

@@ -6,13 +6,13 @@
 from glhe.utilities.functions import load_json, lower_obj
 
 
-class InputProcessor(object):
+class InputProcessor:
 
-    def __init__(self, json_input_path: str) -> None:
+    def __init__(self, json_input_path: str):
         """
         Initialize the input processor, process input file, and store the input information.
 
-        :raises FileNotFoundError when input file not found.
+        :raises: FileNotFoundError when input file not found.
 
         :param json_input_path: input file path
         """
@@ -38,8 +38,8 @@ def validate_inputs(input_dict: dict) -> None:
 
         :param input_dict: input object
 
-        :raises ValidationError if the input object is not correct
-        :raises SchemaError is the schema is not correct
+        :raises: ValidationError if the input object is not correct
+        :raises: SchemaError is the schema is not correct
         """
 
         # shortcut

glhe/interface/response.py

@@ -1,4 +1,4 @@
-class SimulationResponse(object):
+class SimulationResponse:
     def __init__(self, time: float, time_step: float, flow_rate: float, temperature: float,
                  bh_wall_temp: float = None, hp_src_heat_rate: float = None):
         self.time = time

glhe/output_processor/output_processor.py

@@ -5,9 +5,9 @@
 import pandas as pd
 
 
-class OutputProcessor(object):
+class OutputProcessor:
 
-    def __init__(self, output_dir: str, output_name: str) -> None:
+    def __init__(self, output_dir: str, output_name: str):
         """
         Output processor manages output data
         """
@@ -40,7 +40,7 @@ def write_to_file(self) -> None:
         self.df.to_csv(self.write_path)
 
     def convert_time_to_timestamp(self) -> None:
-        """"
+        """
         Convert the 'Elapsed Time' column to a standardized date/time format.
         """
         try:

glhe/profiles/constant_flow.py

@@ -16,8 +16,8 @@ def __init__(self, inputs: dict, ip: InputProcessor, op: OutputProcessor):
         self.ip = ip
         self.op = op
 
-    def simulate_time_step(self, inputs: SimulationResponse):
+    def simulate_time_step(self, inputs: SimulationResponse) -> SimulationResponse:
         return SimulationResponse(inputs.time, inputs.time_step, self.flow_rate, inputs.temperature)
 
-    def report_outputs(self):
+    def report_outputs(self) -> dict:
         return {'{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.FlowRate): float(self.flow_rate)}

glhe/profiles/constant_load.py

@@ -19,7 +19,7 @@ def __init__(self, inputs: dict, ip: InputProcessor, op: OutputProcessor):
         self.inlet_temp = ip.init_temp()
         self.outlet_temp = ip.init_temp()
 
-    def simulate_time_step(self, inputs: SimulationResponse):
+    def simulate_time_step(self, inputs: SimulationResponse) -> SimulationResponse:
         self.inlet_temp = inputs.temperature
         flow_rate = inputs.flow_rate
 

glhe/profiles/constant_temp.py

@@ -18,9 +18,9 @@ def __init__(self, inputs: dict, ip: InputProcessor, op: OutputProcessor):
 
         self.inlet_temperature = ip.init_temp()
 
-    def simulate_time_step(self, inputs: SimulationResponse):
+    def simulate_time_step(self, inputs: SimulationResponse) -> SimulationResponse:
         return SimulationResponse(inputs.time, inputs.time_step, inputs.flow_rate, self.temperature)
 
-    def report_outputs(self):
+    def report_outputs(self) -> dict:
         return {'{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.InletTemp): self.inlet_temperature,
                 '{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.OutletTemp): self.temperature}

glhe/profiles/external_base.py

@@ -20,5 +20,5 @@ def __init__(self, path, col_num):
 
         self._interp_values = interp1d(x_range, y_range)
 
-    def get_value(self, time):
+    def get_value(self, time) -> float:
         return float(self._interp_values(time % self.max_time))

glhe/profiles/external_flow.py

@@ -25,9 +25,9 @@ def __init__(self, inputs: dict, ip: InputProcessor, op: OutputProcessor):
         # report variables
         self.flow_rate = self.get_value(0)
 
-    def simulate_time_step(self, inputs: SimulationResponse):
+    def simulate_time_step(self, inputs: SimulationResponse) -> SimulationResponse:
         self.flow_rate = self.get_value(inputs.time + inputs.time_step)
         return SimulationResponse(inputs.time, inputs.time_step, self.flow_rate, inputs.temperature)
 
-    def report_outputs(self):
+    def report_outputs(self) -> dict:
         return {'{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.FlowRate): float(self.flow_rate)}

glhe/profiles/external_load.py

@@ -26,7 +26,7 @@ def __init__(self, inputs: dict, ip: InputProcessor, op: OutputProcessor):
         self.load = self.get_value(0)
         self.outlet_temp = 0
 
-    def simulate_time_step(self, inputs: SimulationResponse):
+    def simulate_time_step(self, inputs: SimulationResponse) -> SimulationResponse:
         flow_rate = inputs.flow_rate
 
         if flow_rate == 0:
@@ -41,6 +41,6 @@ def simulate_time_step(self, inputs: SimulationResponse):
         self.outlet_temp = self.load / (flow_rate * specific_heat) + inlet_temp
         return SimulationResponse(inputs.time, inputs.time_step, inputs.flow_rate, self.outlet_temp)
 
-    def report_outputs(self):
+    def report_outputs(self) -> dict:
         return {'{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.OutletTemp): float(self.outlet_temp),
                 '{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.HeatRate): float(self.load)}

glhe/profiles/external_temps.py

@@ -25,9 +25,9 @@ def __init__(self, inputs: dict, ip: InputProcessor, op: OutputProcessor):
         # report variables
         self.outlet_temp = self.get_value(0)
 
-    def simulate_time_step(self, inputs: SimulationResponse):
+    def simulate_time_step(self, inputs: SimulationResponse) -> SimulationResponse:
         self.outlet_temp = self.get_value(inputs.time + inputs.time_step)
         return SimulationResponse(inputs.time, inputs.time_step, inputs.flow_rate, self.outlet_temp)
 
-    def report_outputs(self):
+    def report_outputs(self) -> dict:
         return {'{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.OutletTemp): float(self.outlet_temp)}

glhe/profiles/flow_factory.py

@@ -1,17 +1,14 @@
-from typing import Union
-
 from glhe.input_processor.input_processor import InputProcessor
 from glhe.output_processor.output_processor import OutputProcessor
 from glhe.profiles.constant_flow import ConstantFlow
 from glhe.profiles.external_flow import ExternalFlow
 
 
-def make_flow_profile(inputs: dict, ip: InputProcessor, op: OutputProcessor) -> Union[ConstantFlow,
-                                                                                      ExternalFlow]:
+def make_flow_profile(inputs: dict, ip: InputProcessor, op: OutputProcessor) -> ConstantFlow | ExternalFlow:
     load_profile_type = inputs['flow-profile-type']
     if load_profile_type == 'constant':
         return ConstantFlow(inputs, ip, op)
     elif load_profile_type == 'external':
         return ExternalFlow(inputs, ip, op)
     else:
-        raise ValueError("Flow profile '{}' is not valid.".format(load_profile_type))
+        raise ValueError(f"Flow profile '{load_profile_type}' is not valid.")

glhe/profiles/load_factory.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 from glhe.input_processor.input_processor import InputProcessor
 from glhe.output_processor.output_processor import OutputProcessor
 from glhe.profiles.constant_load import ConstantLoad
@@ -9,9 +7,8 @@
 from glhe.profiles.synthetic_load import SyntheticLoad
 
 
-def make_load_profile(inputs: dict, ip: InputProcessor, op: OutputProcessor) -> Union[ConstantLoad, PulseLoad,
-                                                                                      ExternalLoad, SinusoidLoad,
-                                                                                      SyntheticLoad]:
+def make_load_profile(inputs: dict, ip: InputProcessor,
+                      op: OutputProcessor) -> ConstantLoad | PulseLoad | ExternalLoad | SinusoidLoad | SyntheticLoad:
     load_profile_type = inputs['load-profile-type']
     if load_profile_type == 'constant':
         return ConstantLoad(inputs, ip, op)

glhe/profiles/pulse_load.py

@@ -19,7 +19,7 @@ def __init__(self, inputs, ip, op):
         # report variables
         self.outlet_temp = 0
 
-    def simulate_time_step(self, inputs: SimulationResponse):
+    def simulate_time_step(self, inputs: SimulationResponse) -> SimulationResponse:
 
         if self.start_time <= inputs.time + inputs.time_step < self.end_time:
             flow_rate = inputs.flow_rate
@@ -36,6 +36,6 @@ def simulate_time_step(self, inputs: SimulationResponse):
             self.load = 0
             return inputs
 
-    def report_outputs(self):
+    def report_outputs(self) -> dict:
         return {'{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.OutletTemp): float(self.outlet_temp),
                 '{:s}:{:s}:{:s}'.format(self.Type, self.name, ReportTypes.HeatRate): float(self.load)}