Spaces:
Running
Running
| from __future__ import annotations | |
| import math | |
| from pathlib import Path | |
| from typing import Any | |
| import numpy as np | |
| import pandas as pd | |
| COORD_COLUMN_NAMES = { | |
| "lat", | |
| "latitude", | |
| "siat_latitude", | |
| "lon", | |
| "long", | |
| "longitude", | |
| "siat_longitude", | |
| } | |
| def _is_coordinate_column(col_name: Any) -> bool: | |
| return str(col_name).strip().lower() in COORD_COLUMN_NAMES | |
| def sanitize_value(value: Any) -> Any: | |
| if isinstance(value, np.ndarray): | |
| return [sanitize_value(item) for item in value.tolist()] | |
| if isinstance(value, np.generic): | |
| return sanitize_value(value.item()) | |
| if isinstance(value, (np.floating, float)): | |
| if math.isnan(value) or math.isinf(value): | |
| return None | |
| return float(value) | |
| if isinstance(value, (np.integer, int)): | |
| return int(value) | |
| if isinstance(value, (np.bool_, bool)): | |
| return bool(value) | |
| if isinstance(value, pd.Timestamp): | |
| return value.isoformat() | |
| if isinstance(value, pd.Timedelta): | |
| return str(value) | |
| if isinstance(value, pd.Series): | |
| return [sanitize_value(item) for item in value.tolist()] | |
| if isinstance(value, pd.Index): | |
| return [sanitize_value(item) for item in value.tolist()] | |
| if isinstance(value, pd.DataFrame): | |
| return sanitize_value(value.to_dict(orient="records")) | |
| if isinstance(value, Path): | |
| return str(value) | |
| if value is None: | |
| return None | |
| if isinstance(value, str): | |
| return value | |
| if isinstance(value, dict): | |
| return {str(k): sanitize_value(v) for k, v in value.items()} | |
| if isinstance(value, (list, tuple, set)): | |
| return [sanitize_value(v) for v in value] | |
| try: | |
| if pd.isna(value): | |
| return None | |
| except Exception: | |
| pass | |
| return value | |
| def dataframe_to_payload( | |
| df: pd.DataFrame | None, | |
| decimals: int | None = None, | |
| max_rows: int | None = None, | |
| ) -> dict[str, Any] | None: | |
| if df is None: | |
| return None | |
| # Evita cópia quando não há transformação necessária. | |
| df_work = df if decimals is None else df.copy() | |
| if decimals is not None: | |
| numeric_cols = [ | |
| col | |
| for col in df_work.select_dtypes(include=[np.number]).columns | |
| if not _is_coordinate_column(col) | |
| ] | |
| if numeric_cols: | |
| df_work.loc[:, numeric_cols] = df_work.loc[:, numeric_cols].round(decimals) | |
| total_rows = len(df_work) | |
| # Regra de integridade: payloads nunca devem suprimir linhas. | |
| truncated = False | |
| columns = [str(c) for c in df_work.columns] | |
| rows: list[dict[str, Any]] = [] | |
| # itertuples é significativamente mais leve que iterrows para payloads grandes. | |
| for tuple_row in df_work.itertuples(index=True, name=None): | |
| payload_row = {"_index": sanitize_value(tuple_row[0])} | |
| for col, value in zip(columns, tuple_row[1:]): | |
| payload_row[col] = sanitize_value(value) | |
| rows.append(payload_row) | |
| columns_payload = ["_index"] + columns | |
| return { | |
| "columns": columns_payload, | |
| "rows": rows, | |
| "total_rows": total_rows, | |
| "returned_rows": len(rows), | |
| "truncated": truncated, | |
| } | |
| def figure_to_payload(fig: Any) -> dict[str, Any] | None: | |
| if fig is None: | |
| return None | |
| try: | |
| return sanitize_value(fig.to_plotly_json()) | |
| except Exception: | |
| return None | |