typing: adding types to extensions outside of breakeven

Author: ParticularlyPythonicBS

pyproject.toml

@@ -141,7 +141,7 @@ python_version = "3.12"
 mypy_path = "stubs"
 
 # Exclude specific directories from type checking will try to add them back gradually
-exclude = "(?x)(^temoa/extensions/|^temoa/utilities/|^stubs/)"
+exclude = "(?x)(^temoa/utilities/|^stubs/|^temoa/extensions/breakeven/)"
 
 # Strict typing for our own code
 disallow_untyped_defs = true

temoa/extensions/get_comm_tech.py

@@ -1,22 +1,24 @@
+from __future__ import annotations
+
 import getopt
 import os
 import re
 import sqlite3
 import sys
 from collections import OrderedDict
+from typing import Any
 
 
-def get_tperiods(inp_f):
+def get_tperiods(inp_f: str) -> dict[str, list[int]]:
     file_ty = re.search(r'(\w+)\.(\w+)\b', inp_f)  # Extract the input filename and extension
 
     if not file_ty:
-        raise 'The file type %s is not recognized.' % inp_f
+        raise Exception(f'The file type {inp_f} is not recognized.')
 
     elif file_ty.group(2) not in ('db', 'sqlite', 'sqlite3', 'sqlitedb'):
         raise Exception('Please specify a database for finding scenarios')
 
-    periods_list = {}
-    periods_set = set()
+    periods_list: dict[str, list[int]] = {}
 
     con = sqlite3.connect(inp_f)
     cur = con.cursor()  # a database cursor is a control structure that enables traversal over
@@ -30,7 +32,7 @@ def get_tperiods(inp_f):
         x.append(row[0])
     for y in x:
         cur.execute(
-            "SELECT DISTINCT period FROM output_flow_out WHERE scenario is '" + str(y) + "'"
+            f"SELECT DISTINCT period FROM output_flow_out WHERE scenario is '{y}'"
         )
         periods_list[y] = []
         for per in cur:
@@ -42,17 +44,16 @@ def get_tperiods(inp_f):
     return dict(OrderedDict(sorted(periods_list.items(), key=lambda x: x[1])))
 
 
-def get_scenario(inp_f):
+def get_scenario(inp_f: str) -> dict[str, str]:
     file_ty = re.search(r'(\w+)\.(\w+)\b', inp_f)  # Extract the input filename and extension
 
     if not file_ty:
-        raise 'The file type %s is not recognized.' % inp_f
+        raise Exception(f'The file type {inp_f} is not recognized.')
 
     elif file_ty.group(2) not in ('db', 'sqlite', 'sqlite3', 'sqlitedb'):
         raise Exception('Please specify a database for finding scenarios')
 
-    scene_list = {}
-    scene_set = set()
+    scene_list: dict[str, str] = {}
 
     con = sqlite3.connect(inp_f)
     cur = con.cursor()  # a database cursor is a control structure that enables traversal over
@@ -70,9 +71,9 @@ def get_scenario(inp_f):
     return dict(OrderedDict(sorted(scene_list.items(), key=lambda x: x[1])))
 
 
-def get_comm(inp_f, db_dat):
-    comm_list = {}
-    comm_set = set()
+def get_comm(inp_f: str, db_dat: bool) -> OrderedDict[str, str]:
+    comm_list: dict[str, str] = {}
+    comm_set: set[str] = set()
     is_query_empty = False
 
     if not db_dat:
@@ -138,9 +139,9 @@ def get_comm(inp_f, db_dat):
     return OrderedDict(sorted(comm_list.items(), key=lambda x: x[1]))
 
 
-def get_tech(inp_f, db_dat):
-    tech_list = {}
-    tech_set = set()
+def get_tech(inp_f: str, db_dat: bool) -> OrderedDict[str, str]:
+    tech_list: dict[str, str] = {}
+    tech_set: set[str] = set()
     is_query_empty = False
 
     if not db_dat:
@@ -199,13 +200,13 @@ def get_tech(inp_f, db_dat):
     return OrderedDict(sorted(tech_list.items(), key=lambda x: x[1]))
 
 
-def is_db_overwritten(db_file, inp_dat_file):
+def is_db_overwritten(db_file: str, inp_dat_file: str) -> bool:
     if os.path.basename(db_file) == '0':
         return False
 
     try:
         con = sqlite3.connect(db_file)
-    except:
+    except Exception:
         return False
     cur = con.cursor()  # A database cursor enables traversal over DB records
     con.text_factory = str  # This ensures data is explored with UTF-8 encoding
@@ -214,15 +215,15 @@ def is_db_overwritten(db_file, inp_dat_file):
     # IF output file is empty database.
     cur.execute('SELECT * FROM Technology')
     is_db_empty = False  # False for empty db file
-    for elem in cur:
+    for _ in cur:
         is_db_empty = True  # True for non-empty db file
         break
     # This file could be schema with populated results from previous run. Or it could be a normal
     # db file.
     if is_db_empty:
         cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='input_file';")
         does_input_file_table_exist = False
