| |
| """Schema Detector Plugin""" |
| import pandas as pd |
| from typing import Dict, Any |
|
|
| class SchemaDetector: |
| """Detects and reports data schema.""" |
| def get_schema(self, df: pd.DataFrame) -> Dict[str, Any]: |
| schema = {} |
| for col in df.columns: |
| dtype = str(df[col].dtype) |
| if pd.api.types.is_numeric_dtype(df[col]): |
| base_type = "Numeric" |
| elif pd.api.types.is_datetime64_any_dtype(df[col]): |
| base_type = "Datetime" |
| elif df[col].nunique() < min(10, len(df) / 5): |
| base_type = "Categorical" |
| else: |
| base_type = "Text/Object" |
| schema[col] = { |
| "inferred_type": base_type, |
| "pandas_dtype": dtype, |
| "non_null_count": int(df[col].count()), |
| "unique_values": int(df[col].nunique()) |
| } |
| return schema |
|
|
