Skip to content

API Reference

OMR - Omni Data Refinement v0.3.0 The standard framework for dataset quality, validation, profiling, monitoring, and reliability.

Dataset

The primary OMR interface. Represents a dataset with built-in quality, validation, and profiling intelligence.

Source code in omr/core/dataset.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class Dataset:
    """
    The primary OMR interface.
    Represents a dataset with built-in quality, validation, and profiling intelligence.
    """

    def __init__(self, data: Any):
        """
        Initializes the Dataset.
        Supports Pandas and Polars DataFrames.
        """
        self._df = ensure_pandas(data).copy()
        self.metadata = DatasetMetadata.from_dataframe(self._df)

        # Engines
        self._health_engine       = HealthEngine()
        self._cleaning_engine     = CleaningEngine()
        self._profiling_engine    = ProfilingEngine()
        self._validation_engine   = ValidationEngine()
        self._statistics_engine   = StatisticsEngine()
        self._drift_engine        = DriftEngine()
        self._explain_engine      = ExplainabilityEngine()
        self._version_system      = VersioningSystem()
        self._report_generator    = ReportGenerator()

        # State
        self._last_health = None
        self._last_profile = None

        # Save initial version
        self._version_system.snapshot(self._df, self.metadata, name="v0 — Initial")

    # ── 1. Core Health & Profiling ───────────────────────────────────────────

    def summary(self) -> "Dataset":
        """Prints a one-line health summary."""
        # Fast compute just the score without deep inspection
        health = self._health_engine.run(self._df)
        display_summary(self.metadata, health.score)
        return self

    def health(self) -> HealthReport:
        """Runs the 5-pillar health check and returns a HealthReport."""
        report = self._health_engine.run(self._df)
        self._last_health = report
        display_health(report)
        return report

    def profile(self) -> "Dataset":
        """Generates a full statistical profile."""
        report = self._profiling_engine.run(self._df)
        self._last_profile = report
        display_profile(report)
        return self

    # ── 2. Cleaning ──────────────────────────────────────────────────────────

    def clean(self) -> "Dataset":
        """Auto-resolves all issues detected by health()."""
        if self._last_health is None:
            self._last_health = self._health_engine.run(self._df)

        self._df = self._cleaning_engine.run(self._df, self._last_health, self.metadata)

        # Update metadata shape after cleaning (e.g. dropped rows)
        self.metadata.num_rows = len(self._df)
        self.metadata.num_cols = len(self._df.columns)
        self.metadata.missing_cells = int(self._df.isnull().sum().sum())

        self._version_system.snapshot(self._df, self.metadata, name="After clean()")

        # Clear health state so next call re-evaluates
        self._last_health = None
        return self

    def explain_changes(self) -> "Dataset":
        """Displays the transformation history."""
        display_changelog(self.metadata)
        return self

    # ── 3. Validation & Rules ────────────────────────────────────────────────

    def validate(self, schema: Dict[str, Any]) -> "Dataset":
        """Validates the dataset against a schema of ConstraintTypes."""
        report = self._validation_engine.run(self._df, schema)
        display_validation(report)
        return self

    # ── 4. Advanced Analytics ────────────────────────────────────────────────

    def analyze(self) -> "Dataset":
        """Runs deep statistical analysis (outliers, skew, multicollinearity)."""
        report = self._statistics_engine.run(self._df)
        display_statistics(report)
        return self

    def compare(self, other: "Dataset") -> "Dataset":
        """Detects distribution drift against another dataset (e.g. production)."""
        report = self._drift_engine.run(self._df, other._df)
        display_drift(report)
        return self

    def explain(self, issue: str) -> dict:
        """Explains a data quality issue conceptually and offers fixes."""
        exp = self._explain_engine.run(issue)
        from ..utils.formatting import console
        from rich.panel import Panel
        from rich.text import Text
        t = Text()
        t.append("Definition: ", style="bold cyan"); t.append(f"{exp['definition']}\n\n")
        t.append("Why it matters: ", style="bold yellow"); t.append(f"{exp['why_it_matters']}\n\n")
        t.append("Risks: ", style="bold red"); t.append(f"{exp['risks']}\n\n")
        t.append("Recommended fixes:\n", style="bold green"); t.append(exp['recommended_fixes'])
        console.print(Panel(t, title=f"Explainability — {issue.title()}", border_style="cyan"))
        return exp

    # ── 5. Ops & Versioning ──────────────────────────────────────────────────

    def pipeline(self) -> Pipeline:
        """Returns a fluent Pipeline interface."""
        return Pipeline(self)

    def report(self, format: str = "html", path: str = "omr_report") -> str:
        """Generates a physical report file."""
        if not self._last_health:
            self._last_health = self._health_engine.run(self._df)
        return self._report_generator.generate(self, format=format, path=path)

    def snapshot(self, name: str = "", description: str = "") -> int:
        """Saves a version snapshot."""
        return self._version_system.snapshot(self._df, self.metadata, name, description)

    def rollback(self, version_id: int) -> "Dataset":
        """Restores a previous snapshot."""
        self._df, self.metadata = self._version_system.rollback(version_id)
        self._last_health = None
        self._last_profile = None
        return self

    def export(self) -> pd.DataFrame:
        """Returns the underlying pandas DataFrame."""
        return self._df.copy()

