"""FastAPI application for the CleanOps data-cleaning environment.""" from __future__ import annotations import copy import random from openenv.core import create_app from fastapi.responses import HTMLResponse, JSONResponse from cleanops_env.environment import CleanOpsEnvironment from cleanops_env.models import DataCleaningAction, DataCleaningObservation from cleanops_env.tasks import first_table_name, get_task_spec, sorted_rows app = create_app( CleanOpsEnvironment, DataCleaningAction, DataCleaningObservation, env_name="cleanops_env", max_concurrent_envs=4, ) @app.get("/demo/compare", include_in_schema=False) def demo_compare(task_id: str = "customer_contacts_easy", table_name: str | None = None, seed: int | None = None) -> JSONResponse: task_spec = get_task_spec(task_id) selected_table = table_name if table_name in task_spec.dirty_tables else first_table_name(task_spec) primary_key = task_spec.primary_keys[selected_table] before_rows = _seed_preview_rows(task_spec.dirty_tables[selected_table], primary_key, selected_table, seed) after_rows = _seed_preview_rows(task_spec.gold_tables[selected_table], primary_key, selected_table, seed) columns = sorted({column_name for row in before_rows + after_rows for column_name in row}) return JSONResponse( { "task_id": task_spec.task_id, "task_title": task_spec.title, "table_name": selected_table, "requested_seed": seed, "available_tables": list(task_spec.dirty_tables.keys()), "columns": columns, "before_rows": before_rows[:4], "after_rows": after_rows[:4], "before_row_count": len(before_rows), "after_row_count": len(after_rows), "solution_operation_ids": list(task_spec.solution_operation_ids), } ) def _seed_preview_rows( rows: list[dict[str, str]], primary_key: str, table_name: str, seed: int | None, ) -> list[dict[str, str]]: ordered_rows = sorted_rows(rows, primary_key) if seed is None or len(ordered_rows) <= 1: return ordered_rows shuffled_rows = copy.deepcopy(ordered_rows) random.Random(max(0, int(seed)) + sum(ord(char) for char in table_name)).shuffle(shuffled_rows) return shuffled_rows @app.get("/", include_in_schema=False) def root() -> HTMLResponse: return HTMLResponse( """ CleanOps OpenEnv

OpenEnv Benchmark

Real-world Data Cleaning

Deterministic Graders

See real data cleaning tasks working live.

CleanOps simulates the kind of operational cleanup analysts actually do before data reaches a CRM, warehouse, or billing system. The UI below runs the same hosted benchmark API used by the evaluator.

Choose task

Seed

Ready to run a live benchmark task.

Changing the seed changes the visible preview ordering and compare view. It does not change the task score itself.

Fixed tasks, typed actions, shaped rewards, and reproducible graders.

At a glance

Task ladder Easy → Hard

Core API /reset /step /state

Domain CRM + Orders + Billing

Reward signal Dense + partial progress

This homepage is a thin demo over the live environment. It doesn’t fake results: every task button calls the deployed API.

Live Task Snapshot

The cards and table below are populated from a real POST /reset response. Use the task buttons above to switch between benchmark scenarios, or choose your own task and seed.

Task - -

Seed Used -

Initial Score -

Validation Issues -

Focus Table Rows -

Objective

Loading...

Validation Issues

Available Operations

Before / After Cleaning

Loading compare view...

Dirty input -

Expected clean output -

Focus Table Preview

Raw Demo Payload

Loading live task data...

API & Submission Notes

The evaluator checks these endpoints directly. This page exists to make the environment easier to inspect visually.

GET /health
Service liveness check

Open

GET /schema
Typed OpenEnv schema

Open

GET /docs
Interactive FastAPI docs

Open

POST /reset
Start a task episode

live

POST /step
Apply a typed action

live

GET /state
Inspect current environment state

live

Sample curl

curl -X POST /reset -H "Content-Type: application/json" -d '{"task_id":"customer_contacts_easy","seed":7}'

""" ) def main(host: str = "0.0.0.0", port: int = 8000) -> None: import uvicorn uvicorn.run(app, host=host, port=port) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=8000) args = parser.parse_args() if args.host == "0.0.0.0" and args.port == 8000: main() else: main(host=args.host, port=args.port)