naneos.frames

pandas DataFrame helpers for NaneosDeviceDataPoint lists.

Frames are indexed by unix_timestamp (ms) and keyed by serial number in the dict[int, pd.DataFrame] structures that flow from the managers to the upload.

add_data_points_to_dict(devices, points)

Append data points to the per-serial frames, one pandas round per device.

Source code in src/naneos/frames.py

def add_data_points_to_dict(
    devices: dict[int, pd.DataFrame], points: list[NaneosDeviceDataPoint]
) -> dict[int, pd.DataFrame]:
    """Append data points to the per-serial frames, one pandas round per device."""
    by_serial: dict[int, list[NaneosDeviceDataPoint]] = {}
    for point in points:
        if point.serial_number is None:
            continue
        by_serial.setdefault(point.serial_number, []).append(point)

    for serial, serial_points in by_serial.items():
        new_rows = to_pandas_df(serial_points)
        if new_rows.empty:
            continue

        existing = devices.get(serial)
        if existing is None or existing.empty:
            devices[serial] = new_rows
        else:
            devices[serial] = pd.concat([existing, new_rows], ignore_index=False)

        # Only look at the index once the frame could be over the limit at 1 Hz.
        df = devices[serial]
        if len(df) > MAX_BUFFER_SECONDS:
            oldest_kept = int(df.index.max()) - MAX_BUFFER_SECONDS * 1000
            devices[serial] = df[df.index > oldest_kept]

    return devices

add_to_existing_naneos_data(data, new_data)

Merge a second dict of frames into the first, concatenating per serial.

Source code in src/naneos/frames.py

def add_to_existing_naneos_data(
    data: dict[int, pd.DataFrame], new_data: dict[int, pd.DataFrame]
) -> dict[int, pd.DataFrame]:
    """Merge a second dict of frames into the first, concatenating per serial."""
    for serial, df in new_data.items():
        if serial in data:
            data[serial] = pd.concat([data[serial], df], ignore_index=False)
        else:
            data[serial] = df

    return data

aggregate_duplicate_index(df)

Collapse rows that share an index label into one row per label.

Used by the upload after the index has been rounded to whole seconds: a device read at 10 Hz or 100 Hz must still reach the backend at 1 Hz. Measurements are averaged, device_status is OR-ed so that an error flagged in any sample survives, everything else keeps its last known value.

Source code in src/naneos/frames.py

def aggregate_duplicate_index(df: pd.DataFrame) -> pd.DataFrame:
    """Collapse rows that share an index label into one row per label.

    Used by the upload after the index has been rounded to whole seconds: a
    device read at 10 Hz or 100 Hz must still reach the backend at 1 Hz.
    Measurements are averaged, device_status is OR-ed so that an error flagged
    in any sample survives, everything else keeps its last known value.
    """
    if not df.index.has_duplicates:
        return df  # the normal 1 Hz case, nothing to do

    def bitwise_or(values: pd.Series) -> object:
        known = values.dropna()
        if known.empty:
            return pd.NA
        result = 0
        for value in known:
            result |= int(value)
        return result

    aggregations: dict[str, str | Callable[[pd.Series], object]] = {}
    for column in df.columns:
        if column == "device_status":
            aggregations[column] = bitwise_or
        elif pd.api.types.is_float_dtype(df[column].dtype):
            aggregations[column] = "mean"
        else:
            aggregations[column] = "last"

    aggregated = df.groupby(level=0, sort=True).agg(aggregations)
    return aggregated.astype(df.dtypes.to_dict())

device_type_of(df, default=DeviceType.P2)

The device type recorded in a frame, or default when none is known.

Source code in src/naneos/frames.py

def device_type_of(df: pd.DataFrame, default: DeviceType = DeviceType.P2) -> DeviceType:
    """The device type recorded in a frame, or default when none is known."""
    if "device_type" not in df.columns:
        return default
    known = df["device_type"].dropna()
    if known.empty:
        return default
    return DeviceType(int(known.iloc[-1]))

sort_and_clean_naneos_data(data, serial_only=None)

Prepare gathered frames for the upload.

Per device: keep only the rows of the best connection type (serial over BLE link), sort by time, drop duplicate timestamps (last wins) and settle on one device type. Devices listed in serial_only are restricted to their serial rows even if none arrived.

Source code in src/naneos/frames.py

def sort_and_clean_naneos_data(
    data: dict[int, pd.DataFrame], serial_only: list[int | None] | None = None
) -> dict[int, pd.DataFrame]:
    """Prepare gathered frames for the upload.

    Per device: keep only the rows of the best connection type (serial over
    BLE link), sort by time, drop duplicate timestamps (last wins) and settle
    on one device type. Devices listed in serial_only are restricted to their
    serial rows even if none arrived.
    """
    if serial_only is None:
        serial_only = []

    data_return: dict[int, pd.DataFrame] = {}

    for serial, df in data.items():
        if serial is None or df.empty:
            continue

        if "connection_type" in df.columns:
            connection = df["connection_type"]
            if (connection == ConnectionType.SERIAL).any() or serial in serial_only:
                df = df[connection == ConnectionType.SERIAL]
            elif (connection == ConnectionType.CONNECTED).any():
                df = df[connection == ConnectionType.CONNECTED]

        df = df.sort_index()
        df = df[~df.index.duplicated(keep="last")]
        df = _resolve_device_type(df)

        if not df.empty:
            data_return[serial] = df

    return data_return

to_pandas_df(points)

Build a single DataFrame from many data points, indexed by unix_timestamp.

Points are converted in one batch: a DataFrame construction, an astype and a concat per point is the single most expensive thing this library does on a Raspberry Pi Zero 2 W.

Source code in src/naneos/frames.py

def to_pandas_df(points: list[NaneosDeviceDataPoint]) -> pd.DataFrame:
    """Build a single DataFrame from many data points, indexed by unix_timestamp.

    Points are converted in one batch: a DataFrame construction, an astype and
    a concat per point is the single most expensive thing this library does
    on a Raspberry Pi Zero 2 W.
    """
    if not points:
        return pd.DataFrame()

    df = pd.DataFrame([p.to_dict(remove_nan=False) for p in points])
    df = _apply_dtypes(df)
    df = df.set_index(["unix_timestamp"], drop=True)
    # A point that never received a timestamp cannot be placed on the time axis.
    return df[df.index.notna()]