Das deutsche Python-Forum

Code: Alles auswählen

import csv
from datetime import datetime as dt
from datetime import timedelta as td
from urllib.request import urlopen

import matplotlib.pyplot as plt


url= "https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/T_AM1H.csv"
dateiname = "Temperatur.csv"


with open (dateiname, "wb") as datei:
    req = urlopen(url)
    # nicht alles aufeinmal laden
    while True:
        chunk = req.read(64 * 1024)
        if not chunk:
            # keine Daten mehr, Datei zuende.
            break
        # Schreibe den block
        datei.write(chunk)


field = "BONN T AM1H [°C]"
dt_format = '%d.%m.%Y %H:%M'


with open("Temperatur.csv") as datei:
    next(datei)
    # datei ist ein Iterator
    # Im Header steht ein Kommentar, der zuerst verworfen werden muss
    # next(datei) gibt die aktuelle Zeile aus, die hier aber nicht verwnedet wird
    dict_reader = csv.DictReader(datei, delimiter=";")
    # der DictReader bekommt als erstes Argument das Dateiobjekt
    # dieser liest dann die erste Zeile mit den Feldnamen ein.
    # Da der Kommentar bereits übersprungen ist, funktioniert das
    dates = []
    values = []
    for row in dict_reader:
        value = row[field].replace(",", ".")
        if not value:
            # wenn kein Wert vorhanden ist,
            # dann überspringen
            continue
        
        datum = row["Datum"]
        zeit = row["Zeit"]
        # offensichtlich kommt auch 24 Uhr in den Datensätzen vor
        # das wäre dann schon der nächste Tag
        # 24:00 lässt sich aber nicht parsen
        # also setze ich das auf 00:00 und
        # füge einen Tag hinzu.
        if '24:00' == zeit:
          zeit = '00:00'
          next_day = True
        else:
          next_day = False
          
        zeitstempel =  datum + " " + zeit
        # Hier tritt ein Fehler auf, wenn die Zeitangabe
        # ungültig ist
        # ggf. mit try... except abfangen und überspringen
        zeitstempel = dt.strptime(zeitstempel, dt_format)

        # nächster Tag
        if next_day:
          zeitstempel += td(days=1)

        
        # Deutsch > Englisch
        # float("1,5") geht z.b. nicht
        # deswegen , durch . ersetzen

        # Wert auslesen und Komma durch Punkt ersetzen
        # Wert in float umwandeln
        # Bei der Umwandlung kann es zu einem ValueError
        # kommen, sofern der string nicht in einen float
        # umgewandelt werden kann
        # auch hier, kann man mit try und except den Fehler
        # abfangen und entsprechend reagieren.
        value = float(value)
        # wert zur Liste hinzufügen
        values.append(value)
        dates.append(zeitstempel)
        


plt.plot_date(dates, values, '-')
plt.show()

Code: Alles auswählen

import csv
import os
from datetime import datetime
from datetime import timedelta
from urllib.request import urlopen

import matplotlib.pyplot as plt


def chunked_download(url, dateiname):
  with open (dateiname, "wb") as datei:
      req = urlopen(url)
      # nicht alles aufeinmal laden
      while True:
          chunk = req.read(64 * 1024)
          if not chunk:
              # keine Daten mehr, Datei zuende.
              break
          # Schreibe den block
          datei.write(chunk)


def zeitstempel_umwandeln(datum, zeit):
    datetime_format = '%d.%m.%Y %H:%M'
    if '24:00' == zeit:
      zeit = '00:00'
      next_day = True
    else:
      next_day = False
    zeitstempel =  datum + " " + zeit
    zeitstempel = datetime.strptime(zeitstempel, datetime_format)
    if next_day:
      zeitstempel += timedelta(days=1)
    return zeitstempel


