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)

        # Keep the newest rows by position, not by index label: advertisement
        # timestamps are whole seconds, so duplicate labels are normal.
        if len(devices[serial]) > MAX_ROWS_PER_DEVICE:
            devices[serial] = devices[serial].iloc[-MAX_ROWS_PER_DEVICE:]

    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

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()]