__init__(data)

Initializes the Dataset. Supports Pandas and Polars DataFrames.

Source code in omr/core/dataset.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(self, data: Any):
    """
    Initializes the Dataset.
    Supports Pandas and Polars DataFrames.
    """
    self._df = ensure_pandas(data).copy()
    self.metadata = DatasetMetadata.from_dataframe(self._df)

    # Engines
    self._health_engine       = HealthEngine()
    self._cleaning_engine     = CleaningEngine()
    self._profiling_engine    = ProfilingEngine()
    self._validation_engine   = ValidationEngine()
    self._statistics_engine   = StatisticsEngine()
    self._drift_engine        = DriftEngine()
    self._explain_engine      = ExplainabilityEngine()
    self._version_system      = VersioningSystem()
    self._report_generator    = ReportGenerator()

    # State
    self._last_health = None
    self._last_profile = None

    # Save initial version
    self._version_system.snapshot(self._df, self.metadata, name="v0 — Initial")

analyze()

Runs deep statistical analysis (outliers, skew, multicollinearity).

Source code in omr/core/dataset.py 
116
117
118
119
120
def analyze(self) -> "Dataset":
    """Runs deep statistical analysis (outliers, skew, multicollinearity)."""
    report = self._statistics_engine.run(self._df)
    display_statistics(report)
    return self

clean()

Auto-resolves all issues detected by health().

Source code in omr/core/dataset.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def clean(self) -> "Dataset":
    """Auto-resolves all issues detected by health()."""
    if self._last_health is None:
        self._last_health = self._health_engine.run(self._df)

    self._df = self._cleaning_engine.run(self._df, self._last_health, self.metadata)

    # Update metadata shape after cleaning (e.g. dropped rows)
    self.metadata.num_rows = len(self._df)
    self.metadata.num_cols = len(self._df.columns)
    self.metadata.missing_cells = int(self._df.isnull().sum().sum())

    self._version_system.snapshot(self._df, self.metadata, name="After clean()")

    # Clear health state so next call re-evaluates
    self._last_health = None
    return self

compare(other)

Detects distribution drift against another dataset (e.g. production).

Source code in omr/core/dataset.py 
122
123
124
125
126
def compare(self, other: "Dataset") -> "Dataset":
    """Detects distribution drift against another dataset (e.g. production)."""
    report = self._drift_engine.run(self._df, other._df)
    display_drift(report)
    return self

explain(issue)

Explains a data quality issue conceptually and offers fixes.

Source code in omr/core/dataset.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
def explain(self, issue: str) -> dict:
    """Explains a data quality issue conceptually and offers fixes."""
    exp = self._explain_engine.run(issue)
    from ..utils.formatting import console
    from rich.panel import Panel
    from rich.text import Text
    t = Text()
    t.append("Definition: ", style="bold cyan"); t.append(f"{exp['definition']}\n\n")
    t.append("Why it matters: ", style="bold yellow"); t.append(f"{exp['why_it_matters']}\n\n")
    t.append("Risks: ", style="bold red"); t.append(f"{exp['risks']}\n\n")
    t.append("Recommended fixes:\n", style="bold green"); t.append(exp['recommended_fixes'])
    console.print(Panel(t, title=f"Explainability — {issue.title()}", border_style="cyan"))
    return exp

explain_changes()

Displays the transformation history.

Source code in omr/core/dataset.py 
101
102
103
104
def explain_changes(self) -> "Dataset":
    """Displays the transformation history."""
    display_changelog(self.metadata)
    return self

export()

Returns the underlying pandas DataFrame.

Source code in omr/core/dataset.py 
165
166
167
def export(self) -> pd.DataFrame:
    """Returns the underlying pandas DataFrame."""
    return self._df.copy()

health()

Runs the 5-pillar health check and returns a HealthReport.

Source code in omr/core/dataset.py 
67
68
69
70
71
72
def health(self) -> HealthReport:
    """Runs the 5-pillar health check and returns a HealthReport."""
    report = self._health_engine.run(self._df)
    self._last_health = report
    display_health(report)
    return report

pipeline()

Returns a fluent Pipeline interface.

Source code in omr/core/dataset.py 
144
145
146
def pipeline(self) -> Pipeline:
    """Returns a fluent Pipeline interface."""
    return Pipeline(self)

profile()

Generates a full statistical profile.

