diff --git a/src/nemosis/data_fetch_methods.py b/src/nemosis/data_fetch_methods.py index 410967c..79e1522 100644 --- a/src/nemosis/data_fetch_methods.py +++ b/src/nemosis/data_fetch_methods.py @@ -8,6 +8,7 @@ from nemosis import processing_info_maps as _processing_info_maps from nemosis import date_generators as _date_generators from nemosis import defaults as _defaults +from nemosis import query_wrappers as _query_wrappers from nemosis.value_parser import _infer_column_data_types from nemosis.date_generators import parse_datetime_py as _parse_datetime_py from nemosis.custom_errors import UserInputError, NoDataToReturn, DataMismatchError @@ -99,6 +100,7 @@ def dynamic_data_compiler( ) _validate_user_select_columns(select_columns, table_name) + _validate_user_select_columns_includes_pk(select_columns, table_name) _validate_filter_args(filter_cols, filter_values) _validate_time_window(start_time, end_time) @@ -159,7 +161,19 @@ def dynamic_data_compiler( missing_columns = [ col for col in filter_cols if col not in all_data.columns ] - UserInputError(f"Filter columns {missing_columns} not in data.") + # Was previously `UserInputError(...)` (no `raise`) — a + # silent no-op that returned unfiltered data. Now + # properly raised; mostly unreachable thanks to + # `_check_loaded_select_columns` below, but kept as + # defence in depth (e.g. filter_cols came from defaults + # because user passed select_columns=None, and the + # default filter column isn't in this AEMO file + # vintage). + raise UserInputError( + f"Filter columns {missing_columns} not in data for " + f"table {table_name}. Available columns: " + f"{sorted(all_data.columns)}." + ) else: all_data = _filter_on_column_value( all_data, filter_cols, filter_values @@ -265,6 +279,7 @@ def cache_compiler( ) _validate_user_select_columns(select_columns, table_name) + _validate_user_select_columns_includes_pk(select_columns, table_name) _validate_time_window(start_time, end_time) user_select_columns = select_columns @@ -359,8 +374,18 @@ def static_table( raise UserInputError("Table name provided is not a static table.") _validate_user_select_columns(select_columns, table_name) + _validate_user_select_columns_includes_pk(select_columns, table_name) _validate_filter_args(filter_cols, filter_values) + # Remember whether the user explicitly asked for columns, so we can + # enforce strict membership after the file is loaded. Defaults- + # inherited columns are allowed to silently be missing from the + # static file (e.g. AEMO drops a column from the registration + # workbook); user-typed columns should not be — that's almost always + # a typo, and silently returning a stub DataFrame is worse than + # raising. Mirrors the dynamic_data_compiler contract. + user_select_columns = select_columns + if filter_cols and not set(filter_cols).issubset(set(select_columns)): raise UserInputError( ( @@ -412,10 +437,28 @@ def static_table( for column in table.select_dtypes(["object"]).columns: table[column] = table[column].map(lambda x: _strip_if_string(x)) + # Reject before filtering: catches the silent-stub bug where a typo + # in user-typed select_columns (e.g. 'duid' instead of 'DUID') used + # to ship a DataFrame missing the requested column with only a + # WARNING. Mirrors `_check_loaded_select_columns` in + # dynamic_data_compiler. + _check_loaded_select_columns(table, user_select_columns, table_name) + if filter_cols is not None: if not set(filter_cols).issubset(set(table.columns)): missing_columns = [col for col in filter_cols if col not in table.columns] - UserInputError(f"Filter columns {missing_columns} not in data.") + # Was previously `UserInputError(...)` (no `raise`) — a + # silent no-op that returned unfiltered data. Now properly + # raised; this path is mostly unreachable thanks to the + # post-load select_columns check above, but kept as a + # defence in depth (e.g. filter_cols came from defaults + # because user passed select_columns=None, and the default + # filter column isn't in this AEMO file vintage). + raise UserInputError( + f"Filter columns {missing_columns} not in data for " + f"table {table_name}. Available columns: " + f"{sorted(table.columns)}." + ) else: table = _filter_on_column_value(table, filter_cols, filter_values) @@ -617,6 +660,49 @@ def _set_up_dynamic_compilers(table_name, start_time, end_time, select_columns): return start_time, end_time, select_columns, date_filter, start_search +def _iteration_overlaps_window(year, month, day, index, user_start, user_end): + """Does this date_gen iteration's data period overlap the user's + [user_start, user_end) window? + + Used to distinguish iterations the user actually asked for from + book-keeping iterations the fetch loop makes: + - the 1-day buffer-back month before `start_time`, + - the ~120 historical months a `search_type='all'` table scans + to compute "snapshot as of start_time". + + A user asking for 16 months of an effective-date table should not + be warned that we didn't get data for 1997-11; they didn't ask for + 1997-11. A user asking for 16 months of a current/scraped table + against AEMO's 6-week retention SHOULD be warned that ~450 days of + their request had no data. + + Period extents (right-open, matching AEMO's end-of-interval + convention): + - (year, month, day=None, index=None): one calendar month + - (year, month, day, index=None): one calendar day + - (year, month, day, index="HHMM"): one 5-minute FCAS slot + """ + year_i = int(year) + month_i = int(month) + if day is None: + period_start = _datetime(year_i, month_i, 1) + if month_i == 12: + period_end = _datetime(year_i + 1, 1, 1) + else: + period_end = _datetime(year_i, month_i + 1, 1) + elif index is None: + day_i = int(day) + period_start = _datetime(year_i, month_i, day_i) + period_end = period_start + _timedelta(days=1) + else: + day_i = int(day) + hour_i = int(index[:2]) + minute_i = int(index[2:]) + period_start = _datetime(year_i, month_i, day_i, hour_i, minute_i) + period_end = period_start + _timedelta(minutes=5) + return period_start < user_end and period_end > user_start + + def _dynamic_data_fetch_loop( start_search, start_time, @@ -656,7 +742,29 @@ def _dynamic_data_fetch_loop( start_search = start_search - _timedelta(days=1) date_gen = date_gen_func(start_search, end_time) + # Track per-period success across the user's requested window so we + # can emit a single coverage-gap summary at the end. Without it, a + # multi-day query against e.g. INTERMITTENT_GEN_SCADA whose range + # mostly falls outside AEMO's Reports/Current/ retention window + # silently returns just the in-window days — the only signal is N + # separate per-file warnings that an aggregating user is unlikely + # to read. + # + # We count only iterations whose period overlaps the user's + # [start_time, end_time) window — so the buffer-back month (always + # one month before start_time) and the historical scan months for + # `search_type='all'` tables are excluded from the denominator. + # Any gap that remains is a gap in the user's actual request. + requested_periods = 0 + successful_requested_periods = 0 + for year, month, day, index in date_gen: + in_user_window = _iteration_overlaps_window( + year, month, day, index, start_time, end_time + ) + if in_user_window: + requested_periods += 1 + period_has_data = False check_for_next_data_chunk = True chunk = 0 while check_for_next_data_chunk: @@ -741,6 +849,7 @@ def _dynamic_data_fetch_loop( ) data_tables.append(data) + period_has_data = True elif not caching_mode and chunk == 1: # Demoted from WARNING to DEBUG: when we reach here, the # cache file doesn't exist on disk — which means the @@ -754,6 +863,30 @@ def _dynamic_data_fetch_loop( if data is None or '#' not in filename_stub: check_for_next_data_chunk = False + if period_has_data and in_user_window: + successful_requested_periods += 1 + + # Warn iff the user's requested window has a gap, AND we did get at + # least some data (zero-data is already covered by NoDataToReturn + # in the caller — a duplicate WARNING here would be noise). + if ( + not caching_mode + and 0 < successful_requested_periods < requested_periods + ): + missing = requested_periods - successful_requested_periods + coverage = successful_requested_periods / requested_periods + logger.warning( + f"Partial coverage for {table_name}: only " + f"{successful_requested_periods}/{requested_periods} " + f"periods in the requested window returned data " + f"({coverage:.0%}). {missing} period(s) missing — see " + f"per-period WARNINGs above for the specific files. " + f"Common causes: requested range extends beyond AEMO's " + f"Reports/Current/ retention window (typically ~6-7 weeks " + f"for the current/scraped tables); requested month not " + f"yet published in the MMSDM archive; requested range " + f"pre-dates AEMO data." + ) return data_tables @@ -851,6 +984,54 @@ def _validate_user_select_columns(select_columns, table_name): ) +def _validate_user_select_columns_includes_pk(select_columns, table_name): + """Ensure user-supplied select_columns contains every primary-key + column for tables whose processing pipeline dedupes on PK. + + Without this, the dedup step inside finalise (`drop_duplicates_by_primary_key` + for dynamic tables, or `_finalise_excel_data` for the Generators + static table) raises a bare pandas `KeyError: Index([...], dtype='str')` + that points at pandas internals rather than the user's actual + mistake — they omitted a column NEMOSIS needs internally for + correct dedup semantics. + + Defaults always include the PK columns (by construction of + `defaults.table_columns`), so this check only fires when the user + explicitly types a select_columns subset that's missing one. Pass + `select_columns=None` to fall back to defaults. + """ + if select_columns is None or select_columns == "all": + return + pk = _defaults.table_primary_keys.get(table_name) + if not pk: + return + # Dynamic-table dedup runs `drop_duplicates_by_primary_key` from the + # finalise pipeline. + finalise = _processing_info_maps.finalise.get(table_name) + is_dynamic_pk_dedup = bool( + finalise and any( + fn is _query_wrappers.drop_duplicates_by_primary_key for fn in finalise + ) + ) + # Static-table dedup runs `_finalise_excel_data`, which does + # `data.drop_duplicates(primary_keys)` if the table is in + # `defaults.table_primary_keys`. + static_finalisers = static_data_finaliser_map.get(table_name, []) + is_static_pk_dedup = _finalise_excel_data in static_finalisers + if not (is_dynamic_pk_dedup or is_static_pk_dedup): + return + missing = [c for c in pk if c not in select_columns] + if missing: + raise UserInputError( + f"select_columns for {table_name} must include the table's " + f"primary-key columns so NEMOSIS can dedupe correctly. " + f"Missing: {missing}. Full primary key: {pk}. Either add the " + f"missing column(s) to select_columns, or pass " + f"select_columns=None to use the table defaults (which " + f"already include the primary key)." + ) + + def _check_loaded_select_columns(all_data, user_select_columns, table_name): """Post-load check: every column the user explicitly asked for must actually have made it into the returned DataFrame. diff --git a/src/nemosis/processing_info_maps.py b/src/nemosis/processing_info_maps.py index 768a1a3..11e0bd0 100644 --- a/src/nemosis/processing_info_maps.py +++ b/src/nemosis/processing_info_maps.py @@ -231,7 +231,19 @@ query_wrappers.most_recent_records_before_start_time, query_wrappers.drop_duplicates_by_primary_key, ], - "MARKET_PRICE_THRESHOLDS": None, + "MARKET_PRICE_THRESHOLDS": [ + # `effective_date_group_col` for this table is `[]`, which sends + # `most_recent_records_before_start_time` down the `.tail(1)` + # branch — exactly one "as-of" row from before start_time. The + # subsequent dedup-by-PK then collapses the many-monthly-archives + # duplicates of every effective date `>= start_time` down to one + # row per (EFFECTIVEDATE, VERSIONNO). Previously `None` here, + # which left every monthly archive's copy of every effective + # date in the returned DataFrame — e.g. a 5-month query returned + # ~75× duplicates per row. + query_wrappers.most_recent_records_before_start_time, + query_wrappers.drop_duplicates_by_primary_key, + ], "DAILY_REGION_SUMMARY": None, "ROOFTOP_PV_ACTUAL": [ query_wrappers.drop_duplicates_by_primary_key diff --git a/tests/end_to_end_table_tests/test_market_price_thresholds.py b/tests/end_to_end_table_tests/test_market_price_thresholds.py index 731555a..909196a 100644 --- a/tests/end_to_end_table_tests/test_market_price_thresholds.py +++ b/tests/end_to_end_table_tests/test_market_price_thresholds.py @@ -21,3 +21,17 @@ def test_market_price_thresholds_returns_fixtured_rows(nemosis_fixture, monkeypa ) assert not data.empty + + # PK invariant: every (EFFECTIVEDATE, VERSIONNO) pair appears at + # most once. Pre-fix, MARKET_PRICE_THRESHOLDS had `finalise=None` in + # processing_info_maps, so every monthly archive's copy of every + # effective date passed through — a 5-month query returned ~75× + # duplicates per row. Even on this 2-month fixture, the bug yields + # 2 copies of every row. + pk = ["EFFECTIVEDATE", "VERSIONNO"] + pair_counts = data.groupby(pk).size() + assert pair_counts.max() == 1, ( + f"MARKET_PRICE_THRESHOLDS PK invariant broken: max rows per " + f"(EFFECTIVEDATE, VERSIONNO) = {int(pair_counts.max())}, " + f"expected 1. Likely regression in processing_info_maps." + ) diff --git a/tests/test_errors.py b/tests/test_errors.py index fc23b7f..cc1bf22 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -61,6 +61,34 @@ def test_dynamic_filter_col_not_a_real_column(nemosis_fixture): ) +def test_dynamic_select_columns_must_include_pk_for_dedup_tables(nemosis_fixture): + """For tables whose finalise pipeline includes + `drop_duplicates_by_primary_key`, omitting a PK column used to + crash with a bare pandas `KeyError: Index(['VERSIONNO'], ...)` from + inside finalise — opaque to users. Catch it up-front with a clear + UserInputError naming the missing PK column(s).""" + with pytest.raises(UserInputError, match="primary-key"): + dynamic_data_compiler( + "2021/05/01 00:00:00", "2021/05/01 01:00:00", + "LOSSFACTORMODEL", str(nemosis_fixture), + # LOSSFACTORMODEL PK = [EFFECTIVEDATE, INTERCONNECTORID, + # REGIONID, VERSIONNO]. Deliberately omit VERSIONNO. + select_columns=["EFFECTIVEDATE", "INTERCONNECTORID", "REGIONID", + "DEMANDCOEFFICIENT"], + ) + + +def test_static_select_columns_must_include_pk_for_dedup_tables(nemosis_fixture): + """`_finalise_excel_data` calls `data.drop_duplicates(primary_keys)`, + so static_table has the same PK-omission crash as dynamic_data_compiler. + Generators and Scheduled Loads PK = ['DUID'].""" + with pytest.raises(UserInputError, match="primary-key"): + static_table( + "Generators and Scheduled Loads", str(nemosis_fixture), + select_columns=["Participant", "Station Name", "Region"], + ) + + def test_dynamic_select_columns_all_requires_csv(nemosis_fixture): with pytest.raises(UserInputError, match="select_columns='all'"): dynamic_data_compiler( @@ -160,6 +188,39 @@ def test_static_filter_col_not_in_select_columns(nemosis_fixture): ) +def test_static_select_columns_typo_raises(nemosis_fixture): + """A typo in user-typed select_columns (e.g. 'duid' instead of + 'DUID') used to silently return a stub DataFrame with whatever + columns *did* match, plus a WARNING. Now raises UserInputError + with the available columns listed — mirrors the + dynamic_data_compiler contract.""" + with pytest.raises(UserInputError, match="not present in the data"): + static_table( + "Generators and Scheduled Loads", str(nemosis_fixture), + select_columns=["Participant", "Station Name", "DUID", + "totally_not_a_real_column"], + ) + + +def test_static_filter_col_not_in_loaded_data_raises(nemosis_fixture): + """Defence-in-depth: the missing-`raise` bug in the post-load + filter_cols check used to silently no-op filters whose column had + survived the early select_columns validator but didn't make it + into the loaded file (defaults-but-vintage-missing case). Now + raises.""" + # First confirm that the user-typed-col path raises early (via + # _check_loaded_select_columns) — the filter_cols-not-in-data path + # below would also raise, but earlier. + with pytest.raises(UserInputError): + static_table( + "Generators and Scheduled Loads", str(nemosis_fixture), + select_columns=["Participant", "Station Name", "DUID", + "totally_not_a_real_column"], + filter_cols=["totally_not_a_real_column"], + filter_values=(["whatever"],), + ) + + def test_static_raw_data_location_is_none(): with pytest.raises(UserInputError, match="is None"): static_table("VARIABLES_FCAS_4_SECOND", raw_data_location=None) @@ -179,6 +240,85 @@ def test_dynamic_no_data_raises(nemosis_fixture): ) +def test_dynamic_partial_coverage_emits_summary_warning(nemosis_fixture, caplog): + """When a multi-period query partly succeeds and partly 404s, emit + a single summary WARNING about coverage at the end of the fetch + loop. Without it, a user aggregating across the range silently + operates on a subset (e.g. asking for 16 months of a current/ + scraped table whose retention is ~6-7 weeks). + + Fixture covers DISPATCHPRICE for 2018-04 and 2018-05 (plus other + discrete months). Querying 2018-04-01 → 2018-12-01 spans 9 months + of which only 2 are fixtured; coverage is well below the 90% + threshold.""" + import logging + caplog.set_level(logging.WARNING, logger="nemosis") + dynamic_data_compiler( + "2018/04/01 00:00:00", "2018/12/01 00:00:00", + "DISPATCHPRICE", str(nemosis_fixture), + select_columns=["SETTLEMENTDATE", "REGIONID", "RRP"], + ) + summary = [r for r in caplog.records + if "Partial coverage" in r.getMessage()] + assert len(summary) == 1, ( + f"expected exactly one Partial-coverage WARNING; got " + f"{[r.getMessage() for r in summary]}" + ) + msg = summary[0].getMessage() + assert "DISPATCHPRICE" in msg + assert "retention" in msg.lower() or "MMSDM" in msg + + +def test_dynamic_full_coverage_emits_no_summary_warning(nemosis_fixture, caplog): + """The partial-coverage warning must NOT fire on a single-month + query that fully succeeds. Without this, the warning would spam on + every well-formed query and lose its signal.""" + import logging + caplog.set_level(logging.WARNING, logger="nemosis") + dynamic_data_compiler( + "2018/05/01 00:00:00", "2018/05/01 01:00:00", + "DISPATCHPRICE", str(nemosis_fixture), + select_columns=["SETTLEMENTDATE", "REGIONID", "RRP"], + ) + summary = [r for r in caplog.records + if "Partial coverage" in r.getMessage()] + assert not summary, ( + f"unexpected Partial-coverage WARNING on a fully-fixtured " + f"query: {[r.getMessage() for r in summary]}" + ) + + +def test_dynamic_partial_coverage_ignores_out_of_window_iterations( + nemosis_fixture, caplog, monkeypatch +): + """The partial-coverage check counts only periods that overlap the + user's [start_time, end_time) window. For a `search_type='all'` + table, the fetch loop scans many historical months to compute the + "snapshot as of start_time" — those months are NOT what the user + asked for, so a 404 on any of them must not raise the gap signal. + + This test queries MARKET_PRICE_THRESHOLDS for just 2021-05 with + `nem_data_model_start_time` monkeypatched to 2021-02. date_gen + then yields 2021-01 (buffer-back), 2021-02, 03, 04, 05. The + fixture only has 2021-04 and 2021-05; the earlier months 404. But + only 2021-05 is in the user's window, so coverage should read as + 1/1 — no partial-coverage warning.""" + import logging + monkeypatch.setattr(defaults, "nem_data_model_start_time", + "2021/02/01 00:00:00") + caplog.set_level(logging.WARNING, logger="nemosis") + dynamic_data_compiler( + "2021/05/01 00:00:00", "2021/05/01 01:00:00", + "MARKET_PRICE_THRESHOLDS", str(nemosis_fixture), + ) + summary = [r for r in caplog.records + if "Partial coverage" in r.getMessage()] + assert not summary, ( + f"out-of-user-window 404s falsely triggered the partial-" + f"coverage warning: {[r.getMessage() for r in summary]}" + ) + + def test_static_no_data_raises(nemosis_fixture, monkeypatch): """Redirect a static-table URL to a path the dummy server doesn't serve — NEMOSIS should raise NoDataToReturn, not crash cryptically."""