Spaces:
Sleeping
Sleeping
File size: 2,245 Bytes
199bfa3 | 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 | """
Export utilities for the Advanced Data Explorer.
Downsampling, multi-window CSV packaging, size estimation.
"""
import io
import zipfile
import pandas as pd
from typing import List, Optional
def downsample_dataframe(df: pd.DataFrame, target_hz: float) -> pd.DataFrame:
"""Downsample a time-indexed DataFrame to a target frequency.
Args:
df: DataFrame with a 'timestamp' column
target_hz: Target frequency in Hz
Returns:
Downsampled DataFrame
"""
if "timestamp" not in df.columns or len(df) < 2:
return df
sorted_df = df.sort_values("timestamp")
dt = sorted_df["timestamp"].diff().dt.total_seconds().dropna()
current_hz = 1.0 / dt.median() if dt.median() > 0 else 1.0
if target_hz >= current_hz:
return sorted_df
step = max(1, int(round(current_hz / target_hz)))
return sorted_df.iloc[::step].reset_index(drop=True)
def package_multi_window_csv(
dfs: List[pd.DataFrame],
labels: Optional[List[str]] = None,
base_filename: str = "export",
) -> bytes:
"""Package multiple DataFrames into a zip of CSVs.
Args:
dfs: List of DataFrames to export
labels: Optional labels for each window (used in filenames)
base_filename: Base filename for the zip
Returns:
Bytes of the zip file
"""
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zf:
for i, df in enumerate(dfs):
label = labels[i] if labels and i < len(labels) else f"window_{i + 1}"
safe_label = label.replace(" ", "_").replace("/", "-")
csv_name = f"{base_filename}_{safe_label}.csv"
csv_bytes = df.to_csv(index=False).encode("utf-8")
zf.writestr(csv_name, csv_bytes)
buffer.seek(0)
return buffer.getvalue()
def estimate_csv_size(df: pd.DataFrame) -> str:
"""Estimate CSV file size as a human-readable string."""
# Rough estimate: 50 bytes per cell
cells = df.shape[0] * df.shape[1]
bytes_est = cells * 50
if bytes_est < 1024:
return f"{bytes_est} B"
elif bytes_est < 1024 * 1024:
return f"{bytes_est / 1024:.0f} KB"
else:
return f"{bytes_est / (1024 * 1024):.1f} MB"
|