Das deutsche Python-Forum

Hallo liebe Python Community,

ich arbeite an einem Projekt, bei dem ich eine Lösung mit Python finden möchte. Es geht darum, dass ich einen Impuls an einem digitalen Input auslese und diesen verarbeiten möchte. Das Signal sieht folgendermaßen aus:

Bild

Die Anzahl der 1er kann variieren, ist aber auch nicht wichtig. Der entscheidende Faktor ist der Wechsel von 0 zu 1 und wieder zurück. Dieser Wert lieg in einer Variable vor, die sich ca. 50-100x die Sekunde aktualisiert. Also eine Abfrage nach Polling Methode.

Meine Idee ist jetzt irgendwie diesen Wechsel zu ermitteln und sobald der Wechsel stattgefunden hat, eine Liste um einen Wert zu erweitern. Sobald ein bestimmtes Zeitintervall abgelaufen ist, wird die Länge der Liste ausgelesen und übergeben. Daraufhin wird die Liste geleert und kann nue befüllt werden, bis das nächste Zeitintervall abgelaufen ist.

Mein Problem liegt darin, dass ich nicht genau weiß, wie ich diesen Wechsel aufnehmen kann. Hat hier jemand eine Idee? Unterstützung wäre super. Vielen Dank für jederlei Hilfe!

Gruß
mobby

Wenn man ein Funktion hat um den Eingang zu lesen, kann man sich eine Schreiben die die Werte in einer bestimmten Frequenz abfragt, und dann mit `itertools.groupby()` eine Funktion die nur noch die Wechsel liefert:

Code: Alles auswählen

#!/usr/bin/env python
import random
from itertools import groupby
from time import sleep


def get_value():
    return random.choice([0, 1])


def iter_values(frequency):
    while True:
        yield get_value()
        sleep(1.0 / frequency)


def iter_changes(frequency):
    for value, _ in groupby(iter_values(frequency)):
        yield value


def main():
    for value in iter_changes(200):
        print value


if __name__ == '__main__':
    main()

Hey BlackJack,

vielen Dank für diesen Antoß. Die Methode funktioniert perfekt. Ich habe das Ganze in mein Skript eingebaut und dabei ist Folgendes herausgekommen:

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import pifacedigitalio as p
from itertools import groupby
from time import sleep
import time 
from datetime import date
import MySQLdb
import thread

p.init()
buffer = []

def write():
	thread.start_new_thread(list_clear, ())

def mysql(timestamp, amount):
	db = MySQLdb.connect(host='localhost',
					   user='root',
					   passwd='',
					   db='')				   
	cursor = db.cursor()
	sql = "INSERT INTO water(timestamp, amount) \
	VALUES ('%i','%i')" %\
	(timestamp, amount)
	cursor.execute(sql)
        db.commit()
        db.close()

def get_value():
	return p.digital_read(0)
	
def list_value(add):
	buffer.append(add)
		
def iter_values(frequency):
	while True:
		yield get_value()
		sleep(1.0 / frequency)

def iter_changes(frequency):
	for value, _ in groupby(iter_values(frequency)):
		yield value

def list_clear():
	while True:
		timestamp = time.time()
		amount = len(buffer)
		mysql(timestamp, amount)
		del buffer[:]
		sleep(300.0)		
def main():
	for value in iter_changes(100):
		if value == 1:
			list_value(1)
			
if __name__ == '__main__':
	write()
	main()

Jetzt habe ich noch eine paar entscheidende Fragen:

1. Dieses Skript soll dazu dienen Sensoren auszulesen. Das soll über einen langen Zeitraum geschehen. Ist es möglich dieses Skritp über Tage/Wochen laufen zu lassen? Falls ja, wie resourcenfressend ist das? (Läuft alles auf einem Raspberry Pi) Falls nein, wie muss ich vorgehen, die oben zu sehenden Funktionen über einen längeren Zeitraum auszuführen?
2. Gibt es generell im Code verbesserungswürdiges? Falls ja immer her damit!

Danke!

@mobby: Einrückung ist per Konvention vier Leerzeichen pro Ebene.

Namen sollte man nur abkürzen wenn die Abkürzung allgemein bekannt ist. Einbuchstabige Namen möglichst nur verwenden wenn es tatsächlich einbuchstabige Namen sind, es also keine längere Schreibweise gibt, oder wenn der Gültigkeitsbereich stark begrenzt ist, also zum Beispiel in ``lambda``-Definitionen, „list comprehensions”, oder Generatorausdrücken.

