Skip to content

typing: adding types to extensions #251

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ParticularlyPythonicBS merged 2 commits into from
Mar 24, 2026
Merged

typing: adding types to extensions #251

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ca4afcc
typing: adding types to extensions outside of breakeven
ParticularlyPythonicBS Jan 12, 2026
6a0ef52
improving robustness and type safety of extensions
ParticularlyPythonicBS Mar 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,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/)"

Comment on lines 146 to 148
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot Jan 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Tighten the mypy exclude regex to match breakeven both with and without trailing slash.

Current regex only matches ^temoa/extensions/breakeven/… paths.

Proposed tweak 
-exclude = "(?x)(^temoa/utilities/|^stubs/|^temoa/extensions/breakeven/)"
+exclude = "(?x)(^temoa/utilities/|^stubs/|^temoa/extensions/breakeven(?:/|$))"
🤖 Prompt for AI Agents 
In @pyproject.toml around lines 143 - 145, The mypy exclude regex in
pyproject.toml only matches paths with a trailing slash for the breakeven
directory; update the exclude entry (the string assigned to exclude) so the
^temoa/extensions/breakeven part allows either a trailing slash or end-of-string
(i.e., make the trailing slash optional using a group like a non-capturing
(?:/|$) after breakeven) so both breakeven and breakeven/ are excluded.

# Strict typing for our own code
disallow_untyped_defs = true
Expand Down
49 changes: 25 additions & 24 deletions temoa/extensions/get_comm_tech.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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')
Comment on lines +12 to 19
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot Jan 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial

Prefer ValueError/custom exceptions over generic Exception for argument validation.

Also applies to: 47-55

🧰 Tools
🪛 Ruff (0.14.10)

16-16: Create your own exception

(TRY002)

16-16: Avoid specifying long messages outside the exception class

(TRY003)

19-19: Create your own exception

(TRY002)

19-19: Avoid specifying long messages outside the exception class

(TRY003)

🤖 Prompt for AI Agents 
In @temoa/extensions/get_comm_tech.py around lines 12 - 19, The function
get_tperiods uses generic Exception for input validation; replace those raises
with ValueError (or a module-specific custom exception) to clearly signal
invalid arguments: update the two raises in get_tperiods (the "file type not
recognized" and the "Please specify a database..." cases) to raise ValueError
(or a defined custom exception class) and apply the same change to the similar
checks at the other location referenced (the block around lines 47-55) so
argument errors consistently use the appropriate exception type.


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
Expand All @@ -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) + "'"
'SELECT DISTINCT period FROM output_flow_out WHERE scenario = ?', (y,)
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
periods_list[y] = []
for per in cur:
Expand All @@ -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
Expand All @@ -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()
Comment on lines +74 to +76
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot Jan 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Boolean positional params (db_dat: bool) are hard to read at call sites—prefer keyword-only usage.

Also applies to: 142-145

🧰 Tools
🪛 Ruff (0.14.10)

74-74: Boolean-typed positional argument in function definition

(FBT001)

🤖 Prompt for AI Agents 
In @temoa/extensions/get_comm_tech.py around lines 74 - 76, Change the get_comm
signature to require db_dat as a keyword-only argument (e.g., def
get_comm(inp_f: str, *, db_dat: bool) -> OrderedDict[str, str]) so callers must
use db_dat=...; apply the same change to the other functions in this file that
accept db_dat (the ones around lines 142-145) and update any local callers to
pass db_dat by name rather than position. Ensure only the signature is changed
(keep type hints and return type) and run tests to catch any missed call sites.

is_query_empty = False

if not db_dat:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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 sqlite3.Error:
return False
Comment thread
coderabbitai[bot] marked this conversation as resolved.
cur = con.cursor() # A database cursor enables traversal over DB records
con.text_factory = str # This ensures data is explored with UTF-8 encoding
Expand All @@ -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.
Expand All @@ -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>
Expand All @@ -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
Comment on lines +263 to +264
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot Mar 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major

get_info() still exports Any.

This dispatcher only returns period lists or string-keyed mappings, so Any throws away most of the benefit of the typing added in this PR. A small union or alias here would keep callers type-safe without changing runtime behavior.

🧰 Tools
🪛 Ruff (0.15.6)

[warning] 263-338: Missing explicit return at the end of function able to return non-None value

Add explicit return statement

(RET503)

[warning] 263-263: Too many branches (22 > 12)

(PLR0912)

[warning] 263-263: Dynamically typed expressions (typing.Any) are disallowed in get_info