-        for i in cur:  # This means that the 'input_file' table exists in db.
+        for _ in cur:  # This means that the 'input_file' table exists in db.
             does_input_file_table_exist = True
         if does_input_file_table_exist:  # This block distinguishes normal database from schema.
             # This is schema file.
@@ -247,7 +248,7 @@ def is_db_overwritten(db_file, inp_dat_file):
     return False
 
 
-def help_user():
+def help_user() -> None:
     print(
         """Use as:
     python get_comm_tech.py -i (or --input) <input filename>
@@ -259,8 +260,8 @@ def help_user():
     )
 
 
-def get_info(inputs):
-    inp_file = None
+def get_info(inputs: dict[str, str]) -> Any:
+    inp_file: str | None = None
     tech_flag = False
     comm_flag = False
     scene = False
@@ -317,8 +318,8 @@ def get_info(inputs):
 
     else:
         print(
-            'The input file type %s is not recognized. Please specify a database or a text file.'
-            % inp_file
+            f'The input file type {inp_file} is not recognized. Please specify a database '
+            'or a text file.'
         )
         sys.exit(2)
 

temoa/extensions/method_of_morris/morris.py

@@ -1,28 +1,28 @@
-# from __future__ import division
-import time
+from __future__ import annotations
+
+import csv
+import sqlite3
 from importlib import resources
 from pathlib import Path
+from typing import Any
 
+from joblib import Parallel, delayed  # type: ignore[import-untyped]
+from numpy import array
 from pyomo.dataportal import DataPortal
+from SALib.analyze import morris  # type: ignore[import-untyped]
+from SALib.sample.morris import sample  # type: ignore[import-untyped]
+from SALib.util import compute_groups_matrix, read_param_file  # type: ignore[import-untyped]
 
 from temoa._internal import run_actions
 from temoa._internal.table_writer import TableWriter
 from temoa.core.config import TemoaConfig
 from temoa.data_io.hybrid_loader import HybridLoader
 
-start_time = time.time()
-import sqlite3
-
-from joblib import Parallel, delayed
-from numpy import array
-from SALib.analyze import morris
-from SALib.sample.morris import sample
-from SALib.util import compute_groups_matrix, read_param_file
-
 seed = 42
 
 
-def evaluate(param_names, param_values, data: dict, k):
+def evaluate(param_names: dict[int, list[Any]], param_values: Any,
+            data: dict[str, Any], k: int) -> list[Any]:
     m = len(param_values)
     for j in range(0, m):
         names = param_names[j]
@@ -51,16 +51,16 @@ def evaluate(param_names, param_values, data: dict, k):
     cur.execute('SELECT * FROM output_objective')
     output_query = cur.fetchall()
     for row in output_query:
-        Y_OF = row[-1]
+        y_of = row[-1]
     cur.execute("SELECT emis_comm, SUM(emission) FROM output_emissionn WHERE emis_comm='co2'")
     output_query = cur.fetchall()
     for row in output_query:
-        Y_CumulativeCO2 = row[-1]
-    Morris_Objectives = []
-    Morris_Objectives.append(Y_OF)
-    Morris_Objectives.append(Y_CumulativeCO2)
+        y_cumulative_co2 = row[-1]
+    morris_objectives = []
+    morris_objectives.append(y_of)
+    morris_objectives.append(y_cumulative_co2)
     con.close()
-    return Morris_Objectives
+    return morris_objectives
 
 
 morris_root = Path(__file__).parent
@@ -137,7 +137,7 @@ def evaluate(param_names, param_values, data: dict, k):
             file.write('\n')
 
         # load a data portal, retrieve the data dict for the problem
-        config = TemoaConfig.build_config(config_file=config_path, output_path='.')
+        config = TemoaConfig.build_config(config_file=config_path, output_path=Path('.'))
         loader = HybridLoader(db_connection=con, config=config)
         loader.load_data_portal()
         data = loader.data
@@ -157,7 +157,7 @@ def evaluate(param_names, param_values, data: dict, k):
 )
 Morris_Objectives = array(Morris_Objectives)
 print(Morris_Objectives)
-Si_OF = morris.analyze(
+si_of = morris.analyze(
     problem,
     param_values,
     Morris_Objectives[:, 0],
@@ -168,7 +168,7 @@ def evaluate(param_names, param_values, data: dict, k):
     seed=seed + 1,
 )
 
-Si_CumulativeCO2 = morris.analyze(
+si_cumulative_co2 = morris.analyze(
     problem,
     param_values,
     Morris_Objectives[:, 1],
@@ -189,19 +189,18 @@ def evaluate(param_names, param_values, data: dict, k):
 for j in list(range(number_of_groups)):
     print(
         '{:30} {:10.3f} {:10.3f} {:15.3f} {:10.3f}'.format(
-            Si_OF['names'][j],
-            Si_OF['mu_star'][j],
-            Si_OF['mu'][j],
-            Si_OF['mu_star_conf'][j],
-            Si_OF['sigma'][j],
+            si_of['names'][j],
+            si_of['mu_star'][j],
+            si_of['mu'][j],
+            si_of['mu_star_conf'][j],
+            si_of['sigma'][j],
         )
     )
-import csv
 
-line1 = Si_OF['mu_star']
-line2 = Si_OF['mu_star_conf']
-line3 = Si_CumulativeCO2['mu_star']
-line4 = Si_CumulativeCO2['mu_star_conf']
+line1 = si_of['mu_star']
+line2 = si_of['mu_star_conf']
+line3 = si_cumulative_co2['mu_star']
+line4 = si_cumulative_co2['mu_star_conf']
 with open('MMResults.csv', 'w') as f:
     writer = csv.writer(f, delimiter=',')
     writer.writerow(unique_group_names)
@@ -211,6 +210,3 @@ def evaluate(param_names, param_values, data: dict, k):
     writer.writerow('Cumulative CO2 Emissions')
     writer.writerow(line3)
     writer.writerow(line4)
-
-f.close
-print('--- %s seconds ---' % (time.time() - start_time))

temoa/extensions/method_of_morris/morris_evaluate.py

@@ -2,20 +2,25 @@
 This module contains the core "evaluation" function for Method Of Morris.  It needs to be isolated
 (outside of class) to enable parallelization.
 """
+from __future__ import annotations
 
 import logging
 import sqlite3
 import sys
 from logging.handlers import QueueHandler
+from typing import TYPE_CHECKING, Any
 
 from pyomo.dataportal import DataPortal
 
 from temoa._internal import run_actions
 from temoa._internal.table_writer import TableWriter
-from temoa.core.config import TemoaConfig
 
+if TYPE_CHECKING:
+    from temoa.core.config import TemoaConfig
 
-def configure_worker_logger(log_queue, log_level):
+
+
+def configure_worker_logger(log_queue: Any, log_level: int) -> logging.Logger:
     """configure the logger"""
     worker_logger = logging.getLogger('MM evaluate')
     if not worker_logger.hasHandlers():
@@ -30,7 +35,8 @@ def configure_worker_logger(log_queue, log_level):
     return worker_logger
 
 
-def evaluate(param_info, mm_sample, data, i, config: TemoaConfig, log_queue, log_level):
+def evaluate(param_info: dict[int, list[Any]], mm_sample: Any, data: dict[str, Any],
+            i: int, config: TemoaConfig, log_queue: Any, log_level: int) -> list[float]:
     """
     Run model for params provided and return objective value and emission value
     Note:  This function needs to be a static instance to enable the parallel
@@ -49,19 +55,14 @@ def evaluate(param_info, mm_sample, data, i, config: TemoaConfig, log_queue, log
     log_entry = ['']
     for j in range(0, len(mm_sample)):
         param_name, *set_idx, _ = param_info[j]
-        set_idx = tuple(set_idx)
+        set_idx_tuple = tuple(set_idx)
         # tweak the parameter
         if data.get(param_name) is None:
             raise ValueError(f'Unrecognized parameter: {param_name}')
-        if data[param_name].get(set_idx) is None:
+        if data[param_name].get(set_idx_tuple) is None:
             raise ValueError('index mismatch from data read-in')
-        data[param_name][set_idx] = mm_sample[j]
-        setting_entry = 'run # %d:  Setting param %s[%s] to value:  %f' % (
-            i + 1,
-            param_name,
-            set_idx,
-            mm_sample[j],
-        )
+        data[param_name][set_idx_tuple] = mm_sample[j]
+        setting_entry = 'run # %d:  Setting param %s[%s] to value:  %f'
         log_entry.append(setting_entry)
     logger.debug('\n  '.join(log_entry))
 
@@ -87,23 +88,23 @@ def evaluate(param_info, mm_sample, data, i, config: TemoaConfig, log_queue, log
             'Multiple outputs found in Objective table matching scenario name.  Coding error.'
         )
     else:
-        Y_OF = output_query[0][0]
+        y_of = output_query[0][0]
     cur.execute(
         "SELECT SUM(emission) FROM output_emission WHERE emis_comm='co2' AND scenario=?",
         (scenario_name,),
     )
     output_query = cur.fetchall()
     if len(output_query) == 0:
-        Y_CumulativeCO2 = 0.0
+        y_cumulative_co2 = 0.0
     elif len(output_query) > 1:
         raise RuntimeError(
             'Multiple outputs found in output_emissions table matching scenario name.  Coding '
             'error.'
         )
     else:
-        Y_CumulativeCO2 = output_query[0][0]
-    morris_objectives = [float(Y_OF), float(Y_CumulativeCO2)]
-    logger.info('Finished MM evaluation # %d with OBJ value: %0.2f ', i + 1, Y_OF)
+        y_cumulative_co2 = output_query[0][0]
+    morris_objectives = [float(y_of), float(y_cumulative_co2)]
+    logger.info('Finished MM evaluation # %d with OBJ value: %0.2f ', i + 1, y_of)
     if not config.silent:
         sys.stdout.write(f'Completed MM run {i + 1}\n')
         sys.stdout.flush()