Source code in omr/core/dataset.py 
74
75
76
77
78
79
def profile(self) -> "Dataset":
    """Generates a full statistical profile."""
    report = self._profiling_engine.run(self._df)
    self._last_profile = report
    display_profile(report)
    return self

report(format='html', path='omr_report')

Generates a physical report file.

Source code in omr/core/dataset.py 
148
149
150
151
152
def report(self, format: str = "html", path: str = "omr_report") -> str:
    """Generates a physical report file."""
    if not self._last_health:
        self._last_health = self._health_engine.run(self._df)
    return self._report_generator.generate(self, format=format, path=path)

rollback(version_id)

Restores a previous snapshot.

Source code in omr/core/dataset.py 
158
159
160
161
162
163
def rollback(self, version_id: int) -> "Dataset":
    """Restores a previous snapshot."""
    self._df, self.metadata = self._version_system.rollback(version_id)
    self._last_health = None
    self._last_profile = None
    return self

snapshot(name='', description='')

Saves a version snapshot.

Source code in omr/core/dataset.py 
154
155
156
def snapshot(self, name: str = "", description: str = "") -> int:
    """Saves a version snapshot."""
    return self._version_system.snapshot(self._df, self.metadata, name, description)

summary()

Prints a one-line health summary.

Source code in omr/core/dataset.py 
60
61
62
63
64
65
def summary(self) -> "Dataset":
    """Prints a one-line health summary."""
    # Fast compute just the score without deep inspection
    health = self._health_engine.run(self._df)
    display_summary(self.metadata, health.score)
    return self

validate(schema)

Validates the dataset against a schema of ConstraintTypes.

Source code in omr/core/dataset.py 
108
109
110
111
112
def validate(self, schema: Dict[str, Any]) -> "Dataset":
    """Validates the dataset against a schema of ConstraintTypes."""
    report = self._validation_engine.run(self._df, schema)
    display_validation(report)
    return self

DatasetMetadata dataclass

Captures essential metadata about a dataset at a point in time. Automatically updated by Dataset after every transformation.

Source code in omr/core/metadata.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@dataclass
class DatasetMetadata:
    """
    Captures essential metadata about a dataset at a point in time.
    Automatically updated by Dataset after every transformation.
    """
    num_rows: int
    num_cols: int
    column_names: List[str]
    dtypes: Dict[str, str]
    missing_cells: int
    missing_pct: float
    duplicate_rows: int
    memory_mb: float
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    source: Optional[str] = None
    transformations: List[TransformationRecord] = field(default_factory=list)

    @classmethod
    def from_dataframe(cls, df: pd.DataFrame, source: Optional[str] = None) -> "DatasetMetadata":
        total_cells = df.size
        missing = int(df.isnull().sum().sum())
        return cls(
            num_rows=len(df),
            num_cols=len(df.columns),
            column_names=df.columns.tolist(),
            dtypes={col: str(df[col].dtype) for col in df.columns},
            missing_cells=missing,
            missing_pct=round((missing / total_cells * 100) if total_cells > 0 else 0.0, 2),
            duplicate_rows=int(df.duplicated().sum()),
            memory_mb=round(df.memory_usage(deep=True).sum() / (1024 ** 2), 4),
            source=source,
        )

    def log_transformation(self, operation: str, column: str,
                           rows_affected: int, reason: str) -> None:
        step = len(self.transformations) + 1
        self.transformations.append(TransformationRecord(
            step=step, operation=operation, column=column,
            rows_affected=rows_affected, reason=reason
        ))

    def to_dict(self) -> dict:
        return {
            "num_rows": self.num_rows,
            "num_cols": self.num_cols,
            "missing_cells": self.missing_cells,
            "missing_pct": self.missing_pct,
            "duplicate_rows": self.duplicate_rows,
            "memory_mb": self.memory_mb,
            "created_at": self.created_at,
            "source": self.source,
        }

Monitor

Watches a dataset over time for changes in quality or distribution.

