Abstandsmessung als Zustand ausgeben

[Rene89]

Sehr geehrte Forenteilnehmer!

Folgender Aufbau:
Ein Abstand-Sensor führt eine Entfernungsmessung durch und überträgt im Sekundenabstand via serieller Schnittstelle die Daten an einen RPi.
Somit ist eine Ermittlung möglich, ob sich ein Objekt im Empfangsbereich befindet (Messwert ist kleiner) oder nicht (Messwert ist größer).
Zum Testen werden die Daten zur Verarbeitung an ThingSpeak weiter gereicht.
(siehe bisherigen Code)

Code: Alles auswählen

import serial
import time
import thingspeak

channel_id = xxxxxxx
write_key = 'xxxxxxxxxxxxxxxxxx'
a=0
ser = serial.Serial('/dev/ttyAMA0',19200)
ser.flushInput()

channel=thingspeak.Channel(id=channel_id,write_key=write_key)
while 'true':
    a=ser.readline()
    time.sleep(0.1)
    print(a)
    response = channel.update({1:a})
    time.sleep(0.090)
   

Nun soll der RPi/Python nicht die einzelen Abstandswerte,
sondern lediglich den jeweiligen Zustand (Objekt im Emfangsbereich vorhanden=1 /nicht vorhanden=0)
resultierend aus den Abstands-Messwerten berechnen und ausgeben.


Hierzu hätte ich nun folgende Annahme bzw. Idee zur Umsetzung,

Annahme:
Der Zustand/Abstand, Objekt im Empfangsbereich "nicht vorhanden" (0) ist absolut, da der Abstand vorgegeben und unveränderlich ist.
Der Zustand/Abstand, Objekt im Empfangsbereich "vorhanden" (1) ist relativ, beinflusst durch Form und Größe können die Messwerte leicht variieren.


Idee:
Eine Messung soll alle 30 Sekunden erfolgen.
Nun wird aus den Abstands-Messwerten die in 30 Sekunden gesammelt wurden ein Mittelwert errechnet.
Basierend auf einen Vergleich des Mittelwertes wird der Zustand dargestellt.


Beispiel:
Wenn der Mittelwert aus 30 Abstands-Messwerten <=2(Meter) ergibt, Dann Wert=1 (Objekt im Emfangsbereich VORHANDEN)
Wenn der Mittelwert aus 30 Abstands-Messwerten >2(Meter) ergibt, Dann Wert=0 (Objekt im Emfangsbereich NICHT VORHANDEN)


Geht meine Idee soweit in die richtige Richtung?
Wie geht man das am Besten an?
Ich bin für jeden Hinweis sehr dankbar.


Vielen Dank und schöne Grüße!
Rene

[__deets__]

Hallo,

da ist eine Menge im argen, das erstmal aufgeraeumt gehoert.

Die while-Schleife mit dem 'true' funktioniert, aber das ist so falsch ausgedrueckt. Es gibt einen Unterschied zwischen True und 'true' - letzteres ist ein String, und jeder String der nicht leer ist, wird als wahr ausgewertet. Du koenntest also auch

Code: Alles auswählen

while 'furzkanone':

schreiben, und alles wuerde genauso funktionieren. Ist aber beides verwirrend. Benutze True.

Das gestueckelte warten ist zum einen unnoetig, arithmetisch falsch, und konzeptionell auch ungenuegend. Unnoetig, weil readline wartet, bis die Daten da sind. Da muss nicht noch eine zehntel Sekunde gewartet werden.

Arithmetisch falsch, weil du von 1 Sekunde sprichst, und wahrscheinlich 0.1 + 0.9 rechnen willst. Aber stattdessen 0.09 benutzt. Du wartest also etwa 0.2 Sekunden.

