[PYTHON] Ein Programm, das den Betriebszustand von vollautomatischen Anlagenfotografieanlagen mitteilt

Die Geschichte der Erstellung eines Python-Programms, das den Betriebsstatus des Pflanzenfotografiegeräts per Slack benachrichtigt

Es stellte sich heraus, dass es so etwas war.

Das Aufnahmegerät stoppt mit einem Fehler. ↓ Es kommt eine Nachricht, in der ich gebeten werde, etwas dagegen zu unternehmen.

Hintergrund

* Pressemitteilung des vollautomatischen Pflanzenexpressionstyp-Analysesystems "RIPPS" *

Artikel: Miki Fujita, Takanari Tanabata, Kaoru Urano, Saya Kikuchi, Kazuo Shinozaki, "RIPPS: Ein Pflanzenphänotypisierungssystem zur quantitativen Bewertung des Wachstums unter kontrollierten Umweltstressbedingungen", Pflanzen- und Zellphysiologie, 10.1093 / pcp / pcy122 //academic.oup.com/pcp/advance-article/doi/10.1093/pcp/pcy122/5043525)

Automatisches Pflanzenbau- und Bildgebungsgerät, entwickelt vom CSRS, Institut für physikalische und chemische Forschung. Einzelheiten finden Sie in der Presse (Japanisch) oben oder in der Zeitung (Englisch, Open Access).

Es ist ein Gerät, das Fotos und Wachstumsdaten mit einer eingebauten Kamera im Laufe der Zeit auf einem PC speichert, während Pflanzen auf einem Förderband gedreht werden. Es scheint jedoch, dass es manchmal aufgrund eines Fehlers stoppt. Wir kennen die Details der Ursache nicht, aber es scheint, dass es verschiedene Ursachen gibt, wie z. B. Wassertropfen im Wasserversorgungstank aufgrund von Hardware-Ursachen wie Beschädigung von Teilen und Verschlechterung aufgrund von Haftung von Boden und Lösung. Da sich das Forschungsgebäude zwischen dem Raum, in dem sich der verantwortliche Forscher befindet, und dem Raum, in dem sich das Gerät mit dem Steuer-PC befindet, unterscheidet, war es außerdem schwierig, weiterhin zu überwachen, ob es normal funktioniert. Zuerst dachte ich, ich sollte die letzte Aktualisierungszeit des Speicherzielordners über das Netzwerk anzeigen, aber es gibt Ordner für jeden Kameratyp in einem Gerät und durch Multiplikation mit der Anzahl der Geräte. Angesichts des Anstiegs war dies eine sehr sehr schwierige Situation. Als ich als Benutzer des Bildgebungsgeräts über gemeinsame Forschung sprach, beschloss ich, es eilig zu machen, da dies die Erfassung meiner eigenen experimentellen Daten behindern würde.

Da festgestellt wurde, dass es schwierig ist, die Fehlerbehandlungsfunktion in der Steuerungssoftware zu implementieren (ich möchte die einmal abgeschlossene Hardwaresteuerungssoftware nicht berühren), überwache ich den Aktualisierungsstatus des Ordners von außen und benachrichtige Slack über den Status. Ich habe ein einfaches Programm gemacht. Memorandum und Informationsaustausch unten.

Idee

-python

• Geben Sie den zu überwachenden Ordner an
• Da möglicherweise Unterordner erstellt werden, durchsucht rekursiv der angegebene Ordner.
• Benachrichtigen Sie, wenn sich die Dateien im Ordner mit dem angegebenen Intervall dazwischen nicht unterscheiden.
• Beachten Sie, dass das Programm ungefähr einmal am Tag normal ausgeführt wird. -Die Art zu benachrichtigen ist Twitter \ * oder Slack \ * Ich wusste nicht, wie lange es dauern würde, die API auf Twitter zu registrieren, also entschied ich mich für Slack.
• Für die automatische Veröffentlichung von Python zu Slack suchte ich in Qiita nach eingehenden Webhooks und entschied mich für den häufig verwendeten Code.

Programm

1. Laden Sie die erforderlichen Bibliotheken und die automatische Installation

Als ich Python aus der Ferne laufen ließ, stellte ich fest, dass das Anweisen der Pip-Installation ~~~ tatsächlich eine große Hürde war, daher habe ich diesen Teil auch in das Skript aufgenommen.

from pip._internal import main as _main
import importlib

def _import(name, module, ver=None):
    try:
        globals()[name] = importlib.import_module(module)
    except ImportError:
        try:
            if ver is None:
                _main(['install', module])
            else:
                _main(['install', '{}=={}'.format(module, ver)])
            globals()[name] = importlib.import_module(module)
        except:
            print("can't import: {}".format(module))

_import('requests','requests')
_import('argparse','argparse')
_import('json','json')
_import('threading','threading')

