Merge pull request #108 from NREL/ndr/90-mypy-errors

mypy errors

Author: nreinicke

.github/workflows/tests.yaml

@@ -18,17 +18,20 @@ jobs:
         PYTHON: ${{ matrix.python-version }}
 
       steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
 
       - name: set up python ${{ matrix.python-version }}
-        uses: actions/setup-python@v1
+        uses: actions/setup-python@v4
         with:
           python-version: ${{ matrix.python-version }}
 
       - name: Install package 
         run: |
           pip install ".[dev]" 
 
+      - name: Run mypy
+        run: mypy . --ignore-missing-imports 
+
       - name: Run black
         run: |
           black nrel tests --check

examples/cosim_custom_dispatcher.py

@@ -7,9 +7,7 @@
 
 from nrel.hive import package_root
 from nrel.hive.dispatcher.instruction.instructions import IdleInstruction
-from nrel.hive.dispatcher.instruction_generator.instruction_generator import (
-    InstructionGenerator,
-)
+from nrel.hive.dispatcher.instruction_generator.instruction_generator import InstructionGenerator
 from nrel.hive.runner.runner_payload_ops import (
     get_instruction_generator,
     update_instruction_generator,
@@ -59,9 +57,7 @@ def run():
         crank_result = hc.crank(rp, 10)
 
         # get dispatcher to update it
-        dispatcher: CustomDispatcher = get_instruction_generator(
-            crank_result.runner_payload, CustomDispatcher
-        )
+        dispatcher = get_instruction_generator(crank_result.runner_payload, CustomDispatcher)
 
         # apply some change to the state of the dispatcher
         updated_dispatcher = dispatcher.new_random_state()

mypy.ini

@@ -6,4 +6,4 @@ exclude = (?x)(
     docs/    
     | build/  
     | dist/   
-  )
+  )

nrel/hive/__init__.py

@@ -29,9 +29,7 @@
 from nrel.hive.config import HiveConfig
 from nrel.hive.dispatcher import *
 from nrel.hive.state.simulation_state.update.update import Update
-from nrel.hive.state.simulation_state.update.step_simulation import (
-    StepSimulation,
-)
+from nrel.hive.state.simulation_state.update.step_simulation import StepSimulation
 
 
 def package_root() -> Path:

nrel/hive/app/hive_cosim.py

@@ -1,33 +1,29 @@
 import functools as ft
 from pathlib import Path
-from typing import Tuple, NamedTuple, Optional
+from typing import Tuple, NamedTuple, Optional, TypeVar
 
 import pandas as pd
 from pandas import DataFrame
 from tqdm import tqdm
 
 from nrel.hive import Update
-from nrel.hive.dispatcher.instruction_generator.instruction_generator import (
-    InstructionGenerator,
-)
-from nrel.hive.dispatcher.instruction_generator.charging_fleet_manager import (
-    ChargingFleetManager,
-)
+from nrel.hive.dispatcher.instruction_generator.instruction_generator import InstructionGenerator
+from nrel.hive.dispatcher.instruction_generator.charging_fleet_manager import ChargingFleetManager
 from nrel.hive.dispatcher.instruction_generator.dispatcher import Dispatcher
 from nrel.hive.initialization.load import load_simulation
 from nrel.hive.model.sim_time import SimTime
 from nrel.hive.reporting import reporter_ops
-from nrel.hive.reporting.handler.vehicle_charge_events_handler import (
-    VehicleChargeEventsHandler,
-)
+from nrel.hive.reporting.handler.vehicle_charge_events_handler import VehicleChargeEventsHandler
 from nrel.hive.runner import RunnerPayload
 from nrel.hive.util import SimulationStateError
 from nrel.hive.util.fp import throw_on_failure
 