Und konzeptionell ungenuegend, weil das warten zum einen nicht praezise ist - sleep(x) schlaeft immer garantiert MINDESTENS x Sekunden, aber auch gerne ein bisschen mehr, weil das Betriebssystem dich wieder drannehmen muss, und das ist nicht 100% deterministisch. Vor allem aber verbrauchen die anderen Instruktionen (insbesondere das readline und versenden) ebenfalls Zeit. Selbst wenn das sleep in der Summe 100% 1 Sekunde betruege, die Gesamtzeit der Schleife ist 1 + x, und damit driftet das weg.

Stattdessen geht man so vor:

Code: Alles auswählen

period = 1.0
timestamp = time.time()
while True:
    tuwas()
    timestamp += period
    time.sleep(timestamp - time.time())

Dadurch berechnet man immer den korrekten naechsten Zeitpunkt und wartet dann nur zu.

Und jetzt zu deinem eigentlichen Problem: ob eine solche Mittelung notwendig ist, haengt von der Qualitaet der Daten ab. Wenn die schon ausreichend gut ist, dann kann man sich die auch sparen. Ich persoenlich wuerde wenn auch nicht den Durchschnitt berechnen, sondern einen einfachen Tiefpass benutzen:

Code: Alles auswählen

v = v + (messwert - v) * filter_koeffizient

Vor allem aber fuehrt man bei solchen Aufgaben eigentlich immer eine Hysterese ein. Das System kippt von "unbelegt" zu "belegt" bei Werten > 2m. Zurueck zu "unbelegt" kippt es dann aber erst, wenn der Wert < 1.8m ist. Dadurch vermeidet man flattern des Zustands im Grenzbereich.

[Rene89]

Hallo!

Zuerst einmal Vielen Dank für die ausführlichen Informationen.
Ich habe den Code soweit mal aufgeräumt und Deinen Vorschlag übernommen.

Code: Alles auswählen

import serial
import time
import thingspeak

channel_id = xxxxxxxx
write_key = 'xxxxxxxx'
a=0
period = 30.0
ser = serial.Serial('/dev/ttyAMA0',19200)
ser.flushInput()

timestamp = time.time()

channel=thingspeak.Channel(id=channel_id,write_key=write_key)
while ser > 0:
	timestamp += period
	time.sleep(timestamp - time.time())
	a=ser.readline()
	print(timestamp)
	print(a)

.)Beim Testen ist mir jedoch eine weiterer Fehler bewusst geworden.
Wenn ich die Dauer (period) auf 30 Sekunden einstelle, wird wie gewünscht alle 30 Sekunden ein Wert ausgegeben.
Da die Seriellen Daten/Sensorwerte jedoch im Sekundentakt ankommen, wird in diesem Fall dann nicht mehr der aktulle Sensor-Wert ausgegeben, sondern immer einer der schon längst in der Vergangenheit liegt.
Alle (vergangenen) Werte werden der Reihe nach abgearbeitet.
Je länger die Messung aktiv ist desto länger dauert es, bis die alten Werte abgearbeitet -und die aktuellen Werte ausgegeben werden
Dies ist in meinem Fall natürlich nicht gewünscht, es soll immer der akutell vorherrschende Wert/Zustand angzeigt werden und nicht der,
der vor x Minuten aktuell war.
Wie könnte man das lösen?

.)Ich habe bereits über eine Art "Trägheit" sinniert. Dein Vorschlag mit der "Hysterese" trifft daher voll ins Schwarze.
Kannst du eventuell die Formel zur Tiefpassberechnung näher erklären, bzw. wie sollte ich den "filter_koeffizient" annehmen?


Besten Dank!

[__blackjack__]

@Rene89: Den gezeigten Code hast Du nicht getestet, denn der läuft nicht weil der Vergleich ob ein `Serial`-Objekt grösser als Null ist, zu einer Ausnahme führt. Das macht ja auch gar keinen Sinn.