def werte_auslesen(dateiname, feld, start_datum, end_datum):
    x = []
    y = []
    with open(dateiname) as datei:
        next(datei) # 1. header überspringen
        dict_reader = csv.DictReader(datei, delimiter=";")
        for row in dict_reader:
            datum = row["Datum"]
            zeit = row["Zeit"]
            zeitstempel = zeitstempel_umwandeln(datum, zeit)
            if start_datum <= zeitstempel <= end_datum:
                wert = row[feld].replace(",", ".")
                if not wert:
                    continue
                wert = float(wert)
                x.append(zeitstempel)
                y.append(wert)
    return x, y


url = "https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/T_AM1H.csv"
dateiname = "Temperatur.csv"
feld = "BONN T AM1H [°C]"


if not os.path.exists(dateiname):
    # nur herunterladen, falls die Datei fehlt
    # Das schont den Server
    # Sind nicht unsere Ressourcen
    chunked_download(url, dateiname)


start = datetime(2019, 1, 29)
ende = datetime(2020, 1, 30)
x, y = werte_auslesen(dateiname, feld, start, ende)

# Für Matplotlib gibt es viele
# Tutorials
# Docs: https://matplotlib.org/api/_as_gen/matplotlib.pyplot.plot.html#matplotlib.pyplot.plot
# Eigentlich soll man plt.plot und plt.plot_date nicht nutzen,
# sondern das objektorientierte Interface.
# Kommt daher, weil matplotlib an Matlab angelehnt ist.

plt.plot_date(x, y, '-')
plt.show()

Code: Alles auswählen

from datetime import datetime as DateTime, timedelta as TimeDelta

Code: Alles auswählen

from datetime import (
    datetime as DateTime,
    timedelta as TimeDelta,
)

Code: Alles auswählen

def chunked_download(url, dateiname):
    with open(dateiname, "wb") as datei:
        with urlopen(url) as response:
            copyfileobj(response, datei)

Code: Alles auswählen

import csv
import matplotlib.dates as md
import matplotlib.ticker as mt
#Zeit angaben
from datetime import datetime as dt
from datetime import timedelta as td
#Downloaden der CSV datei
from urllib.request import urlopen
#Für Random Zahl
from random import *
#Matplotlib
import matplotlib.pyplot as plt

#Vorstellung
print("Grafische Darstellung von Wetterdaten \n"
      f"--------------------------------------\n\n"
      f"Dieses Programm erzeugt ein Diagramm mit\n"
      "aktuellen Wetterdaten der letzten Tage für Bonn. \n"
      f"Gib an, für welchen Messwert das Diagramm angezeigt werden \n"
      f"soll und wie viele Tage berücksichtig werden sollen. \n\n"
      f"Folgende Messwerte stehen zur Auswahl: \n\n"
      f"\t Wetterdaten \n"
      f"\t -----------\n\n"
      f"  0) Temperatur \n"
      f"  1) Relative Luftfeuchtigkeit \n"
      f"  2) Windgeschwindigkeit \n"
      f"  3) Windrichtung \n\n"
      f"\t Schadstoffe \n"
      f"\t -----------\n\n"
      f"  4) Stickstoffmonoxid (NO)\n"
      f"  5) Stickstoffdioxid (NO3)\n"
      f"  6) Partikel PM10 \n")

#Input für User
#int = Integer -> NUR ZAHL
try:
   Messwert = int(input("Gib die Kennziffer des gewünschten Messwertes ein: "))
except(ValueError):
      print("\nDies ist leider keine Kennziffer. \n")
      Messwert = int(input("Gib die Kennziffer des gewünschten Messwertes ein: "))


#Solange Messwert nicht 0 - 6       
while Messwert not in (range(0, 7)):
   print("\nDiese Kenziffer existiert nicht. \n")
   
#Wiederholung wenn Messwert nicht 0 - 6
   Messwert = int(input("Gib die Kennziffer des gewünschten Messwertes ein: "))
#---------------------------Temperatur-------------------------------------
if Messwert == 0:
   url = ("https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/T_AM1H.csv")
   stadt = "BONN T AM1H [°C]"
   dateiname = "Temperatur.csv"
   labelname = "Temperatur"
   MesswertEinheit = "Messwert in °C"
   print("\nSie haben die Auswertung für Messwert „Temperatur“ gewählt.")
