naneos.cloud.upload

Upload of gathered snapshots to the naneos IoT service.

backend_status(response)

The status the backend answered with, and what it said (empty if nothing).

The API gateway can wrap the answer of the lambda: a route the deployed lambda does not know comes back as HTTP 200 whose body is {"statusCode": 404, "body": ...}. Trusting the HTTP status alone would count that upload as done.

Source code in src/naneos/cloud/upload.py

def backend_status(response: requests.Response) -> tuple[int, str]:
    """The status the backend answered with, and what it said (empty if nothing).

    The API gateway can wrap the answer of the lambda: a route the deployed
    lambda does not know comes back as HTTP 200 whose body is
    {"statusCode": 404, "body": ...}. Trusting the HTTP status alone would
    count that upload as done.
    """
    if response.status_code != 200:
        return response.status_code, ""
    try:
        wrapper = response.json()
    except ValueError:
        return 200, ""
    if not isinstance(wrapper, dict) or not isinstance(wrapper.get("statusCode"), int):
        return 200, ""
    return wrapper["statusCode"], str(wrapper.get("body", ""))[:200]

build_body(upload_string)

The JSON envelope the backend expects around the base64 protobuf.

Source code in src/naneos/cloud/upload.py

def build_body(upload_string: str) -> str:
    """The JSON envelope the backend expects around the base64 protobuf."""
    return json.dumps(
        {
            "gateway": "python_webhook",
            "data": upload_string,
            "published_at": datetime.datetime.now(datetime.UTC).isoformat(),
        }
    )

build_combined_entry(data, abs_time)

The protobuf message for a snapshot, with timestamps relative to abs_time.

Source code in src/naneos/cloud/upload.py

def build_combined_entry(data: dict[int, pd.DataFrame], abs_time: int) -> pb.CombinedData:
    """The protobuf message for a snapshot, with timestamps relative to abs_time."""
    return build_prepared_entry(prepare_frames(data), abs_time)

build_prepared_entry(frames, abs_time)

The message for frames from prepare_frames(); they must not go through it twice.

Source code in src/naneos/cloud/upload.py

def build_prepared_entry(frames: dict[int, pd.DataFrame], abs_time: int) -> pb.CombinedData:
    """The message for frames from prepare_frames(); they must not go through it twice."""
    devices = [create_proto_device(sn, abs_time, df) for sn, df in frames.items()]
    return create_combined_entry(devices=devices, abs_timestamp=abs_time)

prepare_frames(data)

The frames of a snapshot as the upload needs them, small enough to keep for a day.

One row per second (see to_upload_frame) and only the columns that have a field on the wire and data in them: 21 of the 56 columns of a P2 frame. The message built from the result is the message built from the full frame. Frames from here can be concatenated and sent with send_frames().

Source code in src/naneos/cloud/upload.py

def prepare_frames(data: dict[int, pd.DataFrame]) -> dict[int, pd.DataFrame]:
    """The frames of a snapshot as the upload needs them, small enough to keep for a day.

    One row per second (see to_upload_frame) and only the columns that have a
    field on the wire and data in them: 21 of the 56 columns of a P2 frame. The
    message built from the result is the message built from the full frame.
    Frames from here can be concatenated and sent with send_frames().
    """
    prepared: dict[int, pd.DataFrame] = {}
    for serial, df in data.items():
        if df.empty:
            continue
        frame = to_upload_frame(df)
        keep = [column for column in UPLOAD_COLUMNS if column in frame.columns]
        prepared[serial] = _compact(frame[keep].dropna(axis=1, how="all"))
    return prepared

send_frames(frames)

Upload frames from prepare_frames(), keyed by device serial number.

The frames may hold the rows of several snapshots: the timestamps in the message are relative to now, so old data lands at its own time. Blocks for up to TIMEOUT_SECONDS, more for a request with many rows. Raises requests.RequestException on network problems (requests.ReadTimeout when the server was reached but did not answer in time); HTTP errors are reported by the returned response.

Source code in src/naneos/cloud/upload.py