Mit so schlechten Namen wie `a` sollte man gar nicht erst anfangen. Und auch kryptische Abkürzungen wie `ser` sind nicht gut. Wenn das einen sinnvollen Namen hätte, wäre vielleicht beim schreiben des Codes aufgefallen, dass ein Vergleich grösser Null keinen Sinn macht.

Die 0 die `a` am Anfang zugewiesen wird, wird nirgends verwendet. Man sollte auch nicht im gleichen Namensraum an den gleichen Namen sowohl Zahlen als auch Zeichenketten binden. Das verwirrt den Leser.

Es wäre sauberer wenn man dafür sorgt, dass die serielle Verbindung auch wieder geschlossen wird. `Serial`-Objekte sind Kontextmanager, können also mit ``with`` verwendet werden.

Bei `serial` sollte man die Attribute und Methoden die sich nicht an die Namenskonventionen halten, nicht mehr verwenden. `flushInput` heisst jetzt `reset_input_buffer()`.

Anstelle von `time.time()` sollte man `time.monotonic()` verwenden, weil das garantiert immer nur vorwärts läuft.

`channel` und alles was damit zusammenhängt wird nirgends verwendet.

Überarbeitet und ungetestet:

Code: Alles auswählen

#!/usr/bin/env python3
import time

import serial


def main():
    period = 30
    with serial.Serial("/dev/ttyAMA0", 19200) as sensor_connection:
        sensor_connection.reset_input_buffer()
        timestamp = time.monotonic()
        while True:
            timestamp += period
            time.sleep(timestamp - time.monotonic())
            line = sensor_connection.readline()
            print(timestamp)
            print(line)


if __name__ == "__main__":
    main()

[__deets__]

__blackjack__ hat Recht, ich haette monotonic benutzen sollen.

Die Frage nach Koeffizienten: 1 wuerde bedeuten eine neue Messung wuerde vollkommen ungefiltert uebernommen. 0 wuerde bedeuten eine neue Messung haette keinerlei Einfluss. Der Koeffizient liegt also sinnvollerweise irgendwo dazwischen. Ein Wert von zB 0.1 oder 0.01.

Und ich weiss nicht, was du mit 30 Sekunden erreichen willst - aber nur weil irgendetwas nur alle 30 Sekunden an dem Wert interessiert ist, heisst das ja nicht, dass man den auch nur alle 30 Sekunden erfassen sollte. Am besten erfasst man den so schnell wie es geht, eben genau um nicht einen solchen Wertestau zu bekommen. ser.readline() wartet ja einfach bis genug Daten da sind, und ist hoffentlich selber zeitlich getaktet.

[Sirius3]

Da Du eh auf den nächsten Datensatz wartest, kannst Du auch gleich alle einlesen, und nur die benutzen, die nach einer bestimmten Zeit gelesen wurden:

Code: Alles auswählen

#!/usr/bin/env python3
import time
import serial

def main():
    period = 30
    with serial.Serial("/dev/ttyAMA0", 19200) as sensor_connection:
        sensor_connection.reset_input_buffer()
        timestamp = time.monotonic()
        for line in sensor_connection:
            if timestamp < time.monotonic():
                timestamp += period
                print(timestamp, line)

if __name__ == "__main__":
    main()

[Rene89]

@ __deets__


Hallo!

Frage 1.)
Bezüglich der Nutzung eines Tiefpasses bzw. die Berechnung,

Code: Alles auswählen

v = v + (messwert - v) * filter_koeffizient

ergibt diese bei mir "v=messwert", d.h der messwert wäre ungefiltert
Was verstehe ich da falsch bzw. was mache ich falsch?

Frage 2.)
Bezüglich Hysterese,

Vor allem aber fuehrt man bei solchen Aufgaben eigentlich immer eine Hysterese ein. Das System kippt von "unbelegt" zu "belegt" bei Werten > 2m. Zurueck zu "unbelegt" kippt es dann aber erst, wenn der Wert < 1.8m ist. Dadurch vermeidet man flattern des Zustands im Grenzbereich.