(ANN401)

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@temoa/extensions/get_comm_tech.py` around lines 263 - 264, get_info currently
returns typing.Any; constrain it to the actual shapes returned (period lists or
string-keyed mappings) by replacing Any with a precise union or alias (e.g.,
define PeriodList = list[str] and PeriodMap = dict[str, PeriodList] and use ->
PeriodList | PeriodMap or a single alias like InfoResult). Update the get_info
signature to use that new type and add the small alias(s) near the top of
temoa/extensions/get_comm_tech.py so callers get proper type information without
changing runtime behavior.

tech_flag = False
comm_flag = False
scene = False
Expand Down Expand Up @@ -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)

Expand Down
176 changes: 86 additions & 90 deletions temoa/extensions/method_of_morris/morris.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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]:
Comment on lines +24 to +25
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot Jan 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial

Unused parameter k should be prefixed with underscore.

The k parameter is not used in the function body. Per Python convention, prefix it with _ to indicate it's intentionally unused.

♻️ Proposed fix 
 def evaluate(param_names: dict[int, list[Any]], param_values: Any,
-            data: dict[str, Any], k: int) -> list[Any]:
+            data: dict[str, Any], _k: int) -> list[Any]:
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def evaluate(param_names: dict[int, list[Any]], param_values: Any,
data: dict[str, Any], k: int) -> list[Any]:
def evaluate(param_names: dict[int, list[Any]], param_values: Any,
data: dict[str, Any], _k: int) -> list[Any]:
🧰 Tools
🪛 Ruff (0.14.10)

24-24: Dynamically typed expressions (typing.Any) are disallowed in param_values

(ANN401)

25-25: Unused function argument: k

(ARG001)

🤖 Prompt for AI Agents 
In @temoa/extensions/method_of_morris/morris.py around lines 24 - 25, The
evaluate function has an unused parameter k; rename it to _k in the function
signature (def evaluate(param_names: dict[int, list[Any]], param_values: Any,
data: dict[str, Any], _k: int) -> list[Any]) to follow Python convention for
intentionally unused parameters and update any internal references to use _k if
present.

m = len(param_values)
Comment on lines +24 to 26
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot Jan 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Critical: evaluate() depends on global config/db_file (unsafe with resources.as_file lifetime) + unused k.

db_file is created inside a with resources.as_file(...) block (later in the module) but evaluate() uses it later; once the context exits the extracted temp path may no longer exist. Also k is unused (Ruff ARG001).

Proposed fix (make dependencies explicit) 
-def evaluate(param_names: dict[int, list[Any]], param_values: Any,
-            data: dict[str, Any], k: int) -> list[Any]:
+def evaluate(
+    param_names: dict[int, list[Any]],
+    param_values: Any,
+    data: dict[str, Any],
+    *,
+    config: TemoaConfig,
+    db_path: str,
+) -> list[Any]:
     m = len(param_values)
     for j in range(0, m):
         names = param_names[j]
@@
-    mdl, res = run_actions.solve_instance(instance=instance, solver_name=config.solver_name)
+    mdl, res = run_actions.solve_instance(instance=instance, solver_name=config.solver_name)
@@
-    con = sqlite3.connect(db_file)
+    con = sqlite3.connect(db_path)
@@
     con.close()
     return morris_objectives

You’ll also need to update the delayed(evaluate)(...) call sites to pass config= and a stable db_path string that remains valid for the duration of the parallel run.

Also applies to: 40-63

🧰 Tools
🪛 Ruff (0.14.10)

24-24: Dynamically typed expressions (typing.Any) are disallowed in param_values

(ANN401)

25-25: Unused function argument: k

(ARG001)

🤖 Prompt for AI Agents 
In @temoa/extensions/method_of_morris/morris.py around lines 24 - 26, The
evaluate function currently relies on module globals (config and db_file)
created under resources.as_file and takes an unused parameter k; change
evaluate(param_names, param_values, data, k) to accept explicit dependencies
(e.g., add parameters config and db_path: evaluate(param_names, param_values,
data, k, *, config, db_path)) and stop using the global db_file path so it uses
the stable db_path string passed in; remove or mark k as unused (or drop it) to
fix the ARG001 warning. Then update all call sites using delayed(evaluate)(...)
to pass the new config= and db_path= arguments (ensuring db_path is a persistent
string, not a Path from a short-lived resources.as_file context) and remove any
reliance on module-level config/db_file.

for j in range(0, m):
names = param_names[j]
Expand Down Expand Up @@ -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]
cur.execute("SELECT emis_comm, SUM(emission) FROM output_emissionn WHERE emis_comm='co2'")
y_of = row[-1]
cur.execute("SELECT emis_comm, SUM(emission) FROM output_emission 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
Expand Down Expand Up @@ -137,80 +137,76 @@ 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

problem = read_param_file(str(param_file), delimiter=' ')
param_values = sample(
problem, N=10, num_levels=8, optimal_trajectories=False, local_optimization=False, seed=seed
)
print(param_values)
print(param_names)
n = len(param_values)
# pull the data

num_cores = 1 # multiprocessing.cpu_count()
Morris_Objectives = Parallel(n_jobs=num_cores)(
delayed(evaluate)(param_names, param_values[i, :], data, i) for i in range(0, n)
)
Morris_Objectives = array(Morris_Objectives)
print(Morris_Objectives)
Si_OF = morris.analyze(
problem,
param_values,
Morris_Objectives[:, 0],
conf_level=0.95,
print_to_console=False,
num_levels=8,
num_resamples=1000,
seed=seed + 1,
)

Si_CumulativeCO2 = morris.analyze(
problem,
param_values,
Morris_Objectives[:, 1],
conf_level=0.95,
print_to_console=False,
num_levels=8,
num_resamples=1000,
seed=seed + 2,
)
num_vars = problem['num_vars']
groups, unique_group_names = compute_groups_matrix(problem['groups'])
number_of_groups = len(unique_group_names)
print(
'{:<30} {:>10} {:>10} {:>15} {:>10}'.format(
'Parameter', 'Mu_Star', 'Mu', 'Mu_Star_Conf', 'Sigma'
)
)
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],
problem = read_param_file(str(param_file), delimiter=' ')
param_values = sample(
problem, N=10, num_levels=8, optimal_trajectories=False, local_optimization=False, seed=seed
)
print(param_values)
print(param_names)
n = len(param_values)

# pull the data
num_cores = 1 # multiprocessing.cpu_count()
Morris_Objectives = Parallel(n_jobs=num_cores)(
delayed(evaluate)(param_names, param_values[i, :], data, i) for i in range(0, n)
)
)
import csv

line1 = Si_OF['mu_star']
line2 = Si_OF['mu_star_conf']
line3 = Si_CumulativeCO2['mu_star']
line4 = Si_CumulativeCO2['mu_star_conf']
with open('MMResults.csv', 'w') as f:
writer = csv.writer(f, delimiter=',')
writer.writerow(unique_group_names)
writer.writerow('Objective Function')
writer.writerow(line1)
writer.writerow(line2)
writer.writerow('Cumulative CO2 Emissions')
writer.writerow(line3)
writer.writerow(line4)

f.close
print('--- %s seconds ---' % (time.time() - start_time))
Morris_Objectives = array(Morris_Objectives)
Comment on lines +155 to +159
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot Mar 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial

Inconsistent variable naming: Morris_Objectives should be morris_objectives.

Other variables in this file were renamed to snake_case (y_of, y_cumulative_co2, si_of, si_cumulative_co2), but Morris_Objectives retains PascalCase. Apply consistent naming.

Proposed fix 
-        Morris_Objectives = Parallel(n_jobs=num_cores)(
+        morris_objectives = Parallel(n_jobs=num_cores)(
             delayed(evaluate)(param_names, param_values[i, :], data, i) for i in range(0, n)
         )

-        Morris_Objectives = array(Morris_Objectives)
-        print(Morris_Objectives)
+        morris_objectives = array(morris_objectives)
+        print(morris_objectives)
         si_of = morris.analyze(
             problem,
             param_values,
-            Morris_Objectives[:, 0],
+            morris_objectives[:, 0],
🧰 Tools
🪛 Ruff (0.15.6)

[warning] 156-156: Unnecessary start argument in range

Remove start argument

(PIE808)

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@temoa/extensions/method_of_morris/morris.py` around lines 155 - 159, The
variable name Morris_Objectives is using PascalCase while the rest of the module
uses snake_case; rename Morris_Objectives to morris_objectives consistently
where it is assigned and used (the Parallel(...) call and the subsequent
array(...) assignment) so it matches other variables like y_of and si_of; ensure
you update any downstream references to morris_objectives in this file (e.g.,
where evaluate(param_names, param_values[i, :], data, i) results are consumed).

print(Morris_Objectives)
si_of = morris.analyze(
problem,
param_values,
Morris_Objectives[:, 0],
conf_level=0.95,
print_to_console=False,
num_levels=8,
num_resamples=1000,
seed=seed + 1,
)

si_cumulative_co2 = morris.analyze(
problem,
param_values,
Morris_Objectives[:, 1],
conf_level=0.95,
print_to_console=False,
num_levels=8,
num_resamples=1000,
seed=seed + 2,
)
groups, unique_group_names = compute_groups_matrix(problem['groups'])
number_of_groups = len(unique_group_names)
print(
'{:<30} {:>10} {:>10} {:>15} {:>10}'.format(
'Parameter', 'Mu_Star', 'Mu', 'Mu_Star_Conf', 'Sigma'
)
)
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],
)
)

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)
writer.writerow('Objective Function')
writer.writerow(line1)
writer.writerow(line2)
writer.writerow('Cumulative CO2 Emissions')
writer.writerow(line3)
writer.writerow(line4)
Loading
Loading