Das deutsche Python-Forum

Code: Alles auswählen

#!/usr/bin/env python
import RPi.GPIO as GPIO
import Adafruit_DHT as DHT
from datetime import datetime as DateTime
import json
import time
import pytz
import urllib2
import requests
import thread
import json_lines
from pprint import pprint
global jsonData
Sensor = 11
humiture = 17
def setup():
	print('Setting up, please wait...')
def loop():
        humidity, temperature = DHT.read_retry(Sensor, humiture)
        if humidity is not None and temperature is not None:
            jsonData = []
            measurement = {
                    'time': DateTime.isoformat(DateTime.now(pytz.timezone('Europe/Berlin'))),
                    'type':"c8y_demo",
					'source':{"id":"4420437"} ,
                    'Messung_der_Temperatur': {"T":{
                    "value": temperature, "unit":"C"}},
                    "Messung_der_Luftfeuchte":{"%":{
                    "value": humidity, "unit":"%"}}
            }
            jsonData.append(measurement)
            pprint(jsonData)	
            print('--------------------------------')
            global new
            new = json.dumps(measurement)
            with open('test.json', 'a') as jsonFile:
                for data in jsonData:
                    json.dump(data, jsonFile)
                    jsonFile.write('\n')
                    jsonData = []
            with open('test.json', 'rb') as jsonFile:
                for line in json_lines.reader(jsonFile):
                    jsonData.append(line)
        else:
            print("Failed to get reading!")
            
def send():
        global items
        items = new
        url = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        payload = items
        headers = {
                'authorization': "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
                'content-type': "application/json",
                'accept': "application/vnd.com.nsn.cumulocity.measurement+json",
                'cache-control': "no-cache",
        }
        response = requests.request("POST", url, data=payload, headers=headers)
        if response.status_code == 201:
            print(str(response.status_code) + "Measurement created")
            #Intervall der Messungen festlegen in Sekunden! Differenz s - 2,035sec
            time.sleep (57.965)
        else:
            print("fehlgeschlagen")
def check():
    try:
        urllib2.urlopen('http://216.58.192.142', timeout=1)
        return True
    except urllib2.URLError as err: 
        return False   
def sendagain():
    if check() is True:
        send()
    else:
        print("Senden fehlgeschlagen da keine Internetverbindung besteht!")
        time.sleep( 10 )
        sendagain()
def destroy():
	GPIO.cleanup()
def main():
    if __name__ == "__main__":
            setup()
            i = 1
            while i > -1:
                try:
                    loop()
                    check()
                    if check() is True:
                        send()
                    else:
                        print("Keine Internetverbindung vorhanden")
                        thread.start_new_thread(sendagain,  ())
                        time.sleep ( 57.965 )
                except KeyboardInterrupt:
                    print("Exit")
main()

Code: Alles auswählen

def mache_messung():
      # ... tu was um messwert zu ermitteln, und den zeitstempel
      return messwert, zeitstempel      
      
def verarbeite_messwert(messwert):
      # mach was mit dem messwert
      
      
def arbeite():
      while True:
             messwert, zeitstempel = mache_messung()
             print(zeitstempel)
             verarbeite_messwert(messwert)

Code: Alles auswählen

#!/usr/bin/env python
import RPi.GPIO as GPIO
import Adafruit_DHT as DHT
from datetime import datetime as DateTime
import json
import time
import pytz
import urllib2
import requests
import thread
import json_lines
from pprint import pprint

def Einleitung():
	print('Setting up, please wait...')
def Messung_vornehmen():
    Sensor = 11
    humiture = 17
    humidity, temperature = DHT.read_retry(Sensor, humiture)
    timestamp = DateTime.isoformat(DateTime.now(pytz.timezone('Europe/Berlin')))
    return humidity,  temperature,  timestamp
def Messung_Dictionary(humidity,  temperature,  timestamp):
    measurement = {
                    'time': timestamp,
                    'type':"c8y_demo",
					'source':{"id":"4420437"} ,
                    'Messung_der_Temperatur': {"T":{
                    "value": temperature, "unit":"C"}},
                    "Messung_der_Luftfeuchte":{"%":{
                    "value": humidity, "unit":"%"}}
    }
    return measurement
def erstelle_Liste(measurement):
    jsonData = []
    jsonData.append(measurement)
    pprint(jsonData)
    print("---------------------------------")
    return jsonData
def schreibe_Textdatei(jsonData):
    with open('test.json', 'a') as jsonFile:
        for data in jsonData:
            json.dump(data, jsonFile)
            jsonFile.write('\n')
            jsonData = []
    with open('test.json', 'rb') as jsonFile:
        for line in json_lines.reader(jsonFile):
            jsonData.append(line)
def senden(measurement):
        try:
            items = json.dumps(measurement)
            url = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
            payload = items
            headers = {
                'authorization': "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
                'content-type': "application/json",
                'accept': "application/vnd.com.nsn.cumulocity.measurement+json",
                'cache-control': "no-cache",
            }
            response = requests.request("POST", url, data=payload, headers=headers)
            print(response.text)
            print(str(response.status_code))
        except requests.ConnectionError:
            print("Keine Internetverbindung vorhanden")
def arbeite():
    # Dieser Teil muss noch angepasst werden!!
    while True:
        humidity,  temperature,  zeitstempel = Messung_vornehmen()
        measurement = Messung_Dictionary(humidity,  temperature,  zeitstempel)
        jsonData = erstelle_Liste(measurement)
        schreibe_Textdatei(jsonData)
        senden(measurement)
arbeite()

Code: Alles auswählen

def schreibe_textdatei(json_data):
    with open('test.json', 'a') as json_file:
        json.dump(json_data, json_file)
        json_file.write('\n')

Code: Alles auswählen

import time
import threading
from Queue import Queue

def mess_loop(queue):
    while True:
        messung = messung_vornehmen()
        queue.put(messung)
        time.sleep(5)

def write_to_file(queue):
    while True:
        schreibe_textdatei(queue.get())

def main():
    queue = Queue()
    mess_thread = threading.Thread(target=mess_loop, args=(queue,))
    schreib_thread = threading.Thread(target=write_to_file, args=(queue,))
    mess_thread.start()
    schreib_thread.start()
    while True:
        time.sleep(1000)