def get_baseline_drift()

in custom_resource/sagemaker_query_drift.py [0:0]


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"]
                print(desc)
                matches = re.search("distance: (.+) exceeds threshold: (.+)", desc)
                if matches:
                    match = matches.group(1)
                    threshold = matches.group(2)
                    yield {
                        "feature": violation["feature_name"],
                        "drift": float(match),
                        "threshold": float(threshold),
                    }