Source code in omr/monitoring/monitor.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class Monitor:
    """Watches a dataset over time for changes in quality or distribution."""

    def __init__(self):
        self.reference_dataset = None
        self.schema = None
        self.alerts: List[Alert] = []

    def watch(self, dataset) -> None:
        """Sets the baseline dataset for future comparisons."""
        self.reference_dataset = dataset
        self.alerts = []

    def set_schema(self, schema: dict) -> None:
        """Sets a schema for validation checks."""
        self.schema = schema

    def check(self, df: pd.DataFrame) -> List[Alert]:
        """Runs checks against the baseline and returns any alerts."""
        self.alerts = []
        if self.reference_dataset is None:
            return self.alerts

        # 1. Row count anomaly
        ref_rows = self.reference_dataset.metadata.num_rows
        curr_rows = len(df)
        if ref_rows > 0:
            diff = abs(curr_rows - ref_rows) / ref_rows
            if diff > 0.5:
                self.alerts.append(Alert(
                    "Volume Drop/Spike", "High",
                    f"Row count changed by {diff*100:.1f}% ({ref_rows} -> {curr_rows})"
                ))

        # 2. Schema check (if configured)
        if self.schema:
            from ..core.dataset import Dataset
            ds = Dataset(df)
            report = ds._validation_engine.run(df, self.schema)
            for issue in report.issues:
                self.alerts.append(Alert(
                    "Schema Violation", "High",
                    f"{issue.column}: {issue.description}"
                ))

        # 3. Quick drift check on numeric columns
        import numpy as np
        num_cols = df.select_dtypes(include=[np.number]).columns
        ref_df = self.reference_dataset._df
        for col in num_cols:
            if col in ref_df.columns:
                ref_mean = ref_df[col].mean()
                curr_mean = df[col].mean()
                if ref_mean != 0:
                    drift = abs(curr_mean - ref_mean) / abs(ref_mean)
                    if drift > 0.2:
                        self.alerts.append(Alert(
                            "Mean Shift", "Medium",
                            f"Column '{col}' mean shifted by {drift*100:.1f}%"
                        ))

        return self.alerts

check(df)

Runs checks against the baseline and returns any alerts.

Source code in omr/monitoring/monitor.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def check(self, df: pd.DataFrame) -> List[Alert]:
    """Runs checks against the baseline and returns any alerts."""
    self.alerts = []
    if self.reference_dataset is None:
        return self.alerts

    # 1. Row count anomaly
    ref_rows = self.reference_dataset.metadata.num_rows
    curr_rows = len(df)
    if ref_rows > 0:
        diff = abs(curr_rows - ref_rows) / ref_rows
        if diff > 0.5:
            self.alerts.append(Alert(
                "Volume Drop/Spike", "High",
                f"Row count changed by {diff*100:.1f}% ({ref_rows} -> {curr_rows})"
            ))

    # 2. Schema check (if configured)
    if self.schema:
        from ..core.dataset import Dataset
        ds = Dataset(df)
        report = ds._validation_engine.run(df, self.schema)
        for issue in report.issues:
            self.alerts.append(Alert(
                "Schema Violation", "High",
                f"{issue.column}: {issue.description}"
            ))

    # 3. Quick drift check on numeric columns
    import numpy as np
    num_cols = df.select_dtypes(include=[np.number]).columns
    ref_df = self.reference_dataset._df
    for col in num_cols:
        if col in ref_df.columns:
            ref_mean = ref_df[col].mean()
            curr_mean = df[col].mean()
            if ref_mean != 0:
                drift = abs(curr_mean - ref_mean) / abs(ref_mean)
                if drift > 0.2:
                    self.alerts.append(Alert(
                        "Mean Shift", "Medium",
                        f"Column '{col}' mean shifted by {drift*100:.1f}%"
                    ))

    return self.alerts

set_schema(schema)

Sets a schema for validation checks.

Source code in omr/monitoring/monitor.py 
29
30
31
def set_schema(self, schema: dict) -> None:
    """Sets a schema for validation checks."""
    self.schema = schema

watch(dataset)

Sets the baseline dataset for future comparisons.

Source code in omr/monitoring/monitor.py 
24
25
26
27
def watch(self, dataset) -> None:
    """Sets the baseline dataset for future comparisons."""
    self.reference_dataset = dataset
    self.alerts = []

Pipeline

Enables fluent chaining of Dataset operations.

Source code in omr/pipelines/pipeline.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class Pipeline:
    """Enables fluent chaining of Dataset operations."""

    def __init__(self, dataset):
        self._dataset = dataset

    def health(self):
        self._dataset.health()
        return self

    def clean(self):
        self._dataset.clean()
        return self

    def analyze(self):
        self._dataset.analyze()
        return self

    def validate(self, schema):
        self._dataset.validate(schema)
        return self

    def export(self):
        return self._dataset.export()

get_plugin(name)

Instantiates and returns a registered plugin by name.

Source code in omr/plugins/registry.py 
24
25
26
27
28
def get_plugin(name: str) -> Any:
    """Instantiates and returns a registered plugin by name."""
    if name not in _PLUGINS:
        raise ValueError(f"Plugin '{name}' not found. Did you pip install it?")
    return _PLUGINS[name]()

register_plugin(name)

Decorator to register a third-party plugin class.

Example

@register_plugin("medical") class MedicalPlugin: def validate(self, dataset): ...

Source code in omr/plugins/registry.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
def register_plugin(name: str):
    """
    Decorator to register a third-party plugin class.

    Example:
        @register_plugin("medical")
        class MedicalPlugin:
            def validate(self, dataset): ...
    """
    def wrapper(cls):
        _PLUGINS[name] = cls
        return cls
    return wrapper