Improve ML fault detection pipeline job and ML training - #6437
Open
wabinyai wants to merge 17 commits into
Open
Conversation
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a fault-detection pipeline: BigQuery fetch and SQL template, extended FaultDetectionUtils (rule- and pattern-based detection, IsolationForest lifecycle), DAGs for detection and training, config/env entries, Mongo persistence, and tests for the new functionality. Changes
Sequence DiagramsequenceDiagram
participant Dag as "Airflow DAG"
participant BigQuery as "BigQuery"
participant FDUtils as "FaultDetectionUtils"
participant GCS as "GCS (model storage)"
participant Isolation as "IsolationForest (model)"
participant Mongo as "MongoDB"
Dag->>BigQuery: fetch_fault_detection_raw_readings(lookback_days, device_limit, min_records)
BigQuery-->>FDUtils: raw sensor rows (device_id, timestamp, sensors...)
FDUtils->>FDUtils: flag_rule_based_faults()
FDUtils->>FDUtils: initialize_device_fault_status()
FDUtils->>FDUtils: prepare_pattern_detection_features(freq=HOURLY)
alt model exists in GCS
FDUtils->>GCS: load_isolation_forest_model()
GCS-->>Isolation: return model
else model missing & train_if_missing=True
FDUtils->>FDUtils: train_and_save_isolation_forest()
FDUtils->>GCS: save_isolation_forest_model()
end
FDUtils->>Isolation: score features -> anomaly_score, anomaly_value
Isolation-->>FDUtils: anomaly outputs
FDUtils->>FDUtils: process_faulty_devices_percentage()
FDUtils->>FDUtils: process_faulty_devices_fault_sequence()
FDUtils->>Mongo: bulk_write upserts (fault summary per device)
Mongo-->>FDUtils: write acknowledgement
Estimated Code Review Effort🎯 4 (Complex) | ⏱️ ~50 minutes Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Tip 💬 Introducing Slack Agent: The best way for teams to turn conversations into code.Slack Agent is built on CodeRabbit's deep understanding of your code, so your team can collaborate across the entire SDLC without losing context.
Built for teams:
One agent for your entire SDLC. Right inside Slack. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## staging #6437 +/- ##
========================================
Coverage 12.88% 12.88%
========================================
Files 269 269
Lines 33184 33184
Branches 865 865
========================================
Hits 4276 4276
Misses 28868 28868
Partials 40 40 🚀 New features to boost your workflow:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (1)
src/workflows/airqo_etl_utils/tests/test_bigquery_api.py (1)
187-190: Avoid asserting the path supplied by the test double.
captured["source"]is assigned fromexpected_query.sourceinside the stub, so Line 207 only verifies the fixture value. Either drop that assertion here or add a separate query-manager test that loadsfault_detection_raw_device_readingsfrom the real template registry.♻️ Proposed cleanup
def fake_get_query(name): captured["query_name"] = name - captured["source"] = expected_query.source return expected_query @@ assert captured["query_name"] == "fault_detection_raw_device_readings" - assert captured["source"].parent.name == "faultdetection" assert "project.dataset.raw_measurements" in captured["rendered_query"]Also applies to: 206-207
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/tests/test_bigquery_api.py` around lines 187 - 190, The test stub fake_get_query assigns captured["source"] from expected_query.source causing the subsequent assertion to only verify the fixture value; update the test to either remove the assertion against captured["source"] (since it simply mirrors expected_query) or replace it with a focused integration-style test that calls the real query registry to load the template "fault_detection_raw_device_readings" and asserts the actual template source; locate the fake_get_query stub and the captured dict in test_bigquery_api.py (and the assertion lines around 206-207) and implement one of these two fixes so the test no longer asserts a value that is guaranteed by the test double.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/workflows/airqo_etl_utils/config.py`:
- Line 864: FAULT_DETECTION_LOOKBACK_DAYS is currently parsed with
int(os.getenv(...)) which will raise on blank/non-numeric and allow non-positive
values; replace that direct parse with a small helper (e.g.,
get_positive_int_env or parse_positive_int_env) that reads the env var, returns
the default (14) when the value is missing or not an integer, and enforces >0
(falling back to the default if <=0); use that helper to set
FAULT_DETECTION_LOOKBACK_DAYS so imports never raise and the lookback is always
a sensible positive integer.
In `@src/workflows/airqo_etl_utils/ml_utils.py`:
- Around line 717-726: The `_has_invalid_value_fault` function currently raises
a range fault when either the absolute invalid_count or the invalid_share
threshold is exceeded; change the condition to require both criteria so
transient outliers on sparse series don't trigger faults. Locate
`_has_invalid_value_fault(series: pd.Series, lower: float, upper: float)` and
replace the final boolean expression that uses
FaultDetectionUtils.MIN_INVALID_VALUE_COUNT and
FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD with a conjunction (both must
be true) while keeping the early empty-series check and
invalid_count/invalid_share calculations intact.
- Around line 931-934: When model_df is empty or has fewer than 2 rows (the
early-return in the function that prepares anomalies), explicitly set
anomaly_value to 1 and anomaly_score to 0.0 for any single-row DataFrame instead
of assigning empty Series; update the branch that currently does
model_df["anomaly_value"] = pd.Series(dtype=int) and model_df["anomaly_score"] =
pd.Series(dtype=float) so that for a one-row DataFrame you assign
model_df["anomaly_value"] = 1 and model_df["anomaly_score"] = 0.0 (or equivalent
scalar assignment) before returning model_df to preserve downstream filtering
semantics (e.g., df[df["anomaly_value"] == -1]).
- Around line 1069-1093: The global merged_df.fillna(0) corrupts string metadata
like device_name; instead, only fill numeric fault columns with 0 and
preserve/restore device_name from device_id: drop the global fillna and
explicitly do merged_df["anomaly_percentage"] =
merged_df.get("anomaly_percentage", pd.Series()).fillna(0) and
merged_df["fault_count"] = merged_df.get("fault_count", pd.Series()).fillna(0)
(or use .loc for existing cols), then ensure device_name exists or fill its NaNs
with merged_df["device_id"] (e.g., if "device_name" not in merged_df.columns
create it from device_id; if it exists but contains NaN use
merged_df["device_name"].fillna(merged_df["device_id"], inplace=True)); keep the
subsequent creation of anomaly_percentage_fault/anomaly_sequence_fault and
casting of FaultDetectionUtils.RULE_FAULT_COLUMNS as before.
In
`@src/workflows/airqo_etl_utils/sql/faultdetection/2026041701fault_detection.sql`:
- Line 16: The FROM clause uses the placeholder raw_measurements_table without
quoting, which breaks when the project ID contains hyphens; update the SQL to
quote the fully qualified BigQuery table identifier by wrapping the placeholder
in backticks (i.e., ` {raw_measurements_table} `) wherever FROM
{raw_measurements_table} appears—ensure the replacement occurs in the
2026041701fault_detection.sql file and any other occurrences so the BigQuery
parser accepts project IDs with hyphens.
In `@src/workflows/airqo_etl_utils/tests/test_bigquery_api.py`:
- Around line 138-145: The test sets api.client.query.side_effect earlier which
overrides return_value; to exercise the empty-DataFrame branch clear that
side_effect before stubbing the return. In the test for
BigQueryApi.fetch_raw_readings, set api.client.query.side_effect = None (or
delete the side_effect) prior to assigning
api.client.query.return_value.result.return_value.to_dataframe.return_value =
pd.DataFrame() so the return_value path is used and the ValueError "No data
found from bigquery" is raised.
In `@src/workflows/airqo_etl_utils/tests/test_query_manager.py`:
- Line 2: The test currently only verifies that the SQL template includes the
raw_measurements_table replacement but misses the lookback_days placeholder,
allowing accidental hardcoding; update the test for fetch_raw_readings() to also
assert the lookback_days substitution by either (a) passing a known
lookback_days value into query_manager.fetch_raw_readings() and asserting the
returned SQL string contains that numeric/string value, or (b) if testing the
template before format, assert the template contains the "{lookback_days}"
token; apply the same additional assertion to the other test block around lines
69-74 and reference the fetch_raw_readings and raw_measurements_table symbols
when making the assertion.
In `@src/workflows/dags/fault_detection_job.py`:
- Around line 63-74: The initializer DataFrame from
initialize_device_fault_status(raw_data) should not be passed into save_to_mongo
because save_faulty_devices marks merged rows as fault_detected=1; remove
device_fault_status from the save_to_mongo call and instead pass only the actual
fault/result DataFrames (rule_based_faults, pattern_based_faults,
faulty_devices_percentage, faulty_devices_sequence) so
save_to_mongo/save_faulty_devices only marks devices that appear in those fault
sources as faulty; update the call site where save_to_mongo(...) is invoked
accordingly and leave initialize_device_fault_status used only where a full
device list is genuinely required.
---
Nitpick comments:
In `@src/workflows/airqo_etl_utils/tests/test_bigquery_api.py`:
- Around line 187-190: The test stub fake_get_query assigns captured["source"]
from expected_query.source causing the subsequent assertion to only verify the
fixture value; update the test to either remove the assertion against
captured["source"] (since it simply mirrors expected_query) or replace it with a
focused integration-style test that calls the real query registry to load the
template "fault_detection_raw_device_readings" and asserts the actual template
source; locate the fake_get_query stub and the captured dict in
test_bigquery_api.py (and the assertion lines around 206-207) and implement one
of these two fixes so the test no longer asserts a value that is guaranteed by
the test double.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: d5733a98-a2df-472e-b22e-549bf1a35160
📒 Files selected for processing (12)
src/workflows/airqo_etl_utils/bigquery_api.pysrc/workflows/airqo_etl_utils/config.pysrc/workflows/airqo_etl_utils/ml_utils.pysrc/workflows/airqo_etl_utils/sql/faultdetection/2026041701fault_detection.sqlsrc/workflows/airqo_etl_utils/sql/faultdetection/__init__.pysrc/workflows/airqo_etl_utils/tests/test_bigquery_api.pysrc/workflows/airqo_etl_utils/tests/test_ml_utils.pysrc/workflows/airqo_etl_utils/tests/test_query_manager.pysrc/workflows/dags/dag_docs.pysrc/workflows/dags/fault_detection_job.pysrc/workflows/dags/task_docs.pysrc/workflows/env.sample
| DAILY_FORECAST_PREDICTION_JOB_SCOPE = os.getenv( | ||
| "DAILY_FORECAST_PREDICTION_JOB_SCOPE" | ||
| ) | ||
| FAULT_DETECTION_LOOKBACK_DAYS = int(os.getenv("FAULT_DETECTION_LOOKBACK_DAYS", "14")) |
Contributor
There was a problem hiding this comment.
Validate the lookback value before DAG import depends on it.
int(os.getenv(...)) will fail at import time if the env var is blank/non-numeric, and non-positive values can render an invalid or misleading BigQuery interval. Consider parsing with a small helper that defaults safely and enforces > 0.
Proposed validation
- FAULT_DETECTION_LOOKBACK_DAYS = int(os.getenv("FAULT_DETECTION_LOOKBACK_DAYS", "14"))
+ try:
+ FAULT_DETECTION_LOOKBACK_DAYS = int(
+ os.getenv("FAULT_DETECTION_LOOKBACK_DAYS", "14")
+ )
+ except (TypeError, ValueError):
+ FAULT_DETECTION_LOOKBACK_DAYS = 14
+ if FAULT_DETECTION_LOOKBACK_DAYS <= 0:
+ raise ValueError("FAULT_DETECTION_LOOKBACK_DAYS must be a positive integer")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/config.py` at line 864,
FAULT_DETECTION_LOOKBACK_DAYS is currently parsed with int(os.getenv(...)) which
will raise on blank/non-numeric and allow non-positive values; replace that
direct parse with a small helper (e.g., get_positive_int_env or
parse_positive_int_env) that reads the env var, returns the default (14) when
the value is missing or not an integer, and enforces >0 (falling back to the
default if <=0); use that helper to set FAULT_DETECTION_LOOKBACK_DAYS so imports
never raise and the lookback is always a sensible positive integer.
Comment on lines
+717
to
+726
| def _has_invalid_value_fault(series: pd.Series, lower: float, upper: float) -> bool: | ||
| clean_series = series.dropna() | ||
| if clean_series.empty: | ||
| return False | ||
| invalid_count = ((clean_series < lower) | (clean_series > upper)).sum() | ||
| invalid_share = invalid_count / len(clean_series) | ||
| return bool( | ||
| invalid_count >= FaultDetectionUtils.MIN_INVALID_VALUE_COUNT | ||
| or invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD | ||
| ) |
Contributor
There was a problem hiding this comment.
Require both invalid count and share before raising a range fault.
With the current or, a single invalid reading on a sparse device has invalid_share == 100% and bypasses MIN_INVALID_VALUE_COUNT, so transient outliers can mark a device faulty.
🛡️ Proposed debounce fix
invalid_count = ((clean_series < lower) | (clean_series > upper)).sum()
invalid_share = invalid_count / len(clean_series)
return bool(
invalid_count >= FaultDetectionUtils.MIN_INVALID_VALUE_COUNT
- or invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD
+ and invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD
)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 717 - 726, The
`_has_invalid_value_fault` function currently raises a range fault when either
the absolute invalid_count or the invalid_share threshold is exceeded; change
the condition to require both criteria so transient outliers on sparse series
don't trigger faults. Locate `_has_invalid_value_fault(series: pd.Series, lower:
float, upper: float)` and replace the final boolean expression that uses
FaultDetectionUtils.MIN_INVALID_VALUE_COUNT and
FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD with a conjunction (both must
be true) while keeping the early empty-series check and
invalid_count/invalid_share calculations intact.
Comment on lines
+931
to
+934
| if model_df.empty or len(model_df.index) < 2: | ||
| model_df["anomaly_value"] = pd.Series(dtype=int) | ||
| model_df["anomaly_score"] = pd.Series(dtype=float) | ||
| return model_df |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
python - <<'PY'
import pandas as pd
frame = pd.DataFrame({"device_id": ["device-a"]})
frame["anomaly_value"] = pd.Series(dtype=int)
assert pd.isna(frame.loc[0, "anomaly_value"])
print(frame)
PYRepository: airqo-platform/AirQo-api
Length of output: 186
🏁 Script executed:
# First, check if the file exists and examine the relevant section
wc -l src/workflows/airqo_etl_utils/ml_utils.pyRepository: airqo-platform/AirQo-api
Length of output: 114
🏁 Script executed:
# Read the lines around 931-934 to understand context
sed -n '900,950p' src/workflows/airqo_etl_utils/ml_utils.py | cat -nRepository: airqo-platform/AirQo-api
Length of output: 2433
🏁 Script executed:
# Search for the function containing these lines to understand its purpose
grep -n "def " src/workflows/airqo_etl_utils/ml_utils.py | tail -20Repository: airqo-platform/AirQo-api
Length of output: 1515
🏁 Script executed:
# Look for downstream usage of anomaly_value and anomaly_score
rg "anomaly_value|anomaly_score" --max-count=20 -B 2 -A 2Repository: airqo-platform/AirQo-api
Length of output: 5595
🏁 Script executed:
# Check test file for edge case tests (single row, empty dataframe, etc.)
rg "flag_pattern_based_faults" src/workflows/airqo_etl_utils/tests/test_ml_utils.py -A 50 | head -100Repository: airqo-platform/AirQo-api
Length of output: 2843
🏁 Script executed:
# Search for any test cases with 1 or 2 rows
grep -n "def test" src/workflows/airqo_etl_utils/tests/test_ml_utils.py | grep -i "anomaly\|fault"Repository: airqo-platform/AirQo-api
Length of output: 905
🏁 Script executed:
# Check if there are any comments or documentation about the <2 rows case
sed -n '920,940p' src/workflows/airqo_etl_utils/ml_utils.py | cat -nRepository: airqo-platform/AirQo-api
Length of output: 1075
Fill anomaly fields with explicit "normal" values for single-row inputs.
Assigning empty Series to a one-row model_df leaves anomaly_value and anomaly_score as NaN. This breaks downstream filtering (e.g., df[df["anomaly_value"] == -1] silently skips NaN rows) and aggregation semantics.
Proposed fix
if model_df.empty or len(model_df.index) < 2:
- model_df["anomaly_value"] = pd.Series(dtype=int)
- model_df["anomaly_score"] = pd.Series(dtype=float)
+ model_df["anomaly_value"] = 1
+ model_df["anomaly_score"] = 0.0
return model_dfAssigning 1 (the "normal" class from IsolationForest semantics) and 0.0 (a neutral decision score) treats insufficient data conservatively while maintaining consistency with the downstream logic that filters on anomaly_value == -1 for anomalies.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 931 - 934, When
model_df is empty or has fewer than 2 rows (the early-return in the function
that prepares anomalies), explicitly set anomaly_value to 1 and anomaly_score to
0.0 for any single-row DataFrame instead of assigning empty Series; update the
branch that currently does model_df["anomaly_value"] = pd.Series(dtype=int) and
model_df["anomaly_score"] = pd.Series(dtype=float) so that for a one-row
DataFrame you assign model_df["anomaly_value"] = 1 and model_df["anomaly_score"]
= 0.0 (or equivalent scalar assignment) before returning model_df to preserve
downstream filtering semantics (e.g., df[df["anomaly_value"] == -1]).
Comment on lines
+138
to
+145
| with mock.patch("airqo_etl_utils.bigquery_api.bigquery.Client"): | ||
| api = BigQueryApi() | ||
| api.client = mock_bigquery_client | ||
| api.client.query.return_value.result.return_value.to_dataframe.return_value = ( | ||
| pd.DataFrame() | ||
| ) | ||
| with pytest.raises(Exception) as e: | ||
| df = api.fetch_raw_readings() | ||
| assert "No data found" in str(e.value) | ||
| with pytest.raises(ValueError, match="No data found from bigquery"): | ||
| api.fetch_raw_readings() |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
python - <<'PY'
from unittest import mock
m = mock.Mock()
m.side_effect = lambda *args, **kwargs: "side_effect"
m.return_value = "return_value"
assert m() == "side_effect"
print("Mock.side_effect takes precedence over Mock.return_value")
PYRepository: airqo-platform/AirQo-api
Length of output: 124
🏁 Script executed:
#!/bin/bash
# Find the test file and check the fixture definition for mock_bigquery_client
cd /tmp && git clone https://github.com/airqo-platform/AirQo-api.git 2>&1 | grep -E "(Cloning|fatal)" || true
# Search for mock_bigquery_client fixture definition
rg -n "mock_bigquery_client" src/workflows/airqo_etl_utils/tests/test_bigquery_api.py -B 5 -A 10Repository: airqo-platform/AirQo-api
Length of output: 209
🏁 Script executed:
#!/bin/bash
# Get context around lines 138-145, including the fixture and test function
rg -n "@pytest.fixture|def test_fetch_raw_readings_empty|def mock_bigquery_client" src/workflows/airqo_etl_utils/tests/test_bigquery_api.py -A 15Repository: airqo-platform/AirQo-api
Length of output: 1788
🏁 Script executed:
#!/bin/bash
# Look at the full test function and fixture setup
sed -n '120,180p' src/workflows/airqo_etl_utils/tests/test_bigquery_api.pyRepository: airqo-platform/AirQo-api
Length of output: 2314
🏁 Script executed:
#!/bin/bash
# Get the complete mock_bigquery_client fixture definition
sed -n '12,136p' src/workflows/airqo_etl_utils/tests/test_bigquery_api.pyRepository: airqo-platform/AirQo-api
Length of output: 4328
🏁 Script executed:
#!/bin/bash
# Also check if fake_query or side_effect is set on fake_client.query
rg -n "fake_query|side_effect" src/workflows/airqo_etl_utils/tests/test_bigquery_api.py -B 2 -A 2Repository: airqo-platform/AirQo-api
Length of output: 766
Clear the existing side_effect before stubbing the empty result.
Line 141 sets return_value, but the fixture already configured api.client.query.side_effect to fake_query; side_effect takes precedence, so the test may not exercise the intended empty DataFrame path. Add api.client.query.side_effect = None before configuring return_value.
🧪 Proposed test fix
with mock.patch("airqo_etl_utils.bigquery_api.bigquery.Client"):
api = BigQueryApi()
api.client = mock_bigquery_client
+ api.client.query.side_effect = None
api.client.query.return_value.result.return_value.to_dataframe.return_value = (
pd.DataFrame()
)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/tests/test_bigquery_api.py` around lines 138 -
145, The test sets api.client.query.side_effect earlier which overrides
return_value; to exercise the empty-DataFrame branch clear that side_effect
before stubbing the return. In the test for BigQueryApi.fetch_raw_readings, set
api.client.query.side_effect = None (or delete the side_effect) prior to
assigning
api.client.query.return_value.result.return_value.to_dataframe.return_value =
pd.DataFrame() so the return_value path is used and the ValueError "No data
found from bigquery" is raised.
| @@ -1,4 +1,5 @@ | |||
| import pytest | |||
| from airqo_etl_utils.sql import query_manager as default_query_manager | |||
Contributor
There was a problem hiding this comment.
Assert the lookback_days placeholder too.
This test loads the real SQL template, but only checks raw_measurements_table. Since fetch_raw_readings() passes lookback_days and str.format() ignores extra kwargs, the checked-in SQL could accidentally hardcode/drop the lookback window while tests still pass.
Proposed test tightening
def test_fault_detection_query_is_loaded_from_faultdetection_sql_dir():
q = default_query_manager.get_query("fault_detection_raw_device_readings")
assert q.source.parent.name == "faultdetection"
assert q.source.suffix == ".sql"
assert "raw_measurements_table" in q.placeholders
+ assert "lookback_days" in q.placeholdersAlso applies to: 69-74
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/tests/test_query_manager.py` at line 2, The
test currently only verifies that the SQL template includes the
raw_measurements_table replacement but misses the lookback_days placeholder,
allowing accidental hardcoding; update the test for fetch_raw_readings() to also
assert the lookback_days substitution by either (a) passing a known
lookback_days value into query_manager.fetch_raw_readings() and asserting the
returned SQL string contains that numeric/string value, or (b) if testing the
template before format, assert the template contains the "{lookback_days}"
token; apply the same additional assertion to the other test block around lines
69-74 and reference the fetch_raw_readings and raw_measurements_table symbols
when making the assertion.
…into faultdetection
…d configuration updates - Added fault detection model training DAG to train and save Isolation Forest model. - Updated fault detection workflow to utilize trained model and handle missing models gracefully. - Introduced configuration parameters for training lookback days and model storage. - Enhanced documentation for fault detection processes and model training.
…into faultdetection
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (2)
src/workflows/airqo_etl_utils/ml_utils.py (2)
760-769:⚠️ Potential issue | 🟠 MajorRange fault still triggers on a single invalid reading for sparse devices.
This hasn't changed since the earlier round: with
or, one invalid value on a device that has only a handful of readings givesinvalid_share == 100%, bypassingMIN_INVALID_VALUE_COUNTand marking the device faulty from a single transient outlier. Switching toandkeeps both guards in force.🛡️ Proposed debounce fix
invalid_count = ((clean_series < lower) | (clean_series > upper)).sum() invalid_share = invalid_count / len(clean_series) return bool( invalid_count >= FaultDetectionUtils.MIN_INVALID_VALUE_COUNT - or invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD + and invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 760 - 769, The range-fault check in _has_invalid_value_fault currently triggers when either invalid_count >= FaultDetectionUtils.MIN_INVALID_VALUE_COUNT or invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD; change this to require both conditions (use logical AND) so a device is marked faulty only when there are at least MIN_INVALID_VALUE_COUNT invalid readings and the invalid_share meets or exceeds INVALID_VALUE_SHARE_THRESHOLD, preserving both guards for sparse devices.
980-983:⚠️ Potential issue | 🟡 MinorAssigning empty Series leaves
anomaly_value/anomaly_scoreas NaN on the single-row branch.Same concern as the earlier round and still present. When
model_dfhas exactly one row,pd.Series(dtype=int)broadcasts toNaN, which breaks every downstreamdf["anomaly_value"] == -1filter (NaN silently compares False) and poisons aggregations. Assigning the scalar "normal" defaults keeps the single-row path well-defined.🛡️ Proposed fix
if model_df.empty or len(model_df.index) < 2: - model_df["anomaly_value"] = pd.Series(dtype=int) - model_df["anomaly_score"] = pd.Series(dtype=float) + model_df["anomaly_value"] = 1 + model_df["anomaly_score"] = 0.0 return model_df🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 980 - 983, The branch that handles model_df.empty or len(model_df.index) < 2 assigns empty pd.Series which yields NaN for a single-row DataFrame; change the assignments in that conditional (the block referencing model_df, anomaly_value, anomaly_score) to set scalar default values instead (e.g., model_df["anomaly_value"] = 0 and model_df["anomaly_score"] = 0.0 or whatever your "normal" sentinel values are) so single-row DataFrames get concrete defaults and downstream filters like df["anomaly_value"] == -1 behave correctly.
🧹 Nitpick comments (3)
src/workflows/airqo_etl_utils/ml_utils.py (1)
1238-1245: Tiny readability nit.Ruff RUF005 prefers iterable unpacking over list concatenation inside the comprehension's source iterable:
fault_flag_columns = [ column - for column in ( - FaultDetectionUtils.RULE_FAULT_COLUMNS - + ["anomaly_percentage_fault", "anomaly_sequence_fault"] - ) + for column in [ + *FaultDetectionUtils.RULE_FAULT_COLUMNS, + "anomaly_percentage_fault", + "anomaly_sequence_fault", + ] if column in merged_df.columns ]🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1238 - 1245, Replace the list concatenation inside the comprehension that builds fault_flag_columns with iterable unpacking for readability: iterate over (*FaultDetectionUtils.RULE_FAULT_COLUMNS, "anomaly_percentage_fault", "anomaly_sequence_fault") instead of FaultDetectionUtils.RULE_FAULT_COLUMNS + [...], keeping the existing filter if column in merged_df.columns and preserving the variable name fault_flag_columns and use of merged_df.src/workflows/dags/fault_detection_training_job.py (1)
24-46: Heads-up on XCom payload size with 90-day training data.With
FAULT_DETECTION_TRAINING_LOOKBACK_DAYS=90as the default, the raw readings DataFrame and then the engineered feature DataFrame are both handed between tasks via Airflow's TaskFlow XCom. On the default metadata-DB XCom backend this can balloon quickly (hundreds of MB at fleet scale once lag/rolling features are added), leading to slow serialization and DB bloat.If you've seen this manifest in other training DAGs you may want to either:
- switch to a custom XCom backend (e.g. GCS-backed
BaseXComsubclass),- or collapse fetch/prepare/train into a single task so the intermediate DataFrames stay in one worker's memory.
Not blocking, just worth sizing before this runs in prod.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/dags/fault_detection_training_job.py` around lines 24 - 46, The DAG currently passes large DataFrame payloads between tasks via TaskFlow XCom (fetch_training_data -> prepare_pattern_detection_features -> train_isolation_forest_model) with FAULT_DETECTION_TRAINING_LOOKBACK_DAYS=90, which can bloat the metadata DB; remedy by either switching to a non-DB XCom backend (e.g., implement and configure a GCS-backed BaseXCom) and update the tasks to push/pull XCom pointers/URIs instead of full DataFrames, or collapse fetch_training_data, prepare_pattern_detection_features, and train_isolation_forest_model into a single task so data remains in-worker memory and is not serialized; ensure references to fetch_fault_detection_raw_readings, FaultDetectionUtils.prepare_pattern_detection_features, and FaultDetectionUtils.train_and_save_isolation_forest are adjusted to read/write from the chosen external storage or operate within the combined task.src/workflows/dags/fault_detection_job.py (1)
36-38: Consistency nit:initialize_device_fault_statushas nodoc_md.Every other task in this DAG got a Markdown doc string — this one is the odd task out with a bare
@task(). Not functional, just spoils the tidy "every task self-documents" story in the Airflow UI.- `@task`() + `@task`(doc_md=initialize_device_fault_status_doc) def initialize_device_fault_status(data): return FaultDetectionUtils.initialize_device_fault_status(data)(And add the corresponding
initialize_device_fault_status_docimport if you spin up a doc string for it intask_docs.py.)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/dags/fault_detection_job.py` around lines 36 - 38, The task initialize_device_fault_status lacks a doc_md like the other tasks; add a Markdown doc string by updating the `@task` decorator to include doc_md=initialize_device_fault_status_doc and import initialize_device_fault_status_doc from task_docs, keeping the function name initialize_device_fault_status and its call to FaultDetectionUtils.initialize_device_fault_status unchanged so the Airflow UI shows the same self-documentation as the other tasks.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/workflows/airqo_etl_utils/ml_utils.py`:
- Around line 722-724: The method _get_isolation_forest_model_path currently
returns configuration.FAULT_DETECTION_MODEL_PATH without validation; add a check
that raises a clear exception (e.g., ValueError) if
configuration.FAULT_DETECTION_MODEL_PATH is missing/None/empty so callers get an
immediate config error instead of propagating None into _save_model_object,
storage.load_file_object, or the temporary path f-string (e.g.,
"/tmp/{model_path}"). Update _get_isolation_forest_model_path to mirror the
behavior of _get_model_bucket by validating the config value and raising with a
descriptive message referencing FAULT_DETECTION_MODEL_PATH.
- Around line 1084-1096: The current loader uses a hard-coded world-writable
cache path and a broad except that hides real errors; update the call in the
Isolation Forest loader (where
FaultDetectionUtils._get_isolation_forest_model_path and
FaultDetectionUtils._load_model_object are used) to build a safe cache path
using tempfile.gettempdir() and os.path.basename(model_path) (or an
airflow-owned/cache dir) instead of f"/tmp/{model_path}", and replace the broad
"except Exception" with targeted handling: catch FileNotFoundError and
google.api_core.exceptions.NotFound to handle missing-artifact cases (log and
return None), but re-raise any other exceptions so auth/IO errors surface to the
caller.
---
Duplicate comments:
In `@src/workflows/airqo_etl_utils/ml_utils.py`:
- Around line 760-769: The range-fault check in _has_invalid_value_fault
currently triggers when either invalid_count >=
FaultDetectionUtils.MIN_INVALID_VALUE_COUNT or invalid_share >=
FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD; change this to require both
conditions (use logical AND) so a device is marked faulty only when there are at
least MIN_INVALID_VALUE_COUNT invalid readings and the invalid_share meets or
exceeds INVALID_VALUE_SHARE_THRESHOLD, preserving both guards for sparse
devices.
- Around line 980-983: The branch that handles model_df.empty or
len(model_df.index) < 2 assigns empty pd.Series which yields NaN for a
single-row DataFrame; change the assignments in that conditional (the block
referencing model_df, anomaly_value, anomaly_score) to set scalar default values
instead (e.g., model_df["anomaly_value"] = 0 and model_df["anomaly_score"] = 0.0
or whatever your "normal" sentinel values are) so single-row DataFrames get
concrete defaults and downstream filters like df["anomaly_value"] == -1 behave
correctly.
---
Nitpick comments:
In `@src/workflows/airqo_etl_utils/ml_utils.py`:
- Around line 1238-1245: Replace the list concatenation inside the comprehension
that builds fault_flag_columns with iterable unpacking for readability: iterate
over (*FaultDetectionUtils.RULE_FAULT_COLUMNS, "anomaly_percentage_fault",
"anomaly_sequence_fault") instead of FaultDetectionUtils.RULE_FAULT_COLUMNS +
[...], keeping the existing filter if column in merged_df.columns and preserving
the variable name fault_flag_columns and use of merged_df.
In `@src/workflows/dags/fault_detection_job.py`:
- Around line 36-38: The task initialize_device_fault_status lacks a doc_md like
the other tasks; add a Markdown doc string by updating the `@task` decorator to
include doc_md=initialize_device_fault_status_doc and import
initialize_device_fault_status_doc from task_docs, keeping the function name
initialize_device_fault_status and its call to
FaultDetectionUtils.initialize_device_fault_status unchanged so the Airflow UI
shows the same self-documentation as the other tasks.
In `@src/workflows/dags/fault_detection_training_job.py`:
- Around line 24-46: The DAG currently passes large DataFrame payloads between
tasks via TaskFlow XCom (fetch_training_data ->
prepare_pattern_detection_features -> train_isolation_forest_model) with
FAULT_DETECTION_TRAINING_LOOKBACK_DAYS=90, which can bloat the metadata DB;
remedy by either switching to a non-DB XCom backend (e.g., implement and
configure a GCS-backed BaseXCom) and update the tasks to push/pull XCom
pointers/URIs instead of full DataFrames, or collapse fetch_training_data,
prepare_pattern_detection_features, and train_isolation_forest_model into a
single task so data remains in-worker memory and is not serialized; ensure
references to fetch_fault_detection_raw_readings,
FaultDetectionUtils.prepare_pattern_detection_features, and
FaultDetectionUtils.train_and_save_isolation_forest are adjusted to read/write
from the chosen external storage or operate within the combined task.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 9180f3df-dcbb-42dc-8e63-25361eb01205
📒 Files selected for processing (9)
src/workflows/airqo_etl_utils/bigquery_api.pysrc/workflows/airqo_etl_utils/config.pysrc/workflows/airqo_etl_utils/ml_utils.pysrc/workflows/airqo_etl_utils/tests/test_ml_utils.pysrc/workflows/dags/dag_docs.pysrc/workflows/dags/fault_detection_job.pysrc/workflows/dags/fault_detection_training_job.pysrc/workflows/dags/task_docs.pysrc/workflows/env.sample
✅ Files skipped from review due to trivial changes (3)
- src/workflows/env.sample
- src/workflows/dags/dag_docs.py
- src/workflows/dags/task_docs.py
🚧 Files skipped from review as they are similar to previous changes (2)
- src/workflows/airqo_etl_utils/bigquery_api.py
- src/workflows/airqo_etl_utils/config.py
Comment on lines
+1084
to
+1096
| model_path = FaultDetectionUtils._get_isolation_forest_model_path() | ||
| try: | ||
| model = FaultDetectionUtils._load_model_object( | ||
| source_file=model_path, | ||
| local_cache_path=f"/tmp/{model_path}", | ||
| ) | ||
| logger.info(f"Loaded Isolation Forest model from {bucket}/{model_path}") | ||
| return model | ||
| except Exception as e: | ||
| logger.warning( | ||
| f"Could not load Isolation Forest model: {e}." | ||
| ) | ||
| return None |
Contributor
There was a problem hiding this comment.
Cache path and silent failure handling warrant a small hardening pass.
A couple of mild concerns layered together here:
local_cache_path=f"/tmp/{model_path}"is a hard-coded, world-writable location (Ruff S108). On shared/multi-tenant hosts this is a pre-seeding vector, and ifmodel_pathever contains subdirectories (e.g.models/foo.pkl), the cache read still works but the intent becomes fuzzy. Prefertempfile.gettempdir()joined with a sanitized basename, or an explicit airflow-owned cache dir.- The broad
except Exception(Ruff BLE001) swallows all failures — including transient GCS/auth errors — and silently returnsNone, which then degrades to "no trained model" and (withtrain_if_missing=Falsein the detection DAG) surfaces as a misleadingRuntimeErrorat the call site. Narrowing toFileNotFoundError/google.api_core.exceptions.NotFoundfor the "missing artifact" path, while re-raising on real failures, would make the failure mode accurate.
🛡️ Sketch
- model_path = FaultDetectionUtils._get_isolation_forest_model_path()
+ import os
+ import tempfile
+
+ model_path = FaultDetectionUtils._get_isolation_forest_model_path()
+ cache_path = os.path.join(
+ tempfile.gettempdir(), os.path.basename(model_path)
+ )
try:
model = FaultDetectionUtils._load_model_object(
source_file=model_path,
- local_cache_path=f"/tmp/{model_path}",
+ local_cache_path=cache_path,
)
logger.info(f"Loaded Isolation Forest model from {bucket}/{model_path}")
return model
- except Exception as e:
- logger.warning(
- f"Could not load Isolation Forest model: {e}."
- )
- return None
+ except FileNotFoundError as e:
+ logger.warning(f"Isolation Forest model not found at {bucket}/{model_path}: {e}")
+ return None🧰 Tools
🪛 Ruff (0.15.11)
[error] 1088-1088: Probable insecure usage of temporary file or directory: "/tmp/"
(S108)
[warning] 1092-1092: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1084 - 1096, The
current loader uses a hard-coded world-writable cache path and a broad except
that hides real errors; update the call in the Isolation Forest loader (where
FaultDetectionUtils._get_isolation_forest_model_path and
FaultDetectionUtils._load_model_object are used) to build a safe cache path
using tempfile.gettempdir() and os.path.basename(model_path) (or an
airflow-owned/cache dir) instead of f"/tmp/{model_path}", and replace the broad
"except Exception" with targeted handling: catch FileNotFoundError and
google.api_core.exceptions.NotFound to handle missing-artifact cases (log and
return None), but re-raise any other exceptions so auth/IO errors surface to the
caller.
…into faultdetection
…raining parameters - Added device limit and minimum hourly records parameters to fetch_fault_detection_raw_readings method. - Updated SQL query to incorporate new parameters for filtering device readings. - Introduced configuration options for training device limit and minimum hourly records. - Enhanced tests to validate new functionality and ensure correct query generation.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (7)
src/workflows/dags/fault_detection_training_job.py (2)
46-48: Small consistency nit:save_isolation_forest_modelis the only task without adoc_md.The other three tasks each pull in a docstring constant from
task_docs; this last one uses bare@task(). A short doc string here would round out the Airflow UI nicely (and the empty parens can become a plain@taskwhile you're at it).📝 Suggested polish
- `@task`() - def save_isolation_forest_model(trained_model): - return FaultDetectionUtils.save_isolation_forest_model(trained_model) + `@task`(doc_md=save_fault_detection_model_doc) + def save_isolation_forest_model(trained_model): + return FaultDetectionUtils.save_isolation_forest_model(trained_model)…with a matching
save_fault_detection_model_docadded tosrc/workflows/dags/task_docs.pyand pulled into the import block at the top of this file.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/dags/fault_detection_training_job.py` around lines 46 - 48, Add a short docstring and make the decorator consistent: create a new doc constant (e.g., save_fault_detection_model_doc) in task_docs.py, import it into this module, change the decorator from `@task`() to `@task`, and pass doc_md=save_fault_detection_model_doc into the save_isolation_forest_model task definition (the function name is save_isolation_forest_model and the module that holds docs is task_docs.py) so the Airflow UI shows the same documentation as the other tasks.
25-53: GCS-backed XCom is already handling the payload concern correctly.The DAG hands large pandas DataFrames and a serialized IsolationForest artifact between tasks, but your environment has a custom
GCSXComBackendalready configured (set in docker-compose.yaml). This backend automatically persists DataFrames to GCS and stores only the file path in XCom, which elegantly sidesteps the row-size and performance issues of a default DB-backed XCom. This infrastructure is exactly what the training flow needs, so no changes required on that front.Minor: add
doc_mdtosave_isolation_forest_modelfor consistency. The other three tasks document themselves; the save task does not.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/dags/fault_detection_training_job.py` around lines 25 - 53, The save_isolation_forest_model task is missing a doc_md like the other tasks; update the `@task` decorator on save_isolation_forest_model to include a doc_md argument (use the existing train_fault_detection_model_doc or create a new save_isolation_forest_model_doc) so the task has consistent documentation with prepare_pattern_detection_features and train_isolation_forest_model and references the save_isolation_forest_model function in the decorator.src/workflows/airqo_etl_utils/bigquery_api.py (1)
1273-1278: The post-query hourly resample is redundant given the SQL's hourly bucketing.The SQL query groups by
(device_id, TIMESTAMP_TRUNC(t.timestamp, HOUR)), producing one row per (device, hour) bucket. The subsequentgroupby("device_id").resample("h", on="timestamp")[num_cols].mean()refills the entire hourly grid, inserting NaN for every missing hour in the lookback window. For a sparse device over 90 days, this balloons from ~24–48 records to ~2,160 rows, most NaN.Since
get_lag_and_roll_featuresuses positional shifts (.shift(),.rolling()) rather than timestamp-aware operations, it doesn't require a gap-filled grid. The NaN rows are dropped downstream anyway, so correctness is preserved—but the overhead is avoidable. If deduplication is the goal,drop_duplicates(["device_id", "timestamp"])would be much cheaper. Consider either removing the resample entirely or replacing it with a simpler dedupe operation.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/bigquery_api.py` around lines 1273 - 1278, The post-query hourly resample on results (the results.groupby("device_id").resample("h", on="timestamp")[num_cols].mean() block) is unnecessary and expands sparse devices into a full hourly grid; remove that resample and reset_index call and instead deduplicate the SQL rows by calling drop_duplicates(["device_id", "timestamp"]) on results after converting timestamp to UTC, or simply delete the resample step entirely so downstream get_lag_and_roll_features (which uses .shift()/.rolling()) works on the original hourly-bucketed rows without creating NaN-heavy expanded rows.src/workflows/airqo_etl_utils/ml_utils.py (4)
1550-1559: Use iterable unpacking instead of list concatenation (Ruff RUF005).A small readability nit on the fault-flag assembly — equivalent at runtime, but tidier and matches the lint rule.
♻️ Proposed tweak
fault_flag_columns = [ column for column in ( - FaultDetectionUtils.RULE_FAULT_COLUMNS - + ["anomaly_percentage_fault", "anomaly_sequence_fault"] + *FaultDetectionUtils.RULE_FAULT_COLUMNS, + "anomaly_percentage_fault", + "anomaly_sequence_fault", ) if column in merged_df.columns ]🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1550 - 1559, Replace the list concatenation used to build the fault-flag column iterable with iterable unpacking to satisfy the Ruff RUF005 lint rule: update the comprehension that constructs fault_flag_columns (which currently uses FaultDetectionUtils.RULE_FAULT_COLUMNS + ["anomaly_percentage_fault", "anomaly_sequence_fault"]) to use iterable unpacking (e.g., unpack FaultDetectionUtils.RULE_FAULT_COLUMNS together with the two anomaly names) so the comprehension iterates over a single tuple/iterable; keep the rest unchanged (the variable name fault_flag_columns and the cast to int on merged_df[fault_flag_columns]).
1444-1450: Promote the45%and80-run thresholds to named class constants.The
> 45cutoff appears here and again insave_faulty_devices(Line 1543), and the>= 80run-length cutoff is duplicated inprocess_faulty_devices_fault_sequence(Line 1503) and again at Line 1548. If product ever wants to retune fault sensitivity, drift between these copies would silently produce inconsistent flags for the same device across the two code paths.♻️ Suggested constants
class FaultDetectionUtils(BaseMlUtils): @@ + ANOMALY_PERCENTAGE_FAULT_THRESHOLD = 45 + ANOMALY_SEQUENCE_FAULT_THRESHOLD = 80…and reference
FaultDetectionUtils.ANOMALY_PERCENTAGE_FAULT_THRESHOLD/FaultDetectionUtils.ANOMALY_SEQUENCE_FAULT_THRESHOLDin all four sites.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1444 - 1450, Define two class-level constants on FaultDetectionUtils (e.g., ANOMALY_PERCENTAGE_FAULT_THRESHOLD = 45 and ANOMALY_SEQUENCE_FAULT_THRESHOLD = 80) and replace all hard-coded occurrences of > 45 and >= 80 with references to these constants; specifically update the use in ml_utils.py inside the methods that compute anomaly_percentage (where anomaly_percentage["anomaly_percentage"] > 45 is used), save_faulty_devices, and process_faulty_devices_fault_sequence so they all reference FaultDetectionUtils.ANOMALY_PERCENTAGE_FAULT_THRESHOLD and FaultDetectionUtils.ANOMALY_SEQUENCE_FAULT_THRESHOLD respectively, ensuring the logic and comparisons remain identical.
696-718: Annotate mutable class-level constants withClassVarfor type-checker friendliness.Ruff RUF012 flags
RULE_FAULT_COLUMNS,ML_FAULT_DEFAULTS, andUNSUPERVISED_SCORE_WEIGHTSbecause mutable defaults on a class are easy to mutate accidentally and confuse static type-checkers. Since these are intended as immutable lookups, atyping.ClassVarannotation makes the intent explicit and clears the lint at zero runtime cost.♻️ Suggested annotation
+from typing import ClassVar @@ - RULE_FAULT_COLUMNS = [ + RULE_FAULT_COLUMNS: ClassVar[List[str]] = [ "correlation_fault", "missing_data_fault", "sensor_disagreement_fault", "constant_value_fault", "battery_fault", "range_fault", ] - ML_FAULT_DEFAULTS = { + ML_FAULT_DEFAULTS: ClassVar[Dict[str, Any]] = { "anomaly_percentage": 0.0, "anomaly_count": 0, "observation_count": 0, "anomaly_percentage_fault": 0, "fault_count": 0, "anomaly_sequence_fault": 0, } ISOLATION_FOREST_CONTAMINATION = 0.37 - UNSUPERVISED_SCORE_WEIGHTS = { + UNSUPERVISED_SCORE_WEIGHTS: ClassVar[Dict[str, float]] = { "consistency": 0.35, "stability": 0.30, "silhouette": 0.20, "anomaly_realism": 0.15, }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 696 - 718, These module/class-level mutable constants (RULE_FAULT_COLUMNS, ML_FAULT_DEFAULTS, UNSUPERVISED_SCORE_WEIGHTS) should be annotated with typing.ClassVar to signal they are intended as immutable lookups to the type-checker; import ClassVar (and the appropriate generic types like List and Dict) and change their annotations to ClassVar[List[str]] for RULE_FAULT_COLUMNS, ClassVar[Dict[str, int | float | str]] (or a more specific mapping) for ML_FAULT_DEFAULTS, and ClassVar[Dict[str, float]] for UNSUPERVISED_SCORE_WEIGHTS so Ruff RUF012 is satisfied without changing runtime behavior. Ensure the existing constant names (RULE_FAULT_COLUMNS, ML_FAULT_DEFAULTS, UNSUPERVISED_SCORE_WEIGHTS) are left intact and only their type annotations are added.
1013-1016: ReuseISOLATION_FOREST_CONTAMINATIONinstead of the literal0.37.
train_isolation_forest(Line 1274) and_calculate_unsupervised_model_metrics(Line 1167) both go through the named constant; this in-line training fallback hardcodes the same value, so a future tweak to the class constant would silently desynchronize the inline-trained and artifact-trained models.♻️ Proposed alignment
isolation_forest = IsolationForest( - contamination=0.37, random_state=42, n_estimators=100 + contamination=FaultDetectionUtils.ISOLATION_FOREST_CONTAMINATION, + random_state=42, + n_estimators=100, )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1013 - 1016, Replace the hardcoded contamination value 0.37 with the shared constant ISOLATION_FOREST_CONTAMINATION where the IsolationForest is instantiated in the inline training path (the block that creates isolation_forest and calls isolation_forest.fit(model_df[feature_columns])); ensure this matches the usage in train_isolation_forest and _calculate_unsupervised_model_metrics so inline-trained models remain in sync with the artifact-trained models.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/workflows/airqo_etl_utils/bigquery_api.py`:
- Around line 1242-1251: The lookback_days assignment currently short-circuits
on falsy values and lacks a positivity guard; change the logic around the
lookback_days variable so you only default when it is None (e.g., if
lookback_days is None: lookback_days =
configuration.FAULT_DETECTION_LOOKBACK_DAYS) and then add the same positivity
check used for minimum_hourly_records and device_limit (raise ValueError if
lookback_days <= 0) so negative or zero values are rejected locally before
reaching BigQuery; refer to the lookback_days variable and the existing checks
for minimum_hourly_records and device_limit in bigquery_api.py to place the new
validation.
In `@src/workflows/airqo_etl_utils/ml_utils.py`:
- Around line 1502-1508: The filtered slice assigned to faulty_devices_df can be
a view which causes SettingWithCopy issues; when creating faulty_devices_df from
device_max_anomaly_sequence[...] use .copy() (e.g., faulty_devices_df =
device_max_anomaly_sequence[mask].copy()) or use .assign(...) so that setting
the anomaly_sequence_fault column on faulty_devices_df is performed on an
explicit copy; update the creation of faulty_devices_df (and any subsequent
assignment to anomaly_sequence_fault) accordingly.
- Around line 1078-1100: Add an explicit presence check for the "timestamp"
column at the start of _prepare_isolation_forest_training_data (before calling
pd.to_datetime) so that if "timestamp" is missing you raise a clear ValueError
(e.g. "Missing 'timestamp' column in training_data"); mirror the defensive
pattern used in prepare_pattern_detection_features and flag_pattern_based_faults
to fail fast with a descriptive error rather than letting a KeyError surface
later.
---
Nitpick comments:
In `@src/workflows/airqo_etl_utils/bigquery_api.py`:
- Around line 1273-1278: The post-query hourly resample on results (the
results.groupby("device_id").resample("h", on="timestamp")[num_cols].mean()
block) is unnecessary and expands sparse devices into a full hourly grid; remove
that resample and reset_index call and instead deduplicate the SQL rows by
calling drop_duplicates(["device_id", "timestamp"]) on results after converting
timestamp to UTC, or simply delete the resample step entirely so downstream
get_lag_and_roll_features (which uses .shift()/.rolling()) works on the original
hourly-bucketed rows without creating NaN-heavy expanded rows.
In `@src/workflows/airqo_etl_utils/ml_utils.py`:
- Around line 1550-1559: Replace the list concatenation used to build the
fault-flag column iterable with iterable unpacking to satisfy the Ruff RUF005
lint rule: update the comprehension that constructs fault_flag_columns (which
currently uses FaultDetectionUtils.RULE_FAULT_COLUMNS +
["anomaly_percentage_fault", "anomaly_sequence_fault"]) to use iterable
unpacking (e.g., unpack FaultDetectionUtils.RULE_FAULT_COLUMNS together with the
two anomaly names) so the comprehension iterates over a single tuple/iterable;
keep the rest unchanged (the variable name fault_flag_columns and the cast to
int on merged_df[fault_flag_columns]).
- Around line 1444-1450: Define two class-level constants on FaultDetectionUtils
(e.g., ANOMALY_PERCENTAGE_FAULT_THRESHOLD = 45 and
ANOMALY_SEQUENCE_FAULT_THRESHOLD = 80) and replace all hard-coded occurrences of
> 45 and >= 80 with references to these constants; specifically update the use
in ml_utils.py inside the methods that compute anomaly_percentage (where
anomaly_percentage["anomaly_percentage"] > 45 is used), save_faulty_devices, and
process_faulty_devices_fault_sequence so they all reference
FaultDetectionUtils.ANOMALY_PERCENTAGE_FAULT_THRESHOLD and
FaultDetectionUtils.ANOMALY_SEQUENCE_FAULT_THRESHOLD respectively, ensuring the
logic and comparisons remain identical.
- Around line 696-718: These module/class-level mutable constants
(RULE_FAULT_COLUMNS, ML_FAULT_DEFAULTS, UNSUPERVISED_SCORE_WEIGHTS) should be
annotated with typing.ClassVar to signal they are intended as immutable lookups
to the type-checker; import ClassVar (and the appropriate generic types like
List and Dict) and change their annotations to ClassVar[List[str]] for
RULE_FAULT_COLUMNS, ClassVar[Dict[str, int | float | str]] (or a more specific
mapping) for ML_FAULT_DEFAULTS, and ClassVar[Dict[str, float]] for
UNSUPERVISED_SCORE_WEIGHTS so Ruff RUF012 is satisfied without changing runtime
behavior. Ensure the existing constant names (RULE_FAULT_COLUMNS,
ML_FAULT_DEFAULTS, UNSUPERVISED_SCORE_WEIGHTS) are left intact and only their
type annotations are added.
- Around line 1013-1016: Replace the hardcoded contamination value 0.37 with the
shared constant ISOLATION_FOREST_CONTAMINATION where the IsolationForest is
instantiated in the inline training path (the block that creates
isolation_forest and calls isolation_forest.fit(model_df[feature_columns]));
ensure this matches the usage in train_isolation_forest and
_calculate_unsupervised_model_metrics so inline-trained models remain in sync
with the artifact-trained models.
In `@src/workflows/dags/fault_detection_training_job.py`:
- Around line 46-48: Add a short docstring and make the decorator consistent:
create a new doc constant (e.g., save_fault_detection_model_doc) in
task_docs.py, import it into this module, change the decorator from `@task`() to
`@task`, and pass doc_md=save_fault_detection_model_doc into the
save_isolation_forest_model task definition (the function name is
save_isolation_forest_model and the module that holds docs is task_docs.py) so
the Airflow UI shows the same documentation as the other tasks.
- Around line 25-53: The save_isolation_forest_model task is missing a doc_md
like the other tasks; update the `@task` decorator on save_isolation_forest_model
to include a doc_md argument (use the existing train_fault_detection_model_doc
or create a new save_isolation_forest_model_doc) so the task has consistent
documentation with prepare_pattern_detection_features and
train_isolation_forest_model and references the save_isolation_forest_model
function in the decorator.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 1374942b-a34d-408f-aba8-a7622fdf97e7
📒 Files selected for processing (8)
src/workflows/airqo_etl_utils/bigquery_api.pysrc/workflows/airqo_etl_utils/config.pysrc/workflows/airqo_etl_utils/ml_utils.pysrc/workflows/airqo_etl_utils/sql/faultdetection/2026041701fault_detection.sqlsrc/workflows/airqo_etl_utils/tests/test_bigquery_api.pysrc/workflows/airqo_etl_utils/tests/test_ml_utils.pysrc/workflows/dags/fault_detection_training_job.pysrc/workflows/env.sample
✅ Files skipped from review due to trivial changes (1)
- src/workflows/env.sample
🚧 Files skipped from review as they are similar to previous changes (2)
- src/workflows/airqo_etl_utils/sql/faultdetection/2026041701fault_detection.sql
- src/workflows/airqo_etl_utils/config.py
Comment on lines
+1242
to
+1251
| lookback_days = lookback_days or configuration.FAULT_DETECTION_LOOKBACK_DAYS | ||
| if minimum_hourly_records <= 0: | ||
| raise ValueError( | ||
| f"minimum_hourly_records must be positive, got {minimum_hourly_records}" | ||
| ) | ||
| device_limit_clause = "" | ||
| if device_limit is not None: | ||
| if device_limit <= 0: | ||
| raise ValueError(f"device_limit must be positive, got {device_limit}") | ||
| device_limit_clause = f"ORDER BY RAND()\n LIMIT {int(device_limit)}" |
Contributor
There was a problem hiding this comment.
lookback_days is missing the same positivity check applied to its siblings.
lookback_days = lookback_days or configuration.FAULT_DETECTION_LOOKBACK_DAYS short-circuits on any falsy value, so lookback_days=0 silently falls back to the config default, and a negative value (e.g. -5) flows straight into the SQL as INTERVAL -5 DAY, only to fail later at the BigQuery layer with a less helpful error. Given you already validate minimum_hourly_records and device_limit, mirroring the same guard here keeps the surface consistent and the error messages local.
🛡️ Proposed fix
- lookback_days = lookback_days or configuration.FAULT_DETECTION_LOOKBACK_DAYS
+ if lookback_days is None:
+ lookback_days = configuration.FAULT_DETECTION_LOOKBACK_DAYS
+ if lookback_days <= 0:
+ raise ValueError(
+ f"lookback_days must be positive, got {lookback_days}"
+ )
if minimum_hourly_records <= 0:
raise ValueError(
f"minimum_hourly_records must be positive, got {minimum_hourly_records}"
)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/bigquery_api.py` around lines 1242 - 1251, The
lookback_days assignment currently short-circuits on falsy values and lacks a
positivity guard; change the logic around the lookback_days variable so you only
default when it is None (e.g., if lookback_days is None: lookback_days =
configuration.FAULT_DETECTION_LOOKBACK_DAYS) and then add the same positivity
check used for minimum_hourly_records and device_limit (raise ValueError if
lookback_days <= 0) so negative or zero values are rejected locally before
reaching BigQuery; refer to the lookback_days variable and the existing checks
for minimum_hourly_records and device_limit in bigquery_api.py to place the new
validation.
Comment on lines
+1078
to
+1100
| def _prepare_isolation_forest_training_data( | ||
| training_data: pd.DataFrame, | ||
| ) -> Tuple[pd.DataFrame, List[str], Dict[str, float]]: | ||
| training_data = FaultDetectionUtils._ensure_dataframe(training_data).copy() | ||
| training_data.dropna(subset=["device_id"], inplace=True) | ||
| training_data["timestamp"] = pd.to_datetime(training_data["timestamp"]) | ||
|
|
||
| feature_columns = FaultDetectionUtils._get_isolation_forest_feature_columns( | ||
| training_data | ||
| ) | ||
| if not feature_columns: | ||
| raise ValueError("No numeric feature columns available for training") | ||
|
|
||
| feature_medians = training_data[feature_columns].median(numeric_only=True) | ||
| feature_columns = [ | ||
| column for column in feature_columns if not pd.isna(feature_medians[column]) | ||
| ] | ||
| if not feature_columns: | ||
| raise ValueError("No usable numeric feature columns available for training") | ||
|
|
||
| training_data = training_data.dropna(subset=feature_columns).copy() | ||
| if training_data.empty: | ||
| raise ValueError("No complete feature rows available for training") |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm every caller of train_isolation_forest / _prepare_isolation_forest_training_data
# already provides a 'timestamp' column.
rg -nP --type=py -C3 '\b(train_isolation_forest|_prepare_isolation_forest_training_data|train_and_save_isolation_forest)\s*\('Repository: airqo-platform/AirQo-api
Length of output: 5919
🏁 Script executed:
#!/bin/bash
# Check what data is passed to train_isolation_forest_model in the DAG
# Look at upstream tasks that feed data into it
rg -nB20 'train_isolation_forest_model' src/workflows/dags/fault_detection_training_job.py | head -60Repository: airqo-platform/AirQo-api
Length of output: 1433
🏁 Script executed:
#!/bin/bash
# Check prepare_pattern_detection_features to see what columns it includes
rg -nA50 'def prepare_pattern_detection_features' src/workflows/airqo_etl_utils/ml_utils.py | head -80Repository: airqo-platform/AirQo-api
Length of output: 2227
🏁 Script executed:
#!/bin/bash
# Check the source of the 'data' variable in the DAG task
rg -nB5 '@task.*def train_isolation_forest_model' src/workflows/dags/fault_detection_training_job.py | head -20Repository: airqo-platform/AirQo-api
Length of output: 50
Add timestamp column validation for consistency and clearer error handling.
All current callers pass data through prepare_pattern_detection_features (which validates timestamp is present), but _prepare_isolation_forest_training_data should mirror the defensive validation pattern used by prepare_pattern_detection_features and flag_pattern_based_faults. Adding an explicit check ensures that any future direct callers fail fast with a descriptive ValueError rather than a bare KeyError.
Suggested addition
training_data = FaultDetectionUtils._ensure_dataframe(training_data).copy()
+ if "timestamp" not in training_data.columns:
+ raise ValueError(
+ "Training data must include a 'timestamp' column."
+ )
training_data.dropna(subset=["device_id"], inplace=True)
training_data["timestamp"] = pd.to_datetime(training_data["timestamp"])🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1078 - 1100, Add an
explicit presence check for the "timestamp" column at the start of
_prepare_isolation_forest_training_data (before calling pd.to_datetime) so that
if "timestamp" is missing you raise a clear ValueError (e.g. "Missing
'timestamp' column in training_data"); mirror the defensive pattern used in
prepare_pattern_detection_features and flag_pattern_based_faults to fail fast
with a descriptive error rather than letting a KeyError surface later.
Comment on lines
1502
to
1508
| faulty_devices_df = device_max_anomaly_sequence[ | ||
| device_max_anomaly_sequence["anomaly_sequence_length"] >= 80 | ||
| device_max_anomaly_sequence["fault_count"] >= 80 | ||
| ] | ||
| faulty_devices_df.columns = ["device_id", "fault_count"] | ||
| if not faulty_devices_df.empty: | ||
| faulty_devices_df["anomaly_sequence_fault"] = 1 | ||
|
|
||
| return faulty_devices_df |
Contributor
There was a problem hiding this comment.
Take a .copy() on the filtered slice before assigning anomaly_sequence_fault.
device_max_anomaly_sequence[mask] returns a view-or-copy that pandas typically warns on when you write to it (SettingWithCopyWarning). The module-level pd.options.mode.chained_assignment = None (Line 45) silences the warning globally, but the underlying ambiguity still bites when someone later tweaks the function or removes the global mute. A defensive .copy() (or .assign(...)) makes the intent explicit and future-proof.
🛡️ Proposed fix
- faulty_devices_df = device_max_anomaly_sequence[
- device_max_anomaly_sequence["fault_count"] >= 80
- ]
- if not faulty_devices_df.empty:
- faulty_devices_df["anomaly_sequence_fault"] = 1
+ faulty_devices_df = device_max_anomaly_sequence.loc[
+ device_max_anomaly_sequence["fault_count"] >= 80
+ ].copy()
+ if not faulty_devices_df.empty:
+ faulty_devices_df["anomaly_sequence_fault"] = 1🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1502 - 1508, The
filtered slice assigned to faulty_devices_df can be a view which causes
SettingWithCopy issues; when creating faulty_devices_df from
device_max_anomaly_sequence[...] use .copy() (e.g., faulty_devices_df =
device_max_anomaly_sequence[mask].copy()) or use .assign(...) so that setting
the anomaly_sequence_fault column on faulty_devices_df is performed on an
explicit copy; update the creation of faulty_devices_df (and any subsequent
assignment to anomaly_sequence_fault) accordingly.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (5)
src/workflows/airqo_etl_utils/ml_utils.py (5)
788-797:⚠️ Potential issue | 🟠 MajorRequire both invalid count and invalid share before raising
range_fault.The current
orlets a single bad reading on a sparse series triprange_faultviainvalid_share == 1.0, even whenMIN_INVALID_VALUE_COUNTis not met. That makes transient outliers look like device faults.Suggested fix
return bool( invalid_count >= FaultDetectionUtils.MIN_INVALID_VALUE_COUNT - or invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD + and invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 788 - 797, The _has_invalid_value_fault function currently flags a range fault when either invalid_count >= FaultDetectionUtils.MIN_INVALID_VALUE_COUNT OR invalid_share >= FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD; change this logic to require both conditions (use logical AND) so a fault is raised only when invalid_count meets MIN_INVALID_VALUE_COUNT and invalid_share meets INVALID_VALUE_SHARE_THRESHOLD, keeping the existing NaN drop and empty-series short-circuit as-is.
1085-1087:⚠️ Potential issue | 🟡 MinorFail fast when
training_datahas notimestampcolumn.This path currently falls through to a bare
KeyErrorat Line 1087. An explicit presence check would keep direct callers aligned with the clearer validation used elsewhere in this fault-detection flow.Suggested fix
training_data = FaultDetectionUtils._ensure_dataframe(training_data).copy() + if "timestamp" not in training_data.columns: + raise ValueError("Training data must include a 'timestamp' column.") training_data.dropna(subset=["device_id"], inplace=True) training_data["timestamp"] = pd.to_datetime(training_data["timestamp"])🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1085 - 1087, The code calls FaultDetectionUtils._ensure_dataframe(training_data) then assumes a "timestamp" column exists and calls pd.to_datetime, which leads to a bare KeyError when missing; add an explicit presence check after training_data = FaultDetectionUtils._ensure_dataframe(training_data).copy() (and after dropna on "device_id") to verify "timestamp" in training_data.columns and raise a clear, descriptive exception (e.g., ValueError("training_data must include 'timestamp' column")) before calling training_data["timestamp"] = pd.to_datetime(...), so callers get a validated error rather than a KeyError.
1013-1016:⚠️ Potential issue | 🟡 MinorUse explicit normal defaults for insufficient-data anomaly results.
Assigning empty
Serieshere leavesanomaly_valueandanomaly_scoreasNaNfor one-row inputs. Downstream code then sees an ambiguous state instead of a conservative “normal” prediction.Suggested fix
if model_df.empty or len(model_df.index) < 2: - model_df["anomaly_value"] = pd.Series(dtype=int) - model_df["anomaly_score"] = pd.Series(dtype=float) + model_df["anomaly_value"] = 1 + model_df["anomaly_score"] = 0.0 return model_dfAlso applies to: 1033-1036
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1013 - 1016, The branch that handles insufficient data (checking model_df.empty or len(model_df.index) < 2) currently assigns empty Series which leaves NaNs for single-row inputs; change it to assign explicit conservative defaults by setting model_df["anomaly_value"] = 0 and model_df["anomaly_score"] = 0.0 (instead of empty Series) so downstream logic sees a clear "normal" result, and apply the same change to the similar block around the second occurrence (the other branch at the 1033-1036 area); update only the assignments for model_df["anomaly_value"] and model_df["anomaly_score"] within this function in ml_utils.py.
1406-1418:⚠️ Potential issue | 🟠 MajorPropagate real model-load failures instead of masking them as “missing model”.
The broad catch here converts auth/network/cache errors into
None, so callers withtrain_if_missing=Falselater raise the misleading “No trained fault-detection Isolation Forest model is available” error instead of the real failure. The fixed/tmpcache path is also still worth hardening.Suggested fix
+ import tempfile + - cache_path = Path("/tmp") / Path(model_path).name + cache_path = Path(tempfile.gettempdir()) / Path(model_path).name try: model = FaultDetectionUtils._load_model_object( source_file=model_path, local_cache_path=str(cache_path), ) logger.info(f"Loaded Isolation Forest model from {bucket}/{model_path}") return model - except Exception as e: + except (FileNotFoundError, google_api_exceptions.NotFound) as e: logger.warning( f"Could not load Isolation Forest model: {e}." ) return None + except Exception: + logger.exception( + "Failed to load Isolation Forest model from %s/%s", + bucket, + model_path, + ) + raise🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1406 - 1418, The current broad try/except around FaultDetectionUtils._load_model_object masks real failures by returning None; change it to only swallow a genuine "model missing" case and re-raise other exceptions so callers see auth/network/cache errors. Concretely: build cache_path from tempfile.gettempdir() or tempfile.mkdtemp(...) (instead of hard-coded "/tmp") and call FaultDetectionUtils._load_model_object with source_file=model_path, local_cache_path=str(cache_path); catch only the specific not-found/FileNotFoundError (or whatever sentinel your loader raises) and return None in that case, but for any other Exception log it with logger.exception or logger.error including the exception context and re-raise it. Ensure references: model_path, cache_path, FaultDetectionUtils._load_model_object, logger.warning/logger.exception are updated accordingly.
1543-1569:⚠️ Potential issue | 🟡 MinorAvoid the blanket
fillna(0)after the outer merge.This turns null
device_namecells into integer0, so the fallback at Lines 1544-1545 only helps when the column is missing entirely, not when specific rows are null. Filling numeric fault columns explicitly and backfillingdevice_namefromdevice_idpreserves the metadata.Suggested fix
- merged_df = merged_df.fillna(0) if "device_name" not in merged_df.columns: merged_df["device_name"] = merged_df["device_id"] + else: + merged_df["device_name"] = merged_df["device_name"].fillna( + merged_df["device_id"] + ) merged_df["fault_detected"] = 1 for column, default_value in FaultDetectionUtils.ML_FAULT_DEFAULTS.items(): if column not in merged_df.columns: merged_df[column] = default_value + else: + merged_df[column] = merged_df[column].fillna(default_value) @@ if fault_flag_columns: + merged_df[fault_flag_columns] = merged_df[fault_flag_columns].fillna(0) merged_df[fault_flag_columns] = merged_df[fault_flag_columns].astype(int)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1543 - 1569, The blanket merged_df = merged_df.fillna(0) should be removed and replaced by targeted null-handling: do not coerce device_name to 0—backfill device_name from device_id (e.g., merged_df["device_name"] = merged_df["device_name"].fillna(merged_df["device_id"]) or equivalent) and explicitly fill only numeric fault columns defined by FaultDetectionUtils.ML_FAULT_DEFAULTS and FaultDetectionUtils.RULE_FAULT_COLUMNS with their default values (use those dict keys to locate columns and call fillna(default) per column), then proceed with computing anomaly_percentage_fault, anomaly_sequence_fault and casting fault_flag_columns to int as before.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/workflows/airqo_etl_utils/ml_utils.py`:
- Around line 1227-1238: The helper _load_existing_isolation_forest_metrics
currently swallows all exceptions and returns None which makes
save_isolation_forest_model treat transient read errors as "no previous model";
change it so only FileNotFoundError returns None, but any other Exception should
be logged and re-raised (or raise a custom RuntimeError) so callers like
save_isolation_forest_model abort instead of deploying unconditionally; update
the body around FaultDetectionUtils._load_model_object and
FaultDetectionUtils._get_isolation_forest_model_path to log the failure with
context and then re-raise the exception instead of returning None.
---
Duplicate comments:
In `@src/workflows/airqo_etl_utils/ml_utils.py`:
- Around line 788-797: The _has_invalid_value_fault function currently flags a
range fault when either invalid_count >=
FaultDetectionUtils.MIN_INVALID_VALUE_COUNT OR invalid_share >=
FaultDetectionUtils.INVALID_VALUE_SHARE_THRESHOLD; change this logic to require
both conditions (use logical AND) so a fault is raised only when invalid_count
meets MIN_INVALID_VALUE_COUNT and invalid_share meets
INVALID_VALUE_SHARE_THRESHOLD, keeping the existing NaN drop and empty-series
short-circuit as-is.
- Around line 1085-1087: The code calls
FaultDetectionUtils._ensure_dataframe(training_data) then assumes a "timestamp"
column exists and calls pd.to_datetime, which leads to a bare KeyError when
missing; add an explicit presence check after training_data =
FaultDetectionUtils._ensure_dataframe(training_data).copy() (and after dropna on
"device_id") to verify "timestamp" in training_data.columns and raise a clear,
descriptive exception (e.g., ValueError("training_data must include 'timestamp'
column")) before calling training_data["timestamp"] = pd.to_datetime(...), so
callers get a validated error rather than a KeyError.
- Around line 1013-1016: The branch that handles insufficient data (checking
model_df.empty or len(model_df.index) < 2) currently assigns empty Series which
leaves NaNs for single-row inputs; change it to assign explicit conservative
defaults by setting model_df["anomaly_value"] = 0 and model_df["anomaly_score"]
= 0.0 (instead of empty Series) so downstream logic sees a clear "normal"
result, and apply the same change to the similar block around the second
occurrence (the other branch at the 1033-1036 area); update only the assignments
for model_df["anomaly_value"] and model_df["anomaly_score"] within this function
in ml_utils.py.
- Around line 1406-1418: The current broad try/except around
FaultDetectionUtils._load_model_object masks real failures by returning None;
change it to only swallow a genuine "model missing" case and re-raise other
exceptions so callers see auth/network/cache errors. Concretely: build
cache_path from tempfile.gettempdir() or tempfile.mkdtemp(...) (instead of
hard-coded "/tmp") and call FaultDetectionUtils._load_model_object with
source_file=model_path, local_cache_path=str(cache_path); catch only the
specific not-found/FileNotFoundError (or whatever sentinel your loader raises)
and return None in that case, but for any other Exception log it with
logger.exception or logger.error including the exception context and re-raise
it. Ensure references: model_path, cache_path,
FaultDetectionUtils._load_model_object, logger.warning/logger.exception are
updated accordingly.
- Around line 1543-1569: The blanket merged_df = merged_df.fillna(0) should be
removed and replaced by targeted null-handling: do not coerce device_name to
0—backfill device_name from device_id (e.g., merged_df["device_name"] =
merged_df["device_name"].fillna(merged_df["device_id"]) or equivalent) and
explicitly fill only numeric fault columns defined by
FaultDetectionUtils.ML_FAULT_DEFAULTS and FaultDetectionUtils.RULE_FAULT_COLUMNS
with their default values (use those dict keys to locate columns and call
fillna(default) per column), then proceed with computing
anomaly_percentage_fault, anomaly_sequence_fault and casting fault_flag_columns
to int as before.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 1a3e2907-5be5-4792-88a1-47726b9a25e3
📒 Files selected for processing (3)
src/workflows/airqo_etl_utils/ml_utils.pysrc/workflows/airqo_etl_utils/tests/test_ml_utils.pysrc/workflows/dags/task_docs.py
✅ Files skipped from review due to trivial changes (2)
- src/workflows/airqo_etl_utils/tests/test_ml_utils.py
- src/workflows/dags/task_docs.py
Comment on lines
+1227
to
+1238
| def _load_existing_isolation_forest_metrics() -> Optional[Dict[str, float]]: | ||
| try: | ||
| artifact = FaultDetectionUtils._load_model_object( | ||
| source_file=FaultDetectionUtils._get_isolation_forest_model_path() | ||
| ) | ||
| if isinstance(artifact, dict) and isinstance(artifact.get("metrics"), dict): | ||
| return artifact["metrics"] | ||
| except FileNotFoundError: | ||
| return None | ||
| except Exception as exc: | ||
| logger.warning(f"Failed to load existing fault-detection metrics: {exc}") | ||
| return None |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's look at the file structure and find the relevant functions
cd src/workflows/airqo_etl_utils && wc -l ml_utils.py && echo "---" && head -50 ml_utils.py | cat -nRepository: airqo-platform/AirQo-api
Length of output: 2000
🏁 Script executed:
# Search for save_isolation_forest_model to understand how it uses the return value
rg -n "save_isolation_forest_model" src/workflows/airqo_etl_utils/ml_utils.py -A 30Repository: airqo-platform/AirQo-api
Length of output: 2911
🏁 Script executed:
# Search for _load_model_object to see what exceptions it might raise
rg -n "_load_model_object" src/workflows/airqo_etl_utils/ml_utils.py -B 5 -A 20Repository: airqo-platform/AirQo-api
Length of output: 3385
🏁 Script executed:
# Check the imports at the top of the file to see if google_api_exceptions is imported
head -100 src/workflows/airqo_etl_utils/ml_utils.py | cat -nRepository: airqo-platform/AirQo-api
Length of output: 4920
🏁 Script executed:
# Look at the complete _load_existing_isolation_forest_metrics function and surrounding context
sed -n '1220,1250p' src/workflows/airqo_etl_utils/ml_utils.py | cat -nRepository: airqo-platform/AirQo-api
Length of output: 1527
🏁 Script executed:
# Find the storage implementation
find . -name "storage.py" -o -name "storage" -type d | head -20Repository: airqo-platform/AirQo-api
Length of output: 108
🏁 Script executed:
# Search for load_file_object method
rg -n "def load_file_object" --type pyRepository: airqo-platform/AirQo-api
Length of output: 484
🏁 Script executed:
# Look for GCSFileStorage implementation
rg -n "class GCSFileStorage" -A 30 --type pyRepository: airqo-platform/AirQo-api
Length of output: 2906
🏁 Script executed:
# Get the full GCSFileStorage.load_file_object implementation
sed -n '183,250p' src/workflows/airqo_etl_utils/storage/cloud_storage.py | cat -nRepository: airqo-platform/AirQo-api
Length of output: 3297
🏁 Script executed:
# Also check FileStorage base class to see which one is used by FaultDetectionUtils
rg -n "class FileStorage" -A 5 src/workflows/airqo_etl_utils/storage/Repository: airqo-platform/AirQo-api
Length of output: 587
🏁 Script executed:
# Check which storage backend is configured for FaultDetectionUtils
rg -n "_get_model_storage" src/workflows/airqo_etl_utils/ml_utils.py -A 10Repository: airqo-platform/AirQo-api
Length of output: 1325
🏁 Script executed:
# Check if GCSFileStorage internally catches google_api_exceptions.NotFound before converting to FileNotFoundError
rg -n "google.cloud.exceptions\|NotFound\|blob.exists" src/workflows/airqo_etl_utils/storage/cloud_storage.py -B 2 -A 2Repository: airqo-platform/AirQo-api
Length of output: 50
🏁 Script executed:
# Check what exceptions can be raised by blob.download_as_bytes()
rg -n "download_as_bytes\|download_to_filename" src/workflows/airqo_etl_utils/storage/cloud_storage.py -B 2 -A 5Repository: airqo-platform/AirQo-api
Length of output: 50
🏁 Script executed:
# Verify the exact control flow in save_isolation_forest_model when old_metrics is None
sed -n '1308,1350p' src/workflows/airqo_etl_utils/ml_utils.py | cat -nRepository: airqo-platform/AirQo-api
Length of output: 2413
Don't collapse metric-read failures into "no previous model".
Returning None for every non-FileNotFoundError exception makes save_isolation_forest_model() treat transient storage/auth/deserialization failures as "no previous model" and deploy the candidate unconditionally. Only missing-artifact cases should return None; all other failures should abort to prevent silent deployments on recoverable errors.
Suggested fix
try:
artifact = FaultDetectionUtils._load_model_object(
source_file=FaultDetectionUtils._get_isolation_forest_model_path()
)
if isinstance(artifact, dict) and isinstance(artifact.get("metrics"), dict):
return artifact["metrics"]
except FileNotFoundError:
return None
except Exception as exc:
logger.warning(f"Failed to load existing fault-detection metrics: {exc}")
- return None
+ raise
+ return None🧰 Tools
🪛 Ruff (0.15.12)
[warning] 1236-1236: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/workflows/airqo_etl_utils/ml_utils.py` around lines 1227 - 1238, The
helper _load_existing_isolation_forest_metrics currently swallows all exceptions
and returns None which makes save_isolation_forest_model treat transient read
errors as "no previous model"; change it so only FileNotFoundError returns None,
but any other Exception should be logged and re-raised (or raise a custom
RuntimeError) so callers like save_isolation_forest_model abort instead of
deploying unconditionally; update the body around
FaultDetectionUtils._load_model_object and
FaultDetectionUtils._get_isolation_forest_model_path to log the failure with
context and then re-raise the exception instead of returning None.
wabinyai
changed the title
…MongoDB integration and explanations
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
Contributor
|
Predict service azure changes in this PR available for preview here |
Contributor
|
Predict service azure changes in this PR available for preview here |
Contributor
|
Predict service azure changes in this PR available for preview here |
…into faultdetection
Contributor
|
Predict service azure changes in this PR available for preview here |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces an end-to-end fault-detection workflow in the workflows service (BigQuery fetch + rule-based checks + Isolation Forest anomaly scoring + separate periodic training), and exposes the resulting “faulty devices” data via the predict API.
Changes:
- Added scheduled fault-detection DAG and a bi-monthly model-training DAG, plus supporting docs and configuration.
- Implemented BigQuery extraction + Isolation Forest training/loading/scoring utilities, with MongoDB persistence of consolidated fault results.
- Extended predict API to read faulty-device records (optionally with human-readable explanations) and documented the endpoint.
Reviewed changes
Copilot reviewed 17 out of 18 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| src/workflows/env.sample | Adds sample env vars for fault-detection lookbacks, model storage, and MLflow experiment. |
| src/workflows/dags/task_docs.py | Adds Airflow task documentation strings for fault-detection tasks. |
| src/workflows/dags/fault_detection_training_job.py | New bi-monthly DAG to train and persist the Isolation Forest model. |
| src/workflows/dags/fault_detection_job.py | Refactors scheduled fault detection into a single task and uses new utilities. |
| src/workflows/dags/dag_docs.py | Adds DAG-level documentation for detection and training workflows. |
| src/workflows/airqo_etl_utils/tests/test_ml_utils.py | Adds unit tests for fault detection feature prep, scoring, training, and persistence. |
| src/workflows/airqo_etl_utils/tests/test_bigquery_api.py | Adds unit tests for fault-detection BigQuery query rendering and aggregation behavior. |
| src/workflows/airqo_etl_utils/sql/faultdetection/2026041701fault_detection.sql | New SQL for fault-detection raw readings selection and daily aggregation. |
| src/workflows/airqo_etl_utils/sql/faultdetection/init.py | Initializes the faultdetection SQL package. |
| src/workflows/airqo_etl_utils/ml_utils.py | Adds FaultDetectionUtils with rule-based + Isolation Forest logic, and improves Mongo write behavior. |
| src/workflows/airqo_etl_utils/config.py | Adds fault-detection configuration fields to the workflows config. |
| src/workflows/airqo_etl_utils/bigquery_api.py | Adds BigQueryApi method to fetch fault-detection raw readings. |
| src/predict/README.md | Documents local setup and adds “faulty devices” API documentation and examples. |
| src/predict/api/prediction.py | Enhances faulty-devices endpoint with filtering and optional explanations; improves responses. |
| src/predict/api/helpers.py | Adds faulty-devices Mongo reader, Mongo document serialization, and explanation helpers. |
| src/predict/api/config.py | Adds config for faulty-devices collection and a connector for the fault-detection DB. |
| src/predict/api/.env.local | Adds a reference env file for local predict API development. |
| src/predict/.gitignore | Updates ignore rules (but currently introduces a .env ignore regression). |
Comments suppressed due to low confidence (1)
src/workflows/airqo_etl_utils/ml_utils.py:1017
- When
model_dfhas <2 rows,flag_pattern_based_faults()assignsanomaly_value/anomaly_scoreusing empty Series and returns. Ifmodel_dfhas 1 row, this results in NaNs for the anomaly columns (and inconsistent downstream typing). Prefer returning a frame with the same row count and explicit defaults (e.g., treat as non-anomalous with a neutral score) or return a fully empty frame with the expected columns.
model_df = FaultDetectionUtils._prepare_model_feature_frame(
df, feature_columns=feature_columns
)
if model_df.empty or len(model_df.index) < 2:
model_df["anomaly_value"] = pd.Series(dtype=int)
model_df["anomaly_score"] = pd.Series(dtype=float)
return model_df
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| # Environments | ||
| .env | ||
| .env |
Comment on lines
+1273
to
+1280
| results["timestamp"] = pd.to_datetime(results["timestamp"], utc=True) | ||
| num_cols = results.select_dtypes(include="number").columns | ||
| results = ( | ||
| results.groupby("device_id") | ||
| .resample("D", on="timestamp")[num_cols] | ||
| .mean() | ||
| ) | ||
| results.reset_index(inplace=True) |
Comment on lines
+691
to
+694
| CONSTANT_VALUE_WINDOW = 24 | ||
| MISSING_DATA_WINDOW = 60 | ||
| LOW_BATTERY_THRESHOLD = 3.3 | ||
| LOW_BATTERY_WINDOW = 24 |
| This DAG does not train the ML model. It expects the `AirQo-fault-detection-model-training` DAG to have saved the Isolation Forest artifact first. | ||
|
|
||
| #### Workflow Steps | ||
| 1. **Raw Data Extraction** (`fetch_raw_data`) |
Contributor
|
Hi @wabinyai please resolve the merge conflicts in this PR. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
🚀 Pull Request
📋 Description
What does this PR do?
Why is this change needed?
🔗 Related Issues
🔄 Type of Change
🏗️ Affected Services
Microservices changed:
🧪 Testing
Test summary:
💥 Breaking Changes
📝 Additional Notes
✅ Checklist
Summary by CodeRabbit
New Features
Documentation
Tests