#---------------------------Relative Luftfeuchtigkeit---------------------- 
elif Messwert == 1:
    url = ("https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/F_AM1H.csv")
    stadt = "BONN F AM1H [%]"
    dateiname = "RelativeLuftfeuchtigkeit.csv"
    labelname = "Relative Luftfeuchtigkeit"
    MesswertEinheit = "Messwert in °C"
    print("\nSie haben die Auswertung für Messwert „Relative Luftfeuchtigkeit“ gewählt.")
#---------------------------Windgeschwindigkeit----------------------------     
elif Messwert == 2:
    url = ("https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/WG_SM1H.csv")
    stadt = "BONN WG SM1H [m/s]"
    dateiname = "Windgeschwindigkeit.csv"
    labelname ="Windgeschwindigkeit"
    MesswertEinheit = "Messwert in m/s"
    print("\nSie haben die Auswertung für Messwert „Windgeschwindigkeit“ gewählt.")
#---------------------------Windrichtung-----------------------------------     
elif Messwert == 3:
    url = ("https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/WR_VM1H.csv")
    stadt = "BONN WR VM1H [°]"
    dateiname = "Windrichtung.csv"
    labelname ="Windrichtung"
    MesswertEinheit = "Messwert"
    print("\nSie haben die Auswertung für Messwert „Windrichtung“ gewählt.")
#---------------------------Stickstoffmonoxid (NO)-------------------------   
elif Messwert == 4:
    url = ("https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/NO_AM1H.csv")
    stadt ="BONN NO AM1H [µg/m³]"
    dateiname = "Stickstoffmonoxid.csv"
    labelname ="Stickstoffmonoxid (NO)"
    MesswertEinheit = "Messwert in µg/m³"
    print("\nSie haben die Auswertung für Messwert „Stickstoffmonoxid (NO)“ gewählt.")
#---------------------------Stickstoffdioxid (NO2)-------------------------    
elif Messwert == 5:
    url = ("https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/NO2_AM1H.csv")
    stadt ="BONN NO2 AM1H [µg/m³]"
    dateiname = "Stickstoffdioxid.csv"
    labelname ="Stickstoffdioxid (NO2)"
    MesswertEinheit = "Messwert in µg/m³"
    print("\nSie haben die Auswertung für Messwert „Stickstoffdioxid (NO2)“ gewählt.")
#---------------------------Partikel PM10----------------------------------     
elif Messwert == 6:
    url = ("https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/PM10F_GM24H.csv")
    stadt ="BONN PM10F GM24H [µg/m³]"
    dateiname = "Partikel.csv"
    labelname ="Partikel PM10"
    MesswertEinheit = "Messwert in µg/m³"
    print("\nSie haben die Auswertung für Messwert „Partikel PM10“ gewählt.")

#If = wenn Messwert = 0 Temperatur ausführen
#elif  =  -> Oder (Fall unterscheidung)
#else wäre ->  UND


#CSV-Datei herunterladen
with open(dateiname, "wb") as datei:
   datei.write(urlopen(url).read())

with open(dateiname, encoding="ansi") as datei:
    next(datei)
    # next(datei) überspringen der 1. Zeile 
    zeile = csv.DictReader(datei, delimiter=";")
    # dieser liest dann die erste Zeile mit den Feldnamen ein.
    datumlist = []
    wertlist = []
    for row in zeile:
        werte = row[stadt].replace(",", ".")
        #Alle
        if "<10" == werte:
            werte = randint(0,9)
        elif "<7" == werte:
            werte = randint(0,6)
        #da bei Stickstoffmonoxid und Partike PM10 "<10 und <7"
        #auftauchen und nicht gefloutet werden können,
        #ersetze <10 mit einer Zufälligen Zahl zwischen 0 - 10
        #und <7 mit einer Zufälligen Zahl zwischen 0 - 6
        elif not werte:
           werte == 0
            # wenn kein Wert vorhanden ist,
            # dann = 0 setzen
           continue
            
        
        datum = row["Datum"]
        zeit = row["Zeit"]
        #Da wir 24:00 uhr schon der nächste Tag ist,
        #Setzte ich es zu 00:00
        if '24:00' == zeit:
          zeit = '00:00'
          next_day = True
        else:
          next_day = False

        
          
        zeitstempel =  datum + " " + zeit
        #Datum und Zeit werden in einer einzigen Zeitangabe
        #zusammengebaut -> numerischen Datumswert ungeformt
        zeitstempel = dt.strptime(zeitstempel, '%d.%m.%Y %H:%M')

        # nächster Tag
        if next_day:
          zeitstempel += td(days=1)

        werte = float(werte)
        # wert zur Liste hinzufügen
        wertlist.append(werte)
        datumlist.append(zeitstempel)
        
