def get_baseline_drift()

in batch_pipeline/pipelines/postprocess_monitor_script.py [0:0]

• 12 lines of code
• 5 McCabe index (conditional complexity)

def get_baseline_drift(feature):
    if "violations" in feature:
        for violation in feature["violations"]:
            if violation["constraint_check_type"] == "baseline_drift_check":
                desc = violation["description"]
                matches = re.search("distance: (.+) exceeds threshold: (.+)", desc)
                if matches:
                    yield {
                        "metric_name": f'feature_baseline_drift_{violation["feature_name"]}',
                        "metric_value": float(matches.group(1)),
                        "metric_threshold": float(matches.group(2)),
                    }