fixes from PR feedback

Author: ParticularlyPythonicBS

temoa/extensions/modeling_to_generate_alternatives/mga_sequencer.py

@@ -120,24 +120,6 @@ def __init__(self, config: TemoaConfig):
             self.mga_weighting.name,
         )
 
-    def _run_unit_check(self) -> None:
-        """Run unit checking on the database if configured."""
-        if self.config.check_units:
-            from temoa.model_checking.unit_checking.screener import screen
-
-            logger.info('Running units consistency check on input database...')
-            report_path = self.config.output_path / 'unit_check_reports'
-            success = screen(self.config.input_database, report_path=report_path)
-
-            if not success:
-                logger.warning(
-                    'Units check found errors. See detailed report at: %s',
-                    report_path,
-                )
-                logger.warning('Continuing with model build despite unit check warnings...')
-            else:
-                logger.info('Units check completed successfully - no errors found.')
-
     def start(self):
         """Run the sequencer"""
         # ==== basic sequence ====
@@ -147,9 +129,6 @@ def start(self):
         # 4. Instantiate a Manager to pull in more instances
         # 5. Start the re-solve loop
 
-        # Run unit checking if configured
-        self._run_unit_check()
-
         start_time = datetime.now()
 
         # 1. Load data

temoa/extensions/myopic/myopic_sequencer.py

@@ -137,28 +137,7 @@ def get_connection(self) -> Connection:
 
         return con
 
-    def _run_unit_check(self) -> None:
-        """Run unit checking on the database if configured."""
-        if self.config.check_units:
-            from temoa.model_checking.unit_checking.screener import screen
-
-            logger.info('Running units consistency check on input database...')
-            report_path = self.config.output_path / 'unit_check_reports'
-            success = screen(self.config.input_database, report_path=report_path)
-
-            if not success:
-                logger.warning(
-                    'Units check found errors. See detailed report at: %s',
-                    report_path,
-                )
-                logger.warning('Continuing with model build despite unit check warnings...')
-            else:
-                logger.info('Units check completed successfully - no errors found.')
-
     def start(self):
-        # Run unit checking if configured
-        self._run_unit_check()
-
         # load up the instance queue
         self.characterize_run()
 

temoa/model_checking/unit_checking/__init__.py

@@ -1,8 +1,22 @@
 from importlib import resources as importlib_resources
 
 from pint import UnitRegistry
+from pint.errors import DefinitionSyntaxError
 
+# UnitRegistry is generic but doesn't require type args at instantiation
 ureg: UnitRegistry = UnitRegistry()  # type: ignore[type-arg]
+
 # Load custom unit definitions from the package resources
-data = importlib_resources.files('temoa.model_checking.unit_checking').joinpath('temoa_units.txt')
-ureg.load_definitions(str(data))
+_resource_path = 'temoa.model_checking.unit_checking/temoa_units.txt'
+try:
+    data = importlib_resources.files('temoa.model_checking.unit_checking').joinpath(
+        'temoa_units.txt'
+    )
+    # Ensure we have a real filesystem path (handles zipped resources too)
+    with importlib_resources.as_file(data) as path:
+        ureg.load_definitions(path)
+except (FileNotFoundError, OSError, DefinitionSyntaxError) as exc:
+    raise RuntimeError(
+        f'Failed to load custom Temoa unit definitions from {_resource_path!r}. '
+        'This may indicate a broken installation or missing resource file.'
+    ) from exc

temoa/model_checking/unit_checking/common.py

@@ -143,7 +143,7 @@ class UnitsFormat:
 ACCEPTABLE_CHARACTERS = r'^\s*([A-Za-z0-9\*\^\_\s\/\(\)]+?)\s*$'
 
 