def send_frames(frames: dict[int, pd.DataFrame]) -> requests.Response:
    """Upload frames from prepare_frames(), keyed by device serial number.

    The frames may hold the rows of several snapshots: the timestamps in the
    message are relative to now, so old data lands at its own time.
    Blocks for up to TIMEOUT_SECONDS, more for a request with many rows. Raises
    requests.RequestException on network problems (requests.ReadTimeout when the
    server was reached but did not answer in time); HTTP errors are reported by
    the returned response.
    """
    # ceil, not int: the rows are rounded to the nearest second (to_upload_frame), so a sample
    # from the second half of the current second lies after int(now), and would get an age of -1.
    abs_time = math.ceil(time.time())
    rows = sum(len(df) for df in frames.values())
    timeout = min(MAX_TIMEOUT_SECONDS, TIMEOUT_SECONDS + TIMEOUT_SECONDS_PER_ROW * rows)
    return _post(URL_COMBINED_DATA, build_prepared_entry(frames, abs_time), timeout)

to_upload_frame(df)

Prepare one device frame for the backend: one row per whole second, no inf.

Frames are indexed by unix time in milliseconds (see naneos.frames). The index is rounded, not truncated: the devices sample at ~1Hz with a phase of their own, so truncating puts the two samples that straddle a second boundary into the same second, where one of them wins, and leaves the neighbouring second without a row at all.

The backend takes at most 1 Hz. Rows that land in the same second, as they do for a device read at 10 Hz or 100 Hz, are merged into one.

Source code in src/naneos/cloud/upload.py

def to_upload_frame(df: pd.DataFrame) -> pd.DataFrame:
    """Prepare one device frame for the backend: one row per whole second, no inf.

    Frames are indexed by unix time in milliseconds (see naneos.frames).
    The index is rounded, not truncated: the devices sample at ~1Hz with a
    phase of their own, so truncating puts the two samples that straddle a
    second boundary into the same second, where one of them wins, and
    leaves the neighbouring second without a row at all.

    The backend takes at most 1 Hz. Rows that land in the same second, as
    they do for a device read at 10 Hz or 100 Hz, are merged into one.
    """
    df = df.replace([float("inf"), -float("inf")], 0)
    df.index = pd.Index(
        np.rint(df.index.to_numpy(dtype="float64") / 1e3).astype("int64"),
        name=df.index.name,
    )
    return aggregate_duplicate_index(df)

upload_diagnostic(diagnostic)

Upload a UI curve or a pulse form to its endpoint.

Source code in src/naneos/cloud/upload.py

def upload_diagnostic(diagnostic: UiCurve | PulseForm) -> requests.Response:
    """Upload a UI curve or a pulse form to its endpoint."""
    if isinstance(diagnostic, UiCurve):
        return upload_ui_curve(diagnostic)
    return upload_pulse_form(diagnostic)

upload_pulse_form(form)

Upload one pulse form. Blocks and raises like upload_snapshot().

Source code in src/naneos/cloud/upload.py

def upload_pulse_form(form: PulseForm) -> requests.Response:
    """Upload one pulse form. Blocks and raises like upload_snapshot()."""
    return _post(URL_PULSE_FORM, create_pulse_form(form))

upload_snapshot(data)

Upload the frames of a snapshot, keyed by device serial number.

Blocks for up to TIMEOUT_SECONDS. Raises requests.RequestException on network problems; HTTP errors are reported by the returned response.

Source code in src/naneos/cloud/upload.py

def upload_snapshot(data: dict[int, pd.DataFrame]) -> requests.Response:
    """Upload the frames of a snapshot, keyed by device serial number.

    Blocks for up to TIMEOUT_SECONDS. Raises requests.RequestException on
    network problems; HTTP errors are reported by the returned response.
    """
    return send_frames(prepare_frames(data))

upload_ui_curve(curve)

Upload one UI curve. Blocks and raises like upload_snapshot().

Source code in src/naneos/cloud/upload.py

def upload_ui_curve(curve: UiCurve) -> requests.Response:
    """Upload one UI curve. Blocks and raises like upload_snapshot()."""
    return _post(URL_UI_CURVE, create_ui_curve(curve))