Kannst Du mir diesbezüglich noch ein paar Tipps zur Umsetzung geben?


Vielen Dank!

[__deets__]

Da du keinen Code zeigst, kann ich auch nicht sagen, was du falsch machst.

Und eine Hysterese implementiert man, indem man erstmal nach dem aktuellen Zustand verzweigt, und dann erst den schwellwert prüft.

[Rene89]

Code: Alles auswählen

import random


random.seed()
RandomDistance = random.uniform(1.5, 2.5)
RandomDistance = "{:.1f}".format(RandomDistance)
RandomDistance = float(RandomDistance)

v= (0.5*RandomDistance) / 0.5

print(RandomDistance)
print(v)

[__deets__]

Das ist nicht die Formel die ich gezeigt habe. Und ziemlich viel Unfug obendrauf. Ich weiß jetzt nicht, wozu der Zufall da ins spiel kommen soll. Aber seed ruft man nicht auf. Das passiert von alleine. Ein float zum String und dann gleich wieder zurück zu wandeln ist überflüssig. Und deine Berechnung halbiert und multipliziert danach mit zwei, ist also auch einfach eine Null-Op. Zusammengefasst erzeugst du also einen Zufallswert, und gibst den zweimal aus. Es ist also kein Wunder, dass da zwei mal der gleiche Wert bei rumkommt.

[Rene89]

Den Zufallswert erzeuge ich nur zum Testen, daher nebensächlich.

Wie gesagt geht es in erster Linie jetzt mal nur um die richtige Berechnung.

Wenn ich die Formel

v = v + (messwert - v) * filter_koeffizient

nach

v= (filter_koeffizient * messwert) / filter_koeffizient

umforme, komme ich eben auf eine "eine Null-Op"

[__deets__]

Die Umformung ist schlicht falsch. Und unnötig. Warum muss da was umgeformt werden? Was soll das Bringen?

Das hier ist ein kleines Programm, dass die Konzepte illustriert.

Code: Alles auswählen

 import math
import numpy as np
import matplotlib.pyplot as plt

FREQ = 3
AMPLITUDE = 100
OFFSET = 150
NOISE_FREQ = 100
NOISE_AMPLITUDE = 20


THRESHOLD = 150
H_HIGH = 200
H_LOW = 100

FILTER_C = 0.05

def scale(a):
    return a * AMPLITUDE * 2 - AMPLITUDE + OFFSET
    
a = np.linspace(0, math.pi * 2, 1000)
s = np.sin(a * FREQ) * AMPLITUDE + np.sin(a * NOISE_FREQ) * NOISE_AMPLITUDE + OFFSET
th = s > THRESHOLD
th = scale(th)

hy = []
f = [s[0]]
triggered = False
for v in s:
    if not triggered and v > H_HIGH or triggered and v < H_LOW:
        triggered = not triggered
    hy.append(triggered)
    
    f.append(f[-1] + (v - f[-1]) * FILTER_C)
    
hy = scale(np.array(hy))

plt.plot(s)
plt.plot(th)
plt.plot(hy)
plt.plot(f)
plt.plot(scale(np.array(f) > THRESHOLD))

plt.show()

Das Resultat sieht so aus:

Man erkennt, dass ohne hysterese ein rein durch einen Schwellwert gebildeter Output durch das Rauschen eine Weile lang instabil ist. Wahlweise durch hysterese oder filtern kann man dem beikommen.

Und an die üblichen Verdächtigen: natürlich gehört da mehr in Funktionen, und den Tiefpass kann man bestimmt auf numpisch formulieren. Aber iPad & illustrationszwecke führten zu diesem Ergebnis.

[__deets__]

Deine Umformung mal en Detail

v= (filter_koeffizient * messwert) / filter_koeffizient # * filter_koeffizient

v * filter_koeffizient = (filter_koeffizient * messwert) # / filter_koeffizient

v = messwert

Das ist also nicht richtig.