-def consolidate_lines(line_nums: Sequence[str | int]) -> str:
+def consolidate_lines(line_nums: Sequence[int]) -> str:
     """A little sand wedge function to prevent lists of many line numbers
     and maxing at 5 or 5 + 'more'"""
     listed_lines = (

temoa/model_checking/unit_checking/relations_checker.py

@@ -17,6 +17,9 @@
     SINGLE_ELEMENT,
     CostTableData,
     RelationType,
+    activity_based_tables,
+    capacity_based_tables,
+    commodity_based_tables,
     consolidate_lines,
 )
 from temoa.model_checking.unit_checking.entry_checker import (
@@ -94,8 +97,20 @@ def check_efficiency_table(
             continue
 
         # check that our tech matches the units of the connected commodities
-        invalid_input_flag = input_units != comm_units[ic]
-        invalid_output_flag = output_units != comm_units[oc]
+        expected_input = comm_units.get(ic)
+        expected_output = comm_units.get(oc)
+        if expected_input is None or expected_output is None:
+            invalid_rows.append(idx)
+            logger.warning(
+                'Missing commodity units for input_comm=%s or output_comm=%s in efficiency row %d',
+                ic,
+                oc,
+                idx,
+            )
+            continue
+
+        invalid_input_flag = input_units != expected_input
+        invalid_output_flag = output_units != expected_output
         if invalid_input_flag or invalid_output_flag:
             logger.warning(
                 'Efficiency units conflict with associated commodity for Technology %s near row %d',
@@ -107,9 +122,9 @@ def check_efficiency_table(
                 f'{tech:^20} ----> {f"{oc} [{output_units}]": ^25}'
             )
             if invalid_input_flag:
-                msg += f'\n    Invalid input units: {comm_units[ic]}'
+                msg += f'\n    Invalid input units: {expected_input}'
             if invalid_output_flag:
-                msg += f'\n    Invalid output units: {comm_units[oc]}'
+                msg += f'\n    Invalid output units: {expected_output}'
             error_msgs.append(msg)
 
         # check that the output of this technology is consistent in units with
@@ -165,7 +180,12 @@ def check_inter_table_relations(
     - Missing tables (e.g., some databases may not have all limit tables)
     - Schema variations between v3.1 and v4.0
     """
-    grouped_errors = defaultdict(list)
+    # Validate table_name against known safe tables
+    valid_tables = activity_based_tables + capacity_based_tables + commodity_based_tables
+    if table_name not in valid_tables:
+        raise ValueError(f'Invalid table name: {table_name}')
+
+    grouped_errors: defaultdict[str, list[int]] = defaultdict(list)
 
     # Build query based on relation type, with robustness checks
     match relation_type:
@@ -206,11 +226,9 @@ def check_inter_table_relations(
 
     try:
         rows = conn.execute(query).fetchall()
-    except sqlite3.OperationalError as e:
+    except sqlite3.OperationalError as _:
         # Log the error but don't fail the entire check
-        logger.error(
-            'failed to process query: %s when processing table %s: %s', query, table_name, e
-        )
+        logger.exception('failed to process query: %s when processing table %s', query, table_name)
         msg = (
             f'Failed to process table {table_name} due to SQL error. '
             f'This may indicate missing columns or incompatible schema. '
@@ -232,7 +250,7 @@ def check_inter_table_relations(
                     continue
                 expected_units = io_units.output_units
             case RelationType.ACTIVITY:
-                io_units = tech_lut[tech_or_comm]
+                io_units = tech_lut.get(tech_or_comm)
                 if not io_units:
                     grouped_errors[
                         f'Unprocessed row (missing reference for tech '
@@ -261,25 +279,28 @@ def check_inter_table_relations(
         if c2a_units:
             c2a_valid, units_data = validate_units_format(c2a_units, SINGLE_ELEMENT)
             if c2a_valid and units_data is not None and len(units_data) >= 1:
-                # further ensure the conversion is valid and retain the validity
-                c2a_valid, valid_c2a_units = validate_units_expression(units_data[0])
+                # further ensure the conversion is valid and retain the appropriate units object
+                _, valid_c2a_units = validate_units_expression(units_data[0])
+                if not valid_c2a_units:
+                    grouped_errors[
+                        f'Invalid units or unit format for c2a table: {c2a_units}'
+                    ].append(idx)
+                    continue
             else:
-                valid_c2a_units = None
-        else:  # we are in a valid state: no C2A units provided/needed
-            c2a_valid = True
+                grouped_errors[f'Invalid units or unit format for c2a table: {c2a_units}'].append(
+                    idx
+                )
+                continue
+        else:
             valid_c2a_units = None
 
         if not valid_table_units:
-            grouped_errors['Unprocessed row (invalid units--see earlier tests)'].append(idx)
-        if not c2a_valid:
-            grouped_errors['Unprocessed row (invalid c2a units--see earlier tests)'].append(idx)
-        if not valid_table_units or not c2a_valid:
+            grouped_errors[f'Invalid units or unit format: {table_units}'].append(idx)
             continue
 
         # if we have valid c2a units, combine them to get the units of activity
         if valid_c2a_units:
             res_units = valid_table_units * (valid_c2a_units * ureg.year)
-
         else:
             res_units = valid_table_units
 
@@ -356,7 +377,7 @@ def check_cost_tables(
         try:
             rows = conn.execute(query).fetchall()
         except sqlite3.OperationalError:
-            logger.error(
+            logger.exception(
                 'failed to process query: %s when processing table %s', query, ct.table_name
             )
             msg = f'Failed to process table {ct.table_name}.  See log for failed query.'
@@ -385,32 +406,51 @@ def check_cost_tables(
                 continue
 
             # Test 1: Look for cost commonality
-            if common_cost_unit is None and cost_units is not None:
-                # try to establish it
-                # Check that the units contain currency dimension
-                # Fixed: Changed from .check('[currency]') which only works for pure currency
-                # to checking dimensionality dict which works for composite units
-                # Example: 'Mdollar / (PJ^2/GW)' has currency in dimensionality but fails .check()
-                try:
-                    if '[currency]' in cost_units.dimensionality:
-                        common_cost_unit = cost_units
-                    else:
-                        # Units don't contain currency dimension - invalid for cost table
-                        error_msgs.append(
-                            f'{ct.table_name}:  Unprocessed row '
-                            f'(units lack currency dimension): {cost_units} at row: {idx}'
+            # extract the cost units
+            if not cost_units:
+                label = (
+                    f'{ct.table_name}:  Unprocessed row '
+                    f'(missing cost units): {raw_units_expression}'
+                )
+                table_grouped_errors[label].append(idx)
+                continue
+
+            # Get cost unit object
+            # cost_units is already a pint Unit object from validate_units_expression(elements[0])
+            cost_unit_obj = cost_units
+
+            # Check for currency dimension
+            if '[currency]' not in cost_unit_obj.dimensionality:
+                label = (
+                    f'{ct.table_name}:  Cost units must have currency dimension. '
+                    f'Found: {cost_unit_obj}'
+                )
+                table_grouped_errors[label].append(idx)
+                continue
+
+            # Initialize common_cost_unit on first valid row
+            if common_cost_unit is None:
+                common_cost_unit = cost_unit_obj
+                # No need to continue here, as we still need to check measure units
+            else:
+                # Validate subsequent rows against the established common unit
+                if cost_unit_obj != common_cost_unit:
+                    # Try to see if they're equivalent but differently expressed
+                    try:
+                        # Attempt conversion to check if they're compatible
+                        (1.0 * cost_unit_obj).to(common_cost_unit)
+                    except (ValueError, AttributeError, TypeError) as e:
+                        # Not compatible - this is an error
+                        label = (
+                            f'{ct.table_name}:  Inconsistent cost units: {cost_unit_obj} '
+                            f'does not match common cost unit {common_cost_unit}. Error: {e}'
                         )
+                        table_grouped_errors[label].append(idx)
                         continue
-                except Exception:
-                    error_msgs.append(
-                        f'{ct.table_name}:  Could not check cost units: {cost_units} at row: {idx}'
-                    )
-                    continue
-                if cost_units != common_cost_unit:
+                    # If compatible but not strictly equal, we still flag it as non-standard
                     label = (
                         f'{ct.table_name}:  Non-standard cost found '
                         f'(expected common cost units of {common_cost_unit}) '
-                        f'got: {cost_units}'
                     )
                     table_grouped_errors[label].append(idx)