Spaces:
Running
Running
File size: 3,415 Bytes
d6c9678 f7fe834 d6c9678 86d4f9b 614e632 86d4f9b d6c9678 0a8b8ba d6c9678 f7fe834 d6c9678 614e632 d6c9678 0a8b8ba d6c9678 0a8b8ba d6c9678 0a8b8ba d6c9678 0a8b8ba d6c9678 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | 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
|