Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
70 changes: 70 additions & 0 deletions src/psPlotKit/data_manager/ps_costing.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,8 @@
cm.build()
"""

import difflib

import numpy as np

from psPlotKit.data_manager.ps_data import PsData
Expand Down Expand Up @@ -490,6 +492,7 @@ def __init__(self, data_manager, costing_package, costing_groups=None):
self._flow_type_keys = {} # flow_type → [return_keys]
self._group_flow_type_keys = {} # group_name → {flow_type → [return_keys]}
self._per_group_formula_keys = {} # formula_return_key → [per-group rks]
self._unfound_unit_keys = [] # list of (group, unit, cost_type, suffix)

# ------------------------------------------------------------------
# public API
Expand All @@ -512,6 +515,7 @@ def build(self, load_data=True, error_on_validation_failure=True):
"""
self.data_manager._load_registered_data_files()
self._discover_keys()
self._check_discovery_status()
self._register_parameters()
self._register_discovered_keys()
self._register_validation_keys()
Expand Down Expand Up @@ -562,6 +566,7 @@ def _discover_keys(self):
_logger.info(
"Searching {} available keys for costing data.".format(len(available))
)
self._unfound_unit_keys = []

for group in self.costing_groups:
capex_return_keys = []
Expand All @@ -574,6 +579,10 @@ def _discover_keys(self):
found = self._find_matching_keys(
available, unit_name, group.costing_block_key, suffix
)
if not found:
self._unfound_unit_keys.append(
(group.name, unit_name, "capex", suffix)
)
for file_key in found:
rk = self._make_return_key(
group.name, unit_name, "capex", suffix, file_key
Expand All @@ -586,6 +595,10 @@ def _discover_keys(self):
found = self._find_matching_keys(
available, unit_name, group.costing_block_key, suffix
)
if not found:
self._unfound_unit_keys.append(
(group.name, unit_name, "fixed_opex", suffix)
)
for file_key in found:
rk = self._make_return_key(
group.name, unit_name, "fixed_opex", suffix, file_key
Expand All @@ -601,6 +614,15 @@ def _discover_keys(self):
unit_name,
suffix,
)
if not found:
self._unfound_unit_keys.append(
(
group.name,
unit_name,
"flow({})".format(flow_type),
suffix,
)
)
for file_key in found:
rk = self._make_return_key(
group.name, unit_name, "flow", suffix, file_key
Expand Down Expand Up @@ -636,6 +658,54 @@ def _discover_keys(self):
)
)

def _check_discovery_status(self):
"""Check if all requested unit key patterns were discovered.

Logs any unit key suffixes that were requested via
:meth:`PsCostingGroup.add_unit` but matched zero available keys.
For each unfound suffix, the top 10 nearest available keys are
displayed to help identify typos or alternative key names.

Raises:
KeyError: if any requested key patterns were not discovered.
"""
if not self._unfound_unit_keys:
return

available = sorted(self._get_all_available_keys())
_logger.warning(
"{} requested costing key pattern(s) were NOT discovered.".format(
len(self._unfound_unit_keys)
)
)

for group_name, unit_name, cost_type, suffix in self._unfound_unit_keys:
search_term = "{}.{}.{}".format(unit_name, "costing", suffix)
_logger.warning(
" Group '{}', unit '{}', {} key '{}' - no matching keys found.".format(
group_name, unit_name, cost_type, suffix
)
)
nearest = difflib.get_close_matches(
search_term, available, n=10, cutoff=0.3
Comment thread
avdudchenko marked this conversation as resolved.
)
if nearest:
_logger.warning(" Nearest available keys:")
for n in nearest:
_logger.warning(" {}".format(n))
else:
_logger.warning(" No similar keys found in available data.")

raise KeyError(
"The following requested costing key patterns were not discovered: "
"{}".format(
[
"group='{}' unit='{}' {}='{}'".format(g, u, ct, s)
for g, u, ct, s in self._unfound_unit_keys
]
)
)