Referenz https://vaaaaaanquish.hatenablog.com/entry/2018/12/02/210647 https://stackoverflow.com/questions/12332975/installing-python-module-within-code

Wenn beim Start keine Bibliothek vorhanden ist und ein Importfehler angezeigt wird, installieren Sie diese und versuchen Sie erneut zu importieren. Requests, Argparse, JSON, Threading wurden angegeben, da sie in vielen Fällen nicht standardmäßig installiert wurden.

2. Verwalten Sie Befehlszeilenargumente mit argparse

parser = argparse.ArgumentParser()
parser.add_argument('-i', '--interval',default=1)
parser.add_argument('-d', '--dir', nargs='+', help="pass abs directory pass as list", required = True)
parser.add_argument('-n', '--name',default="RIPPS-Überwachungsprogramm_Erste Ausgabe")

Intervall: Häufigkeit der Dateibestätigung (Zeit) dir: Monitorordner (absoluter Pfad), mehrere können angegeben werden Name: Slack Poster Name

Verwenden Sie wie unten

python monitor.py --dir PATH1 PATH2 --interval 1.5 --name RIPPS_monitoring_robot_1

3. Andere Hardcode-Variablen

SLACK_WEBHOOK = "https://hooks.slack.com/services/XXXXXXXXXXXXX" start = time.time()

4. Erster Scan des Überwachungsordners und Slack Posting

def initiation(path,nfiles):    
	message = "%Überwachung starten s. Derzeit im überwachten Ordner%Es gibt d Dateien.%Suchen Sie alle Stunden nach Updates." % (path,nfiles,args.interval) 
	ellapsed = int(time.time() - start)
	payload_dic = {
		"icon_emoji": ':heartpulse:',
		"text": message,
		"username": args.name + "_" + str(int(ellapsed/(60*60))),
		"channel": "#general", # #Auch benötigt
	}
	try:
		r = requests.post(SLACK_WEBHOOK, data=json.dumps(payload_dic))
	except requests.ConnectionError:
		print(requests.ConnectionError)
		print("Verbindung zum Durchhang konnte nicht hergestellt werden.")

befores = []

for i, path_to_watch in enumerate(args.dir):
	print(path_to_watch)
	assert os.path.isdir(path_to_watch) == True, print("%s is not a valid directory" % path_to_watch)
	if path_to_watch[-1] != "/":
		path_to_watch += "/**"
	else:
		path_to_watch += "**"
	before = dict ([(f, None) for f in glob.glob(path_to_watch,recursive=True)])
	initiation(path_to_watch,len(before))
	args.dir[i] = path_to_watch
	befores.append(before)

• Wenn Sie die Option zum Überwachen mehrerer Verzeichnisse einschließen, ist args.dir eine Liste. Verarbeiten Sie sie daher mit der for-Schleife. Ein einzelner Ordner ist ebenfalls verfügbar. --Wenn es sich nicht um ein Verzeichnis handelt, wird es mit einem Assertionsfehler zurückgewiesen. --Wenn recursive = True von glob.glob und \ * \ * am Ende des Pfads hinzugefügt wird, wird das Unterverzeichnis durchsucht (sollte), also fügen Sie \ * \ * am Ende des Eingabearguments hinzu. Um damit umgehen zu können, auch wenn Sie den Ordner zum Terminal oder zur Eingabeaufforderung ziehen und dort ablegen. Die Anweisung, ~ ~ am Ende des Pfades hinzuzufügen, ist ziemlich kompliziert, und der Benutzer kann sie löschen.
• Machen Sie die von glob.glob erhaltene Dateiliste (obwohl sie genau die Anzahl der Ordner enthält) zu einem Wörterbuchtyp und übergeben Sie die Länge des Wörterbuchs an die Initialisierungsfunktion.
• Mit der Initiierungsfunktion zum Slack-Kanal posten.
• Der Dateiname des Überwachungsordners im Ausgangszustand wird in der vorherigen Liste gespeichert und später für den Differenzvergleich verwendet.

Das Folgende ist das Verhalten des Bots, wenn er so eingestellt ist, dass er zwei Ordner überwacht

5. Ordnerüberwachung

def errorpostslack(path):
	error_message = "Überwachter Ordner(%s)Aber%Nicht länger als s Stunden aktualisiert" % (path,args.interval)  #Keine Update-Nachricht
	ellapsed = int(time.time() - start)
	payload_dic = {
		"icon_emoji": ':cry:',
		"text": error_message,
		"username": args.name + "_" + str(int(ellapsed/(60*60))),
		"channel": "#general", # #Auch benötigt
	}
	try:
		r = requests.post(SLACK_WEBHOOK, data=json.dumps(payload_dic))
	except requests.ConnectionError:
		print(requests.ConnectionError)
		print("Verbindung zum Durchhang konnte nicht hergestellt werden.")

