pyodideWorker.ts (WebWorker Bridge)
dataset_manager.py (DataFrame Cache)
profile_service.py (Dispatch Engine)
quality.py (Health Score Metric)
LOCATION: src/worker/pyodideWorker.ts
ROLE: WebWorker thread bootstrap & Pyodide MEMFS injection
import type { PyodideInterface } from "pyodide" ;
import datasetManagerCode from "../../python/engine/dataset_manager.py?raw" ;
import profileServiceCode from "../../python/engine/profile_service.py?raw" ;
import statsCode from "../../python/profilers/statistics.py?raw" ;
import qualityCode from "../../python/profilers/quality.py?raw" ;
import operationsCode from "../../python/operations/core.py?raw" ;
let pyodide: PyodideInterface | null = null ;
async function initPyodide () {
try {
const pyodideModule = await import ("https://cdn.jsdelivr.net/pyodide/v314.0.2/full/pyodide.mjs" );
pyodide = await pyodideModule.loadPyodide ({
indexURL: "https://cdn.jsdelivr.net/pyodide/v314.0.2/full/" ,
});
await pyodide.loadPackage (["pandas" , "numpy" , "pyarrow" , "micropip" ]);
pyodide.FS.mkdir ("/engine" );
pyodide.FS.mkdir ("/profilers" );
pyodide.FS.mkdir ("/operations" );
pyodide.FS.writeFile ("/engine/__init__.py" , "" );
pyodide.FS.writeFile ("/engine/dataset_manager.py" , datasetManagerCode);
pyodide.FS.writeFile ("/engine/profile_service.py" , profileServiceCode);
pyodide.FS.writeFile ("/profilers/quality.py" , qualityCode);
pyodide.FS.writeFile ("/profilers/statistics.py" , statsCode);
pyodide.FS.writeFile ("/operations/core.py" , operationsCode);
await pyodide.runPythonAsync (`
import sys, json, math
import pandas as pd
sys.path.append('/' )
from engine.dataset_manager import manager
from engine.profile_service import profiler
from operations.core import operator
`);
self.postMessage ({ type: "READY" , success: true });
} catch (error) {
self.postMessage ({ type: "ERROR" , error: String (error) });
}
}
LOCATION: python/engine/dataset_manager.py
ROLE: High-performance DataFrame cache & type-safe ingestion
import pandas as pd
import numpy as np
import io
class DatasetManager :
def __init__ (self):
self._active_df: pd.DataFrame | None = None
self._history = []
self._history_pointer = -1
self._schema_cache = {}
def load_bytes (self, file_bytes: bytes , filename: str ) -> dict :
if filename.endswith (".csv" ):
self._active_df = pd.read_csv (io.BytesIO (file_bytes), low_memory=False )
elif filename.endswith (".parquet" ):
self._active_df = pd.read_parquet (io.BytesIO (file_bytes))
else :
raise ValueError (f"Unsupported extension for {filename}" )
self._history = [self._active_df.copy ()]
self._history_pointer = 0
return self.get_summary ()
def get_summary (self) -> dict :
if self._active_df is None :
return {"rows" : 0 , "cols" : 0 , "columns" : []}
return {
"rows" : len (self._active_df),
"cols" : len (self._active_df.columns),
"columns" : list (self._active_df.columns),
"memory_usage_mb" : round (self._active_df.memory_usage (deep=True ).sum () / (1024 * 1024 ), 2 )
}
manager = DatasetManager ()
LOCATION: python/engine/profile_service.py
ROLE: Multi-profiler orchestrator & JSON serialization
import pandas as pd
from profilers.statistics import compute_statistics
from profilers.missing import compute_missing
from profilers.duplicates import compute_duplicates
from profilers.datatypes import infer_datatypes
from profilers.quality import compute_quality
class ProfileService :
def profile (self, df: pd.DataFrame ) -> dict :
missing = compute_missing (df)
dups = compute_duplicates (df)
dtypes = infer_datatypes (df)
stats = compute_statistics (df)
quality = compute_quality (df, missing, dups, dtypes, stats)
return {
"shape" : {"rows" : len (df), "cols" : len (df.columns)},
"missing" : missing,
"duplicates" : dups,
"datatypes" : dtypes,
"statistics" : stats,
"quality" : quality
}
profiler = ProfileService ()
LOCATION: python/profilers/quality.py
ROLE: Health score calculation & anomaly detection
def compute_quality (df, missing_stats, dup_stats, dtypes, stats):
total_cells = df.size
total_rows = len (df)
if total_rows == 0 or total_cells == 0 :
return {"score" : 0 , "components" : {}, "issues" : ["Empty dataset" ]}
missing_cells = missing_stats.get ("_overall" , {}).get ("total_missing" , 0 )
duplicate_rows = dup_stats.get ("_overall" , {}).get ("duplicate_rows" , 0 )
completeness = 100 - ((missing_cells / total_cells) * 100 ) if total_cells > 0 else 0
uniqueness = 100 - ((duplicate_rows / total_rows) * 100 ) if total_rows > 0 else 0
mixed_cols = sum (1 for _, info in dtypes.items () if info.get ("inferred" ) == "mixed" )
validity = max (0 , 100 - (mixed_cols / len (df.columns) * 50 )) if len (df.columns) > 0 else 100
composite = round (completeness * 0.40 + uniqueness * 0.25 + validity * 0.20 + 15.0 , 1 )
return {
"score" : min (100.0 , composite),
"components" : {
"completeness" : round (completeness, 1 ),
"uniqueness" : round (uniqueness, 1 ),
"validity" : round (validity, 1 )
}
}