@staticmethod
def _match_unit_segments(parts, unit_name):
"""Find the position where *unit_name* matches within *parts*.
Expand Down
83 changes: 83 additions & 0 deletions src/psPlotKit/data_manager/tests/test_ps_costing.py
Original file line number Diff line number Diff line change
Expand Up @@ -1200,3 +1200,86 @@ def test_fraction_with_custom_total_key(self):
]
total = sum(np.asarray(f.data) for f in fracs)
np.testing.assert_array_almost_equal(total, [100.0, 100.0])


# ---------------------------------------------------------------------------
# _check_discovery_status
# ---------------------------------------------------------------------------


class TestCheckDiscoveryStatus:
def test_no_error_when_all_keys_found(self, bgw_dm):
"""When all requested keys match, _check_discovery_status should not raise."""
pkg = PsCostingPackage()
g = PsCostingGroup("Membrane")
g.add_unit("ROUnits", capex_keys=["capital_cost"])

cm = PsCostingManager(bgw_dm, pkg, [g])
cm._discover_keys()
# Should not raise
cm._check_discovery_status()

def test_error_on_unfound_capex_key(self, bgw_dm):
"""A capex key suffix that matches nothing should raise KeyError."""
pkg = PsCostingPackage()
g = PsCostingGroup("Membrane")
g.add_unit("ROUnits", capex_keys=["nonexistent_cost_key"])

cm = PsCostingManager(bgw_dm, pkg, [g])
cm._discover_keys()
with pytest.raises(KeyError, match="nonexistent_cost_key"):
cm._check_discovery_status()

def test_error_on_unfound_fixed_opex_key(self, bgw_dm):
"""A fixed_opex key suffix that matches nothing should raise KeyError."""
pkg = PsCostingPackage()
g = PsCostingGroup("Membrane")
g.add_unit("ROUnits", fixed_opex_keys=["nonexistent_opex"])

cm = PsCostingManager(bgw_dm, pkg, [g])
cm._discover_keys()
with pytest.raises(KeyError, match="nonexistent_opex"):
cm._check_discovery_status()

def test_error_on_unfound_flow_key(self, bgw_dm):
"""A flow key suffix that matches nothing should raise KeyError."""
pkg = PsCostingPackage()
g = PsCostingGroup("Membrane")
g.add_unit(
"ROUnits",
flow_keys={"electricity": ["nonexistent_flow"]},
)

cm = PsCostingManager(bgw_dm, pkg, [g])
cm._discover_keys()
with pytest.raises(KeyError, match="nonexistent_flow"):
cm._check_discovery_status()

def test_build_raises_on_unfound_key(self, bgw_dm):
"""Full build() should raise KeyError when a requested key is not found."""
pkg = PsCostingPackage()
g = PsCostingGroup("Membrane")
g.add_unit("ROUnits", capex_keys=["capital_cost", "bogus_key"])

cm = PsCostingManager(bgw_dm, pkg, [g])
with pytest.raises(KeyError, match="bogus_key"):
cm.build()

def test_mixed_found_and_unfound(self, bgw_dm):
"""Only unfound keys should appear in the error — found ones should not."""
pkg = PsCostingPackage()
g = PsCostingGroup("Membrane")
Comment on lines +1269 to +1271

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

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

@copilot apply changes based on this feedback

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Done in 7928d04. The test now captures the exc_info and adds assert "capital_cost" not in str(exc_info.value) to enforce that found keys are absent from the error message, matching the docstring.

g.add_unit(
"ROUnits",
capex_keys=["capital_cost"],
fixed_opex_keys=["missing_opex_key"],
)

cm = PsCostingManager(bgw_dm, pkg, [g])
cm._discover_keys()
# capital_cost should have been found
assert len(cm._group_capex_keys["Membrane"]) >= 1
# missing_opex_key should be unfound; capital_cost (found) must NOT appear
with pytest.raises(KeyError, match="missing_opex_key") as exc_info:
cm._check_discovery_status()
assert "capital_cost" not in str(exc_info.value)
Loading