129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
"""
|
||
Evaluate a learned RIPPER ruleset against a dataset.
|
||
|
||
Ruleset format: {class_label: [[cond1, cond2, ...], [cond3, ...], ...], ...}
|
||
Conditions: 'col_name' → col == True
|
||
'!col_name' → col == False (negated condition)
|
||
|
||
The last entry in sorted_values is the default class — predicted when no rule fires.
|
||
"""
|
||
|
||
import pandas as pd
|
||
|
||
# Sentinel returned by runtime ontology guidance when no ontologically-valid action exists
|
||
# for the row's asset. It is never a true label, so it always counts as WRONG in accuracy/F1,
|
||
# and it is a plain string (not None) so sklearn's f1_score never chokes on mixed None/str
|
||
# labels. Tracked separately via the abstain_rate metric in run_all_experiments.
|
||
ABSTAIN = "Geen advies mogelijk."
|
||
|
||
|
||
def _row_matches_condition(row: pd.Series, cond: str) -> bool:
|
||
"""Return True if the row satisfies one condition."""
|
||
if cond.startswith('!'):
|
||
return row[cond[1:]] == False
|
||
return row[cond] == True
|
||
|
||
|
||
def _row_matches_rule(row: pd.Series, rule: list) -> bool:
|
||
"""Return True if the row satisfies ALL conditions in a rule (AND logic)."""
|
||
return all(_row_matches_condition(row, cond) for cond in rule)
|
||
|
||
|
||
def _predict_one(row: pd.Series, rule_set: dict, sorted_values: list) -> str:
|
||
"""
|
||
Predict the class for a single row.
|
||
|
||
Checks each class in sorted_values order (rarest first, same as training).
|
||
The first class whose rules fire wins. If nothing fires, return the default
|
||
class (sorted_values[-1], the most common one).
|
||
"""
|
||
for class_label in sorted_values[:-1]:
|
||
for rule in rule_set.get(class_label, []):
|
||
if rule and _row_matches_rule(row, rule):
|
||
return class_label
|
||
return sorted_values[-1] # default class: no rule fired
|
||
|
||
from ontology_guidance.get_onto_guidance_actions_per_asset import get_ontology_info_actions_per_asset
|
||
|
||
|
||
def _row_asset(row: pd.Series) -> str | None:
|
||
"""Return the asset value of a one-hot row: the assetType_* column that is True.
|
||
|
||
Mirrors the column convention used in build_invalid_row_mask (assetType_<value>).
|
||
Returns None if no assetType column is set.
|
||
"""
|
||
for col in row.index:
|
||
if col.startswith('assetType_') and row[col] == True:
|
||
return col.replace('assetType_', '')
|
||
return None
|
||
|
||
|
||
def _predict_one_runtime_onto_guidance(row: pd.Series, rule_set: dict, sorted_values: list) -> str:
|
||
"""
|
||
Predict the class for a single row, with inference-time ontology guidance.
|
||
|
||
Checks each class in sorted_values order (rarest first, same as training).
|
||
The first class whose rules fire wins. Then it checks whether that action is
|
||
ontologically valid for the row's asset. If not, it falls back to the most
|
||
common action that IS valid for the asset.
|
||
"""
|
||
asset = _row_asset(row)
|
||
valid_actions = get_ontology_info_actions_per_asset(asset) if asset is not None else []
|
||
valid_norm = {a.strip().lower() for a in valid_actions}
|
||
|
||
def _most_common_valid() -> str:
|
||
# walk most-common-first; return the first action valid for this asset
|
||
for item in reversed(sorted_values):
|
||
if item.strip().lower() in valid_norm:
|
||
return item
|
||
return ABSTAIN
|
||
|
||
for class_label in sorted_values[:-1]:
|
||
for rule in rule_set.get(class_label, []):
|
||
if rule and _row_matches_rule(row, rule):
|
||
if valid_norm and class_label.strip().lower() not in valid_norm:
|
||
return _most_common_valid()
|
||
return class_label
|
||
|
||
# no rule fired -> default class, but prefer one valid for the asset
|
||
if valid_norm:
|
||
return _most_common_valid()
|
||
return ABSTAIN
|
||
|
||
|
||
def evaluate_ruleset(
|
||
rule_set: dict,
|
||
data: pd.DataFrame,
|
||
sorted_values: list,
|
||
runtime_ontology_guidance_bool: bool
|
||
) -> tuple[pd.DataFrame, float]:
|
||
"""
|
||
Evaluate a ruleset against labelled data.
|
||
|
||
Args:
|
||
rule_set: {class_label: [[cond, ...], ...], ...}
|
||
data: DataFrame with feature columns plus a 'y' column (true labels)
|
||
sorted_values: class labels sorted by frequency ascending;
|
||
sorted_values[-1] is the default (most common) class
|
||
runtime_ontology_guidance_bool: if true will evaluate using inference time ontology guidance.
|
||
|
||
Returns:
|
||
result_df: copy of data with an extra 'y_pred' column
|
||
accuracy: fraction of rows where y_pred == y (0.0 – 1.0)
|
||
"""
|
||
result_df = data.copy()
|
||
|
||
if not runtime_ontology_guidance_bool:
|
||
result_df['y_pred'] = result_df.apply(
|
||
lambda row: _predict_one(row, rule_set, sorted_values),
|
||
axis=1,
|
||
)
|
||
else:
|
||
result_df['y_pred'] = result_df.apply(
|
||
lambda row: _predict_one_runtime_onto_guidance(row, rule_set, sorted_values),
|
||
axis=1,
|
||
)
|
||
|
||
accuracy = (result_df['y_pred'] == result_df['y']).mean()
|
||
|
||
return result_df, accuracy
|