core/export_utils.py

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