+T = TypeVar("T", bound=InstructionGenerator)
+
 
 def load_scenario(
     scenario_file: Path,
-    custom_instruction_generators: Optional[Tuple[InstructionGenerator, ...]] = None,
+    custom_instruction_generators: Optional[Tuple[T, ...]] = None,
 ) -> RunnerPayload:
     """
     load a HIVE scenario from file and return the initial simulation state
@@ -49,13 +45,13 @@ def load_scenario(
             ChargingFleetManager(env.config.dispatcher),
             Dispatcher(env.config.dispatcher),
         )
+        update = Update.build(env.config, instruction_generators)
     else:
-        instruction_generators = custom_instruction_generators
+        update = Update.build(env.config, custom_instruction_generators)
 
     # add a specialized Reporter handler that catches vehicle charge events
     env.reporter.add_handler(VehicleChargeEventsHandler())
 
-    update = Update.build(env.config, instruction_generators)
     initial_payload = RunnerPayload(sim, env, update)
 
     return initial_payload

nrel/hive/app/run.py

@@ -10,9 +10,7 @@
 import pkg_resources
 import yaml
 
-from nrel.hive.dispatcher.instruction_generator.charging_fleet_manager import (
-    ChargingFleetManager,
-)
+from nrel.hive.dispatcher.instruction_generator.charging_fleet_manager import ChargingFleetManager
 from nrel.hive.dispatcher.instruction_generator.dispatcher import Dispatcher
 from nrel.hive.initialization.load import load_simulation
 from nrel.hive.reporting import reporter_ops

nrel/hive/app/run_batch.py

@@ -72,7 +72,12 @@ def run() -> int:
     sim_args = [SimArgs(f, i) for i, f in enumerate(config.scenario_files)]
 
     # check to make sure we don't exceed system CPU
-    max_cpu = os.cpu_count()
+    os_cpu = os.cpu_count()
+
+    if os_cpu is None:
+        max_cpu = 1
+    else:
+        max_cpu = os_cpu
 
     if len(config.scenario_files) > max_cpu:
         cpu = max_cpu

nrel/hive/app/run_tune.py

@@ -7,9 +7,7 @@
 import ray
 from ray import tune
 
-from nrel.hive.dispatcher.instruction_generator.charging_fleet_manager import (
-    ChargingFleetManager,
-)
+from nrel.hive.dispatcher.instruction_generator.charging_fleet_manager import ChargingFleetManager
 from nrel.hive.dispatcher.instruction_generator.dispatcher import Dispatcher
 from nrel.hive.initialization.load import load_simulation
 from nrel.hive.reporting.reporter import Reporter
@@ -54,7 +52,7 @@ def _scoring_function(payload: RunnerPayload) -> float:
         print("VEHICLES ", payload.s.vehicles.values())
         return score
 
-    def _setup(self, config: Dict[str:int]):
+    def _setup(self, config: Dict[str, int]):
         scenarios = {
             1: "denver_demo.yaml",
             2: "denver_demo_constrained_charging.yaml",

nrel/hive/config/config_builder.py

@@ -12,8 +12,8 @@ def build(
         cls,
         default_config: Dict,
         required_config: Tuple[str, ...],
-        config_constructor: Callable[[Dict], Union[Exception, T]],
-        config: Dict = None,
+        config_constructor: Callable[[Dict], T],
+        config: Optional[Dict] = None,
     ) -> T:
         """
         constructs a Config from a configuration Dict

nrel/hive/config/dispatcher_config.py

@@ -3,9 +3,7 @@
 from typing import NamedTuple, Dict, Union, Tuple, Optional
 
 from nrel.hive.config.config_builder import ConfigBuilder
-from nrel.hive.dispatcher.instruction_generator.charging_search_type import (
-    ChargingSearchType,
-)
+from nrel.hive.dispatcher.instruction_generator.charging_search_type import ChargingSearchType
 from nrel.hive.util.units import Ratio, Seconds, Kilometers
 
 
@@ -41,12 +39,12 @@ def build(cls, config: Optional[Dict] = None) -> DispatcherConfig:
         )
 
     @classmethod
