| | |
| | """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 |
| |
|