# Größe auf dem Bildschirm vorgeben, hier ungefähr 8 mal 6 Zoll:
fig, ax = plt.subplots(figsize=(10,6))
# Überschrift:
plt.title(labelname)

# Messkurve, mit Label für Legende, einzeichnen:
plt.plot(datumlist, wertlist, label=labelname )
plt.xlabel("Datum")
plt.ylabel(MesswertEinheit)
# Die gewünschte Datumsformatierung der x-Achse definieren.
ax.xaxis.set_major_formatter(md.DateFormatter("%d.%m.%Y"))


# Für die untergeordneten Teilstriche kann man zusätzlich eine „minor“
# Formatierung festlegen.
# Diese Teilstriche sollen hier beispielsweise 0.5°C auflösen.
ax.yaxis.set_minor_locator(mt.MultipleLocator(0.5))

# Datumsbeschriftung der x-Achse um 30° drehen:
fig.autofmt_xdate(rotation=30)

# Den überflüssigen Freiraum vor und hinter der Diagrammkurve abschneiden:
plt.xlim(datumlist[0],datumlist[-1])

# Ein Raster erleichtert das Ablesen.
plt.grid(color="lightgrey")

# Eine Legende hinzufügen:
plt.legend()

# Vorhandenen Platz optimal ausnutzen, unnötige Ränder entfernen,
# Überlappen von Grafikelementen verhindern:
plt.tight_layout()

# Zeigt Plot an
plt.show()

Code: Alles auswählen

#!/usr/bin/env python3
import csv
import os
from collections import namedtuple
from datetime import datetime as DateTime, timedelta as TimeDelta
from random import randrange
from urllib.request import urlretrieve

from matplotlib import pyplot as plt
from matplotlib.dates import DateFormatter
from matplotlib.ticker import MultipleLocator

BASE_URL = "https://www.lanuv.nrw.de/fileadmin/lanuv/luft/temes/"


MeasurementDescription = namedtuple(
    "MeasurementDescription",
    "name remote_filename column_name local_filename unit",
)

MEASUREMENT_DESCRIPTIONS = [
    MeasurementDescription(
        "Temperatur", "T_AM1H.csv", "BONN T AM1H [°C]", "Temperatur.csv", "°C"
    ),
    MeasurementDescription(
        "Relative Luftfeuchtigkeit",
        "F_AM1H.csv",
        "BONN F AM1H [%]",
        "RelativeLuftfeuchtigkeit.csv",
        "%",
    ),
    MeasurementDescription(
        "Windgeschwindigkeit",
        "WG_SM1H.csv",
        "BONN WG SM1H [m/s]",
        "Windgeschwindigkeit.csv",
        "m/s",
    ),
    MeasurementDescription(
        "Windrichtung",
        "WR_VM1H.csv",
        "BONN WR VM1H [°]",
        "Windrichtung.csv",
        "°",
    ),
    MeasurementDescription(
        "Stickstoffmonoxid (NO)",
        "NO_AM1H.csv",
        "BONN NO AM1H [µg/m³]",
        "Stickstoffmonoxid.csv",
        "µg/m³",
    ),
    MeasurementDescription(
        "Stickstoffdioxid (NO2)",
        "NO2_AM1H.csv",
        "BONN NO2 AM1H [µg/m³]",
        "Stickstoffdioxid.csv",
        "µg/m³",
    ),
    MeasurementDescription(
        "Partikel PM10",
        "PM10F_GM24H.csv",
        "BONN PM10F GM24H [µg/m³]",
        "Partikel.csv",
        "µg/m³",
    ),
]