Das `thread`-Modul sollte nicht mehr verwendet werden. Das ist durch das `threading`-Modul abgelöst worden.

`date` aus `datetime` wird importiert aber nicht verwendet.

Auf Modulebene sollte nur Code stehen der Konstanten, Funktionen, und Klassen definiert und keine globalen Datenstrukturen die verändert werden. Funktionen und Methoden sollten ausser Konstanten nur Werte verwenden die als Argumente übergeben wurden. Das betrifft `buffer`. Der Name ist auch etwas ungünstig, weil das der Name eines eingebauten Datentyps ist.

Funktionsnamen sind üblicherweise Tätigkeiten die beschreiben was die Funktion tut. `mysql()` ist keine Tätigkeit. `write()` schreibt nichts sondern startet einen Thread, `list_clear()` leert zwar eine Liste, schreibt aber auch einen Wert in eine Datenbank. Bei ``def list_value(add)`` sind irgendwie Teile des Funktionsnamens und des Arguments durcheinander gekommen, denn das Argument ist ja der Wert der hinzugefügt wird, würde also passender `value` heissen, während die Tätigkeit der Funktion mit `add` bezeichnet werden könnte.

Das zurücksetzen der Liste würde ich vor dem Eintragen in die Datenbank machen, weil das im Verhältnis zur Messfrequenz doch etwas länger dauern kann, und einem dann Messungen verloren gehen.

Werte sollte man nicht selber in Zeichenketten mit SQL-Anweisungen hineinformatieren. Das ist eine Sicherheitslücke (SQL-Injection) und kann ausserdem ineffizient sein, weil das DBMS jedes mal die SQL-Anweisung neu parsen und in einen Ablaufplan übersetzen muss, statt das nur einmal zu tun und das Ergebnis zu cachen.

Ich komme dann ungefähr bei so etwas heraus (ungetestet):

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time
from contextlib import closing
from itertools import groupby
from threading import Thread
from time import sleep
import MySQLdb
import pifacedigitalio as piface


def insert_into_database(timestamp, amount):
    with closing(
        MySQLdb.connect(host='localhost', user='root', passwd='', db='')
    ) as connection:
        connection.cursor().execute(
            'INSERT INTO water(timestamp, amount) VALUES (%s, %s)',
            (timestamp, amount)
        )
        connection.commit()


def write_leading_edge_count(leading_edges):
    while True:
        timestamp = time.time()
        leading_edge_count = len(leading_edges)
        del leading_edges[:]
        insert_into_database(timestamp, leading_edge_count)
        sleep(300)


def start_writer(leading_edges):
    thread = Thread(target=write_leading_edge_count, args=(leading_edges,))
    thread.setDaemon(True)
    thread.start()


def iter_values(frequency):
    while True:
        yield piface.digital_read(0)
        sleep(1.0 / frequency)


def iter_changes(frequency):
    for value, _ in groupby(iter_values(frequency)):
        yield value


def main():
    piface.init()
    leading_edges = list()
    start_writer(leading_edges)
    for value in iter_changes(100):
        if value == 1:
            leading_edges.append(1)


if __name__ == '__main__':
    main()

Wahrscheinlich würde ich den Code auch noch etwas anders auf Funktionen aufteilen, die Messung in einem eigenen Thread starten und nicht das Schreiben in die DB, und wohl eine Klasse für die Messung schreiben die einen Zähler hat, statt die Länge einer Liste zum Zählen zu verwenden.

@BlackJack: Wieder eine super Antwort, vielen Dank! Das ist echt sehr hilfreich.

Momentan bin ich schon relativ zufrieden mit dem Code. Eine Sache, die noch zu verbessern wäre ist die, dass ich den Eintrag in die Datenbank gerne zu einem Bestimmten Zeitpunkt abarbeiten möchte. Momentan wird das ja mit dem internen Zähler über sleep(Sekunden) gelöst. Das Problem ist aber, dass wenn ich das Skript um 12:03 starte, der Eintrag bei sleep(300) immer um xx:x3/xx:x8 gemacht wird. Dazu kommt noch, dass sich durch den Programmablauf, vor allem den Eintarg in die Datenbank, das Ganze über einen längeren Zeitraum verschiebt. Mein Wunsch wäre es, den Eintrag immer im 5 Minuten Takt analog zur Uhrzeit zu machen. Sprich 12:05, 12:10, 12:15 usw. Dazu habe ich mir Folgendes überlegt:

Code: Alles auswählen

def write_leading_edge_count(leading_edges):
    while True:
        timestamp = time.time()
        digit_one = str(timestamp)[8]
        digit_two = str(timestamp)[9]
        if digit_one == "0" and digit_two == "0":
            leading_edge_count = len(leading_edges)
            del leading_edges[:]
            insert_into_database(timestamp, leading_edge_count)

Ich bin mir aber nicht sicher, ob das eine saubere Lösung darstellt. Vorallem beunruhigt mich die Tatsache, dass die If-Bedingung ja eine Sekunde lang gilt. Ich möchte aber erreichen, dass der Inhalt der Schleife nur ein Mal ausgeführt wird. Gibt es da eine Möglichkeit das umzusetzen, ohne dass ich auch noch eine weitere Stelle hinter dem Komma überprüfen muss?

Was mich auch immer noch beschäftigt, ist die Frage, wie ich das Script sauber über längere Zeit ausführe? Natürlich starte es zu Beginn im Hintergrund auf meinem Unix System. Was passiert aber, sollte irgendein Fehler im Ablauf stattfinden und das Skript wird unterbrochen? Gibt es ein Programm oder eine Regel unter Unix Systemen mit denen ich sicherstellen kann, dass das Skript immer läuft. Also bei Absturtz sofort wieder gestartet wird? Und was mache ich in so einem Fall mit den verlorenen Daten? Würde sich da anbieten, anstatt in einer Liste, in einer externen Datei zu zählen, um so bei Skript-Abbruch die Werte zu sichern, damit bei Neustart die schon geschrieben Werte aus der Datei aufgenommen werden können?

Weiterhin vielen Dank für die ausführliche und nette Unterstützung!

@mobby: dass etwas exakt zu einer Zeit passiert, ist unmöglich, weil während Du auf die Uhr schaust, vergeht ja schon wieder etwas Zeit. Dein Ansatz hat mehrere Probleme: es ist ein busy-loop, das heißt, Du testest ständig, ob die Zeit stimmt, was in 99.999% der Fälle nicht der Fall ist. Das verbrät Strom und hindert andere Prozesse daran, sinnvolle Sachen zu tun. Den anderen Punkt hast Du schon angesprochen: innerhalb einer Sekunde könnte etliche male »insert_into_database« aufgerufen werden. Das Prüfen, ob eine Zahl durch eine andere Zahl teilbar ist, mit Stringvergleichen zu erledigen, ist zudem umständlich und nur bei wenigen Teilern (z.B. 100) möglich. Die Lösung ist, du berechnest, wie lange Du schlafen mußt, bis der nächste Zeitpunkt erreicht ist:

Code: Alles auswählen

INTERVALL = 300
def write_leading_edge_count(leading_edges):
    timestamp = int(time.time() / INTERVALL) * INTERVALL
    while True:
        timestamp += INTERVALL
        time.sleep(max(0, timestamp-time.time()))
        leading_edge_count = len(leading_edges)
        del leading_edges[:]
        insert_into_database(timestamp, leading_edge_count)

@Sirius3: Danke für die Hilfe. Das mit dem busy-loop hab ich mir auch schon überlegt gehabt, so ist es jetzt ja top gelöst. Der Ansatz mit der Sleep Funktion ist auch gut. Außerdem ist es garnicht so wichtig, in welcher Sekunde der Eintrag gemacht wird. Es geht vor allem darum, dass das Minutenintervall von 5/10/15 eingehalten wird, da ich die Einträge später über dieses Intervall wieder auslese. Also ob ein Eintrag um 12:05:00 oder um 12:05:01/59 gemacht wird ist eigentlich egal. Folglich bin ich jetzt mit dem Intervall save sobald ich dieses, für eine 5-Minuten Taktung, auf einen Wert von 300-359 setze.

Das deutsche Python-Forum

Auswertung digitales Signal mit Polling Methode

Auswertung digitales Signal mit Polling Methode

Re: Auswertung digitales Signal mit Polling Methode

Re: Auswertung digitales Signal mit Polling Methode

Re: Auswertung digitales Signal mit Polling Methode

Re: Auswertung digitales Signal mit Polling Methode

Re: Auswertung digitales Signal mit Polling Methode

Re: Auswertung digitales Signal mit Polling Methode