while 1:
	time.sleep(float(args.interval)*60*60)
	for i, (before, path_to_watch) in enumerate(zip(befores,args.directory)):
	    after = dict ([(f, None) for f in glob.glob(path_to_watch,recursive=True)])
	    added = [f for f in after if not f in before]
	    removed = [f for f in before if not f in after]
	    if added:
	        print ("Added: ", ", ".join (added))
	        #goodpostslack(added)
	        pass
	    elif removed:
	        print ("Removed: ", ", ".join (removed))
	        pass
	    else:
	        errorpostslack(path_to_watch)
	    befores[i] = after

• Führen Sie die Schleife mit einem Intervall (Sekunden) aus, das auf dem eingestellten Intervall (Zeit) basiert.
• Die vorherige Liste und das Verzeichnis, die beim ersten Scan erhalten wurden, werden gleichzeitig mit zip ausgespuckt. Aufzählung macht hier keinen Sinn, aber es ist ein Rest der Fehlerbehebung. --Verwenden Sie die aktualisierten Vorgänger ab dem zweiten Mal
• Holen Sie sich die neueste glob.glob-Dateiliste für jede Schleife. Wenn sich die Anzahl der Dateien im Vergleich zu früher erhöht hat, werden sie hinzugefügt, und wenn sie verschwunden sind, werden sie entfernt. Wenn keine Änderung vorgenommen wird, sind sowohl hinzugefügt als auch entfernt leer, und die Verarbeitung wird ausgeführt. --added entfernt Ich brauche keinen Pass für beide, aber ich habe ihn belassen, als ich den Druck auskommentierte. --Wenn die Datei nicht aktualisiert wurde, rufen Sie die Funktion errorpostslack auf, um eine Fehlermeldung an slack zu senden.

Ergebnis

6. Überlebensbericht

Wenn das Aufnahmegerät ordnungsgemäß funktioniert und sich die Bilder weiterhin ansammeln, erfolgt keine Benachrichtigung. Andererseits ist es schwierig, die normale Benachrichtigung in jedem Intervall beizubehalten. Daher werden wir eine Funktion hinzufügen, um das Überleben einmal am Tag zu melden.

def dailynotice():
	message = "Es ist ein regelmäßiger Bericht einmal am Tag."
	ellapsed = int(time.time() - start)
	for i, (before, path_to_watch) in enumerate(zip(befores,args.dir)):
		nfiles = len(glob.glob(path_to_watch,recursive=True))
		message += "%Unter dem Ordner s%Es gibt d Ordner und Dateien" % (path_to_watch,nfiles)
	payload_dic = {
		"icon_emoji": ':smile:',
		"text": message,
		"username": args.name + "_" + str(int(ellapsed/(60*60))),
		"channel": "#general", # #Auch benötigt
	}
	try:
		r = requests.post(SLACK_WEBHOOK, data=json.dumps(payload_dic))
	except requests.ConnectionError:
		print(requests.ConnectionError)
		print("Verbindung zum Durchhang konnte nicht hergestellt werden.")	
    
while 1:
	dailynotice()
	time.sleep(24*60*60)

Ergebnis

7. Koexistenz von Fehlererkennung und Überlebensbericht

Da es zwei while-Schleifen gibt, fädeln Sie sie gleichzeitig ein.

pseudocode
def error_check():
Beobachtete Funktion zur Überprüfung der Ordneraktualisierung
def daily_check():
Überlebensbericht

t1 = Thread(target = error_check)
t2 = Thread(target = daily_check)
t1.start()
t2.start()

8. Schließlich

• Der Code wurde auf github hochgeladen. https://github.com/totti0223/ripps_utility
• Ich habe mich gefragt, ob ich die URL des eingehenden Webhooks von slack als Befehlszeilenargument verwenden soll, aber ich habe den Code hart gemacht, weil er den Benutzer verwirrt.
• Es gibt wahrscheinlich Mac-Autometer, freie Software und andere Methoden, aber es war schneller, sie in Python zu erstellen. Wenn Sie es wissen, lassen Sie es mich bitte wissen, unabhängig von Sprache oder Betriebssystemplattform (auch wenn es nicht über Slack ist). ――Obwohl das Symbol für jeden Überlebensbericht und Fehlerbericht geändert wird, ist es traurig, dass die Threads nicht zusammen reflektiert werden, wenn derselbe Mitwirkende kontinuierlich Beiträge verfasst. ..

• Es ist eine sorgfältige Maßnahme, die Zeit vom Programmstart am Ende des Posternamens (Benutzername) zusammenzuführen. Nicht schön. Selbst wenn der Postername identisch ist, werden die Symbole möglicherweise einzeln angezeigt, sodass ich die Regeln nicht verstehe.

	payload_dic = {
		"icon_emoji": ':heartpulse:',
		"text": message,
		"username": args.name + "_" + str(int(ellapsed/(60*60))),
		"channel": "#general", # #Auch benötigt
	}