def ask_for_measurement_description():
    print(
        "Grafische Darstellung von Wetterdaten \n"
        "--------------------------------------\n\n"
        "Dieses Programm erzeugt ein Diagramm mit\n"
        "aktuellen Wetterdaten der letzten Tage für Bonn.\n"
        "Gib an, für welchen Messwert das Diagramm angezeigt werden\n"
        "soll und wie viele Tage berücksichtig werden sollen.\n\n"
        "Folgende Messwerte stehen zur Auswahl:\n"
    )
    for i, description in enumerate(MEASUREMENT_DESCRIPTIONS):
        print(f"{i:>3}) {description.name}")
    print()
    while True:
        try:
            index = int(
                input("Gib die Kennziffer des gewünschten Messwertes ein: ")
            )
        except ValueError:
            print("\nDies ist leider keine Kennziffer. \n")
        else:
            if 0 <= index < len(MEASUREMENT_DESCRIPTIONS):
                description = MEASUREMENT_DESCRIPTIONS[index]
                print(
                    f"\nSie haben die Auswertung für Messwert"
                    f" „{description.name}“ gewählt."
                )
                return description
            print("\nDiese Kenziffer existiert nicht. \n")


def parse_value(string):
    string = string.strip().replace(",", ".")
    if not string:
        return None
    #
    # Da bei Stickstoffmonoxid und Partikel PM10 "<10" und "<7"
    # auftauchen und nicht gefloutet werden können, ersetze "<n" mit
    # einer zufälligen Zahl zwischen 0 und n-1.
    #
    return (
        randrange(int(string[1:])) if string.startswith("<") else float(string)
    )


def parse_timestamp(date_string, time_string):
    #
    # Statt 00:00 Uhr gibt es 24:00 Uhr was 00:00 Uhr am nächsten
    # Tag entspricht.
    #
    date_needs_adjustment = time_string == "24:00"
    if date_needs_adjustment:
        time_string = "00:00"

    timestamp = DateTime.strptime(
        f"{date_string} {time_string}", "%d.%m.%Y %H:%M"
    )

    if date_needs_adjustment:
        timestamp += TimeDelta(days=1)

    return timestamp


def load_data(filename, column_name):
    with open(filename, encoding="latin-1") as datei:
        next(datei)  # Kommentarzeile am Dateianfang überspringen.
        timestamps = list()
        values = list()
        for row in csv.DictReader(datei, delimiter=";"):
            value = parse_value(row[column_name])
            if value is not None:
                timestamps.append(parse_timestamp(row["Datum"], row["Zeit"]))
                values.append(value)

    assert len(values) == len(timestamps)
    return timestamps, values


def plot_data(timestamps, values, title, unit):
    figure, axes = plt.subplots(figsize=(10, 6))
    plt.title(title)
    plt.plot(timestamps, values, label=title)
    plt.xlabel("Datum")
    plt.ylabel(f"Messwerte in {unit}")
    axes.xaxis.set_major_formatter(DateFormatter("%d.%m.%Y"))
    axes.yaxis.set_minor_locator(MultipleLocator(0.5))
    figure.autofmt_xdate(rotation=30)
    plt.xlim(timestamps[0], timestamps[-1])
    plt.grid(color="lightgrey")
    plt.legend()
    plt.tight_layout()
    plt.show()


def main():
    description = ask_for_measurement_description()

    if not os.path.exists(description.local_filename):
        urlretrieve(
            BASE_URL + description.remote_filename, description.local_filename
        )

    timestamps, values = load_data(
        description.local_filename, description.column_name
    )
    plot_data(timestamps, values, description.name, description.unit)


if __name__ == "__main__":
    main()