multipart-chunk-upload.py

#!/usr/bin/env python3

import base64
import os
import sys
import uuid

import requests


class MemfaultPostChunksMultipart:
    def __init__(self, memfault_project_key, device_serial, use_base64=False):
        self.memfault_project_key = memfault_project_key
        self.device_serial = device_serial
        self.multipart_delimiter = uuid.uuid4().hex

        self.use_base64 = use_base64

    def post(self, files):
        my_stream = b""
        for file in files:
            # load the file data
            if self.use_base64:
                file_data = base64.b64encode(open(file, mode="rb").read())
            else:
                file_data = open(file, mode="rb").read()

            # format the multipart fragment
            my_stream += (
                b"--"
                + self.multipart_delimiter.encode("ascii")
                + b"\r\n"
                + b"Content-Type: application/octet-stream\r\n"
                + b"Content-Length: "
                + str(len(file_data)).encode("ascii")
                + b"\r\n\r\n"
                + file_data
                + b"\r\n"
            )

        headers = {
            "Memfault-Project-Key": self.memfault_project_key,
            "Content-Type": "multipart/mixed; boundary={}".format(self.multipart_delimiter),
            "Content-Length": f"{len(my_stream)}",
        }

        if self.use_base64:
            headers["Content-Encoding"] = "base64"

        req = requests.Request(
            "POST",
            f"https://chunks.memfault.com/api/v0/chunks/{self.device_serial}",
            headers=headers,
            data=my_stream,
        ).prepare()

        print(
            "{}\n{}\r\n{}\r\n\r\n{}".format(
                "-----------START-----------",
                req.method + " " + req.url,
                "\r\n".join("{}: {}".format(k, v) for k, v in req.headers.items()),
                req.body.decode("ascii", "ignore"),
            )
        )

        return requests.Session().send(req)


def main():
    files = sys.argv[1:]
    memfault_project_key = os.environ.get("MEMFAULT_PROJECT_KEY")
    assert memfault_project_key, "MEMFAULT_PROJECT_KEY environment variable must be set"
    device_serial = os.environ.get("MEMFAULT_DEVICE_SERIAL")
    assert device_serial, "MEMFAULT_DEVICE_SERIAL environment variable must be set"

    # do the multipart upload
    print("Uploading chunks...")
    response = MemfaultPostChunksMultipart(
        memfault_project_key, device_serial, use_base64=False
    ).post(files)
    print(response)
    print(response.text)


if __name__ == "__main__":
    main()

You can test it using the test payloads from here:
https://docs.memfault.com/docs/mcu/test-patterns-for-chunks-endpoint#same-event-message-encoded-in-two-chunks

❯ MEMFAULT_DEVICE_SERIAL=CURL_TESTSERIAL MEMFAULT_PROJECT_KEY="YOUR_PROJECT_KEY" \
  ./multipart-chunk-upload.py ~/Downloads/chunk_v2_chunk_msg_pt*.bin