Simple test

Run this test on a WiFi capable CircuitPython device to ensure it works.

examples/json_stream_simpletest.py

# SPDX-FileCopyrightText: 2017 Scott Shawcroft, written for Adafruit Industries
# SPDX-FileCopyrightText: Copyright (c) 2023 Scott Shawcroft for Adafruit Industries
#
# SPDX-License-Identifier: Unlicense

import ssl
import time

import adafruit_requests
import socketpool
import wifi

import adafruit_json_stream as json_stream

pool = socketpool.SocketPool(wifi.radio)
session = adafruit_requests.Session(pool, ssl.create_default_context())

SCORE_URL = "http://site.api.espn.com/apis/site/v2/sports/baseball/mlb/scoreboard"

while True:
    resp = session.get(SCORE_URL)
    json_data = json_stream.load(resp.iter_content(32))
    for event in json_data["events"]:
        if "Seattle" not in event["name"]:
            continue
        for competition in event["competitions"]:
            for competitor in competition["competitors"]:
                print(competitor["team"]["displayName"], competitor["score"])
    resp.close()
    time.sleep(60)

CPython test

Run this with python examples/json_stream_local_file.py <filename> to test in CPython.

examples/json_stream_simpletest.py

# SPDX-FileCopyrightText: Copyright (c) 2023 Scott Shawcroft for Adafruit Industries
#
# SPDX-License-Identifier: Unlicense

import sys

import adafruit_json_stream as json_stream

# import json_stream


class FakeResponse:
    def __init__(self, file):
        self.file = file

    def iter_content(self, chunk_size):
        while True:
            yield self.file.read(chunk_size)


f = open(sys.argv[1], "rb")
obj = json_stream.load(FakeResponse(f).iter_content(32))

currently = obj["currently"]
print(currently)
print(currently["time"])
print(currently["icon"])

for i, day in enumerate(obj["daily"]["data"]):
    print(day["time"], day["summary"], day["temperatureHigh"])
    if i > 6:
        break

for source in obj["flags"]["sources"]:
    print(source)