-    def from_dict(cls, d: Dict) -> Union[IOError, DispatcherConfig]:
+    def from_dict(cls, d: Dict) -> DispatcherConfig:
         try:
             d["valid_dispatch_states"] = tuple(s.lower() for s in d["valid_dispatch_states"])
             d["charging_search_type"] = ChargingSearchType.from_string(d["charging_search_type"])
         except ValueError:
-            return IOError("valid_dispatch_states and active_states must be in a list format")
+            raise IOError("valid_dispatch_states and active_states must be in a list format")
 
         return DispatcherConfig(**d)
 

nrel/hive/config/hive_config.py

@@ -34,7 +34,7 @@ class HiveConfig(NamedTuple):
     def build(
         cls,
         scenario_file_path: Path,
-        config: Dict = None,
+        config: Optional[Dict] = None,
         output_suffix: Optional[str] = datetime.now().strftime("%Y-%m-%d_%H-%M-%S"),
     ) -> Union[Exception, HiveConfig]:
         """
@@ -55,7 +55,7 @@ def build(
     @classmethod
     def from_dict(
         cls, d: Dict, scenario_file_path: Path, output_suffix: Optional[str]
-    ) -> Union[Exception, HiveConfig]:
+    ) -> HiveConfig:
         # collect the global hive configuration
         global_config = fs.global_hive_config_search()
 
@@ -142,7 +142,7 @@ def asdict(self) -> Dict:
         out_dict["cache"] = cache
 
         for name, config in self._asdict().items():
-            if issubclass(config.__class__, Tuple):
+            if issubclass(config.__class__, tuple):
                 out_dict[name] = config.asdict()
             else:
                 out_dict[name] = config

nrel/hive/config/input.py

@@ -3,7 +3,7 @@
 import hashlib
 import logging
 from pathlib import Path
-from typing import NamedTuple, Tuple, Dict, Optional
+from typing import NamedTuple, Tuple, Dict, Optional, Union
 
 from nrel.hive.config.config_builder import ConfigBuilder
 from nrel.hive.util import fs
@@ -19,14 +19,14 @@ class Input(NamedTuple):
     bases_file: str
     stations_file: str
     mechatronics_file: str
-    schedules_file: Optional[str]
-    chargers_file: Optional[str]
-    road_network_file: Optional[str]
-    geofence_file: Optional[str]
-    rate_structure_file: Optional[str]
-    charging_price_file: Optional[str]
-    demand_forecast_file: Optional[str]
-    fleets_file: Optional[str]
+    chargers_file: str
+    schedules_file: Optional[str] = None
+    road_network_file: Optional[str] = None
+    geofence_file: Optional[str] = None
+    rate_structure_file: Optional[str] = None
+    charging_price_file: Optional[str] = None
+    demand_forecast_file: Optional[str] = None
+    fleets_file: Optional[str] = None
 
     @classmethod
     def default_config(cls) -> Dict:
@@ -124,7 +124,7 @@ def from_dict(cls, d: Dict, scenario_file_path: Path, cache: Optional[Dict]) ->
             else None
         )
 
-        input = {
+        input_config = {
             "scenario_directory": str(scenario_directory),
             "scenario_file": scenario_file,
             "vehicles_file": vehicles_file,
@@ -147,11 +147,27 @@ def from_dict(cls, d: Dict, scenario_file_path: Path, cache: Optional[Dict]) ->
             for (
                 name,
                 path,
-            ) in input.items():  # input_config.asdict(absolute_paths=True).items():
+            ) in input_config.items():  # input_config.asdict(absolute_paths=True).items():
                 if path:
                     cls._check_md5_checksum(path, cache[name])
 
-        return Input(**input)
+        return Input(
+            scenario_directory=str(scenario_directory),
+            scenario_file=scenario_file,
+            vehicles_file=vehicles_file,
+            requests_file=requests_file,
+            bases_file=bases_file,
+            stations_file=stations_file,
+            schedules_file=schedules_file,
+            chargers_file=chargers_file,
+            mechatronics_file=mechatronics_file,
+            road_network_file=road_network_file,
+            geofence_file=geofence_file,
+            rate_structure_file=rate_structure_file,
+            charging_price_file=charging_price_file,
+            demand_forecast_file=demand_forecast_file,
+            fleets_file=fleets_file,
+        )
 
     @staticmethod
     def _check_md5_checksum(filepath: str, existing_md5_sum: str):

nrel/hive/config/network.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import NamedTuple, Dict, Union, Tuple
+from typing import NamedTuple, Dict, Optional, Tuple
 
 from nrel.hive.config.config_builder import ConfigBuilder
 
@@ -18,7 +18,7 @@ def required_config(cls) -> Tuple[str, ...]:
         return ()
 
     @classmethod
-    def build(cls, config: Dict = None) -> Union[Exception, Network]:
+    def build(cls, config: Optional[Dict] = None) -> Network:
         return ConfigBuilder.build(
             default_config=cls.default_config(),
             required_config=cls.required_config(),

nrel/hive/config/sim.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import NamedTuple, Dict, Union, Tuple
+from typing import NamedTuple, Dict, Optional, Union, Tuple
 
 from nrel.hive.config.config_builder import ConfigBuilder
 from nrel.hive.model.vehicle.schedules.schedule_type import ScheduleType
@@ -31,7 +31,7 @@ def required_config(cls) -> Tuple[str, ...]:
         )
 
     @classmethod
-    def build(cls, config: Dict = None) -> Union[IOError, Sim]:
+    def build(cls, config: Optional[Dict] = None) -> Sim:
         return ConfigBuilder.build(
             default_config=cls.default_config(),
             required_config=cls.required_config(),
@@ -40,14 +40,10 @@ def build(cls, config: Dict = None) -> Union[IOError, Sim]:
         )
 
     @classmethod
-    def from_dict(cls, d: Dict) -> Union[IOError, Sim]:
+    def from_dict(cls, d: Dict) -> Sim:
         start_time = SimTime.build(d["start_time"])
-        if isinstance(start_time, IOError):
-            return start_time
 
         end_time = SimTime.build(d["end_time"])
-        if isinstance(end_time, IOError):
-            return end_time
 
         schedule_type = ScheduleType.from_string(d["schedule_type"])
 

nrel/hive/dispatcher/forecaster/__init__.py

@@ -1,5 +1,3 @@
 from nrel.hive.dispatcher.forecaster.basic_forecaster import BasicForecaster
 from nrel.hive.dispatcher.forecaster.forecast import Forecast, ForecastType
-from nrel.hive.dispatcher.forecaster.forecaster_interface import (
-    ForecasterInterface,
-)
+from nrel.hive.dispatcher.forecaster.forecaster_interface import ForecasterInterface

nrel/hive/dispatcher/forecaster/basic_forecaster.py

@@ -2,15 +2,16 @@
 from dataclasses import dataclass
 
 from pathlib import Path
-from typing import Tuple, NamedTuple
+from typing import Tuple, NamedTuple, TYPE_CHECKING
 
 from nrel.hive.model.sim_time import SimTime
 from nrel.hive.dispatcher.forecaster.forecast import Forecast, ForecastType
-from nrel.hive.dispatcher.forecaster.forecaster_interface import (
-    ForecasterInterface,
-)
+from nrel.hive.dispatcher.forecaster.forecaster_interface import ForecasterInterface
 from nrel.hive.util.iterators import DictReaderStepper
 
+if TYPE_CHECKING:
+    from hive.state.simulation_state.simulation_state import SimulationState
+
 
 @dataclass(frozen=True)
 class BasicForecaster(ForecasterInterface):
@@ -38,10 +39,13 @@ def build(cls, demand_forecast_file: str) -> BasicForecaster:
         if error:
             raise error
         else:
+            if reader is None:
+                raise Exception("No reader supplied by the DictReaderStepper")
+
             return BasicForecaster(reader)
 
     def generate_forecast(
-        self, simulation_state: "SimulationState"
+        self, simulation_state: SimulationState
     ) -> Tuple[BasicForecaster, Forecast]:
         """
         Generate fleet targets to be consumed by the dispatcher.