Das deutsche Python-Forum

Code: Alles auswählen

import thingspeak

READ_ID = "2018608"
READ_KEY = "OTXA9V7CA8H6KAOE"

def readData():
    # initialize the connection to thingspeak
    channel = thingspeak.Channel(READ_ID, READ_KEY)
    response = channel.get()
    print(response)

if __name__ == "__main__":
    readData()

Code: Alles auswählen

    def get(self, options=None):
        """Get a channel feed.

        `get-a-channel-feed
        <https://mathworks.com/help/thingspeak/get-a-channel-feed.html>`_
        """

Code: Alles auswählen

import urllib.request, json

Code: Alles auswählen

https://api.thingspeak.com/channels/2018608/feeds.json?start=2024-06-19 17:50:18

Code: Alles auswählen

import urllib.request

strDateTime = "2024-06-19 17:50:18"
READ_ID = "2018608"

url = "https://api.thingspeak.com/channels/"+READ_ID+"/feeds.json?start=" + strDateTime
response = urllib.request.urlopen(url)
data = json.loads(response.read())
print(data)

Code: Alles auswählen

http.client.InvalidURL: URL can't contain control characters. '/channels/2018608/feeds.json?start=2024-06-19 17:50:18' (found at least ' ')

Code: Alles auswählen

import urllib
READ_ID = 2018608
FEEDS_URL = f"https://api.thingspeak.com/channels/{READ_ID}/feeds.json"

def main():
    start_time = "2024-06-19 17:50:18"
    parameters = urllib.parse.urlencode({'start': start_time})
    url = f"{URL}?{parameters}"
    response = urllib.request.urlopen(url)
    data = json.loads(response.read())
    print(data)

if __name__ == "__main__":
    main()

Code: Alles auswählen

import copy
import datetime
import urllib.request
import urllib.parse
import json
import pprint
from decimal import Decimal


BASE_URL = "https://api.thingspeak.com/channels"
FEEDS = "feeds.json"


def get_channel(channel: int | str, since: datetime.datetime | None = None, days: int | None = None) -> dict[str, str]:
   if since and days:
       raise RuntimeError("The arguments `since` and `days` are mutually exclusive")

   if days:
       start = datetime.datetime.now() - datetime.timedelta(days=days)
   elif since:
       start = since
   else:
       raise ValueError("Neither `since` nor `days` are given.")

   channel = str(channel)
   url = "/".join((BASE_URL, channel, FEEDS)) + f"?start={start.isoformat()}"

   with urllib.request.urlopen(url) as fd:
       return json.load(fd)


def to_datetime(dt_str: str) -> datetime.datetime:
    return datetime.datetime.fromisoformat(dt_str.removesuffix("Z"))


def transform_data(data: dict[str, str]) -> dict[str: str | int | float | None]:
    # doing a deepcopy to prevent modification of the original dict given by the argument data
    data_copy = copy.deepcopy(data)
    channel = data_copy["channel"]
    feeds = data_copy["feeds"]

    channel["created_at"] = to_datetime(channel["created_at"])
    channel["updated_at"] = to_datetime(channel["updated_at"])
    channel["id"] = int(channel["id"])
    channel["longitude"] = Decimal(channel["longitude"])
    channel["latitude"] = Decimal(channel["latitude"])


    for index, feed in enumerate(feeds):
        current_feed = feeds[index]
        current_feed["created_at"] = to_datetime(current_feed["created_at"])
        current_feed["entry_id"] = int(current_feed["entry_id"])

        for field in current_feed:
            if field.startswith("field"):
                if current_feed[field] is None:
                    continue

                try:
                    current_feed[field] = int(current_feed[field])
                except ValueError:
                    try:
                        feeds[index][field] = float(feeds[index][field])
                    except ValueError:
                        continue


    return data_copy


result = get_channel(2018608, days=1)
transformed_result = transform_data(result)

# pprint.pprint(result)
pprint.pprint(transformed_result)

Code: Alles auswählen

def transform_data(data):
    #
    # doing a deepcopy to prevent modification of the original dict given by the
    # argument data
    #
    data_copy = copy.deepcopy(data)

    channel = data_copy["channel"]
    channel["created_at"] = to_datetime(channel["created_at"])
    channel["updated_at"] = to_datetime(channel["updated_at"])
    channel["id"] = int(channel["id"])
    channel["longitude"] = Decimal(channel["longitude"])
    channel["latitude"] = Decimal(channel["latitude"])

    for feed in data_copy["feeds"]:
        for name, value in feed.items():
            if name == "created_at":
                feed[name] = to_datetime(value)
            elif name == "entry_id":
                feed[name] = int(value)
            elif name.startswith("field"):
                try:
                    feed[name] = int(value)
                except ValueError:
                    try:
                        feed[name] = float(value)
                    except ValueError:
                        pass

    return data_copy

Code: Alles auswählen

def try_int_or_float(value):
    try:
        return int(value)
    except ValueError:
        try:
            return float(value)
        except ValueError:
            return value

def convert_types(data, types, default=lambda x:x):
    return {
        key: types.get(key, default)(value)
        for key, value in data
    }

def transform_data(data):
    return {
        "channel": convert_types(data["channel"], {
            "id": int,
            "longitude": Decimal,
            "latitude": Decimal,
            "created_at": to_datetime,
            "updated_at": to_datetime
        }),
        "feeds": [
            convert_types(feed, {
                "entry_id": int,
                "created_at": to_datetime,
            }, default=try_int_or_float)
            for feed in data["feeds"]
        ]
    }