[PYTHON] Un programme qui notifie l'état de fonctionnement des équipements de photographie d'usine entièrement automatiques

L'histoire de la création d'un programme python qui notifie l'état de fonctionnement de l'appareil de photographie d'usine via slack

Il s'est avéré être quelque chose comme ça.

L'appareil de prise de vue s'arrête avec une erreur. ↓ Un avis me demande de faire quelque chose à ce sujet.

Contexte

* Communiqué de presse du système d'analyse de type d'expression de plante entièrement automatique "RIPPS" *

Articles: Miki Fujita, Takanari Tanabata, Kaoru Urano, Saya Kikuchi, Kazuo Shinozaki, «RIPPS: A Plant Phenotyping System for Quantitative Evaluation of Growth under Controlled Environmental Stress Conditions», Plant & Cell Physiology, 10.1093 / pcp / pcy122 //academic.oup.com/pcp/advance-article/doi/10.1093/pcp/pcy122/5043525)

Dispositif automatique de culture et d'imagerie des plantes développé par le CSRS, Institut de recherche physique et chimique. Pour plus de détails, voir la presse (japonais) ci-dessus ou l'article (anglais, libre accès).

C'est un appareil qui stocke des photos et des données de croissance sur un PC au fil du temps avec une caméra intégrée tout en faisant tourner les plantes sur un convoyeur à bande, mais il semble qu'il s'arrête parfois en raison d'une erreur. Nous ne connaissons pas les détails de la cause, mais il semble qu'il existe diverses causes telles que des gouttes d'eau dans le réservoir d'alimentation en eau dues à des causes matérielles telles que des dommages aux pièces et une détérioration due à l'adhérence du sol et de la solution. De plus, le bâtiment de recherche étant différent entre la pièce où réside le chercheur en charge et la pièce où se trouve l'appareil avec le PC de contrôle, il était difficile de continuer à surveiller s'il fonctionne normalement. Au début, j'ai pensé que je devais regarder la dernière heure de mise à jour du dossier de destination de sauvegarde via le réseau, mais il y a des dossiers pour chaque type de caméra dans un appareil, et en multipliant par le nombre d'appareils. Compte tenu de l'augmentation, c'était une situation très particulière. Quand je parlais de recherche conjointe en tant qu'utilisateur de l'appareil d'imagerie, j'ai décidé de le faire à la hâte car cela entraverait l'acquisition de mes propres données expérimentales.

Puisqu'il a été jugé qu'il était difficile d'implémenter la fonction de gestion des erreurs dans le logiciel de contrôle (je ne veux pas toucher au logiciel de contrôle matériel qui a été terminé une fois), je surveille l'état de mise à jour du dossier de l'extérieur et informe Slack de l'état. J'ai fait un programme simple à faire. Mémorandum et partage d'informations ci-dessous.

idée

-python -Spécifiez le dossier à surveiller -Comme des sous-dossiers peuvent être créés, des recherches récursives dans le dossier spécifié. -Notifiez s'il n'y a pas de différence dans les fichiers dans le dossier avec l'intervalle spécifié entre les deux. -Notez que le programme fonctionne normalement environ une fois par jour. -Le moyen de notifier est twitter \ * ou slack \ * Je ne savais pas combien de temps il faudrait pour enregistrer l'API sur Twitter, j'ai donc choisi slack. --Pour la publication automatique de python vers slack, j'ai décidé d'utiliser le webhook entrant de qiita et d'autres codes couramment utilisés.

programme

1. Charger les bibliothèques requises et l'installation automatique

Quand j'ai fait exécuter python à distance, j'ai trouvé que l'instruction de pip install ~~~ était en fait un obstacle de taille, j'ai donc inclus cette partie dans le script également.

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')

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

Si la bibliothèque n'existe pas au démarrage et que vous obtenez une erreur d'importation, installez-la et réessayez d'importer. Les requêtes, argparse, json, threading ont été spécifiées car elles n'étaient pas installées en standard dans de nombreux cas.

2. Gérez les arguments de ligne de commande avec 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="Programme de surveillance RIPPS_Premier numéro")

intervalle: fréquence de confirmation du fichier (heure) dir: dossier du moniteur (chemin absolu), plusieurs peuvent être spécifiés nom: nom de l'affiche slack

Utilisez comme ci-dessous

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

3. Autres variables de code en dur

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

4. Analyse initiale du dossier de surveillance et publication en attente

def initiation(path,nfiles):    
	message = "%Commencer la surveillance s. Actuellement dans le dossier surveillé%Il existe des fichiers d.%Vérifiez les mises à jour toutes les heures." % (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", # #Nécessaire également
	}
	try:
		r = requests.post(SLACK_WEBHOOK, data=json.dumps(payload_dic))
	except requests.ConnectionError:
		print(requests.ConnectionError)
		print("Impossible de se connecter à slack.")

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)

--Si vous incluez l'option pour surveiller plusieurs répertoires, args.dir sera une liste, alors traitez-la avec la boucle for. Un seul dossier est également disponible.

• S'il ne s'agit pas d'un répertoire, il sera rejeté avec une erreur d'assertion. --Si recursive = True de glob.glob et \ * \ * est ajouté à la fin du chemin, le sous-répertoire sera recherché (devrait), alors ajoutez \ * \ * à la fin de l'argument d'entrée. Pour pouvoir gérer même si vous tirez le dossier vers le terminal ou l'invite de commande et le déposez. L'instruction pour ajouter ~ ~ à la fin du chemin est assez compliquée et l'utilisateur peut la supprimer.
• Transformez la liste de fichiers obtenue par glob.glob (bien qu'elle contienne le nombre de dossiers pour être exact) dans un type de dictionnaire, et transmettez la longueur du dictionnaire à la fonction d'initialisation. --Post au canal mou avec la fonction d'initiation.
• Le nom de fichier du dossier de surveillance à l'état initial est stocké dans la liste précédente et utilisé ultérieurement pour la comparaison des différences.

Voici le comportement du bot lorsqu'il est configuré pour surveiller deux dossiers

5. Surveillance des dossiers

def errorpostslack(path):
	error_message = "Dossier surveillé(%s)Mais%Non mis à jour depuis plus de s heures" % (path,args.interval)  #Aucun message de mise à jour
	ellapsed = int(time.time() - start)
	payload_dic = {
		"icon_emoji": ':cry:',
		"text": error_message,
		"username": args.name + "_" + str(int(ellapsed/(60*60))),
		"channel": "#general", # #Nécessaire également
	}
	try:
		r = requests.post(SLACK_WEBHOOK, data=json.dumps(payload_dic))
	except requests.ConnectionError:
		print(requests.ConnectionError)
		print("Impossible de se connecter à slack.")

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

--Exécute la boucle avec un intervalle (secondes) basé sur l'intervalle défini (temps).

• La liste précédente et le répertoire obtenus lors de l'analyse initiale sont crachés en même temps avec zip. énumérer n'a pas de sens ici, mais c'est un reste de correction de bogue. --Utilisez les mises à jour précédentes à partir de la deuxième fois
• Obtenez la dernière liste de fichiers glob.glob pour chaque boucle. Par rapport à avant, si le nombre de fichiers a augmenté, il sera ajouté, et s'il a disparu, il sera supprimé. S'il n'y a pas de changement, les éléments ajoutés et supprimés sont vides et le traitement est exécuté.
• ajouté supprimé Je n'ai pas besoin d'un laissez-passer pour les deux, mais je l'ai laissé pour quand j'ai commenté l'impression. --Si le fichier n'a pas été mis à jour, appelez la fonction errorpostslack pour publier un message d'erreur dans slack.

résultat

6. Rapport de survie

Si l'appareil de prise de vue fonctionne correctement et que les images continuent de s'accumuler, il n'y a pas de notification. D'un autre côté, il est difficile de conserver la notification normale à chaque intervalle. Par conséquent, nous ajouterons une fonction pour signaler la survie une fois par jour.

def dailynotice():
	message = "C'est un rapport régulier une fois par jour."
	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 += "%Sous le dossier s%Il y a d dossiers et fichiers" % (path_to_watch,nfiles)
	payload_dic = {
		"icon_emoji": ':smile:',
		"text": message,
		"username": args.name + "_" + str(int(ellapsed/(60*60))),
		"channel": "#general", # #Nécessaire également
	}
	try:
		r = requests.post(SLACK_WEBHOOK, data=json.dumps(payload_dic))
	except requests.ConnectionError:
		print(requests.ConnectionError)
		print("Impossible de se connecter à slack.")	
    
while 1:
	dailynotice()
	time.sleep(24*60*60)

résultat

7. Coexistence de la détection des erreurs et du rapport de survie

Puisqu'il y a deux boucles while, enfilez-les en même temps.

pseudocode
def error_check():
Fonction de vérification de la mise à jour du dossier surveillé
def daily_check():
Rapport de survie

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

8. Enfin

• Le code a été téléchargé sur github. https://github.com/totti0223/ripps_utility
• Je me demandais si je devais utiliser l'url du webhook entrant de slack comme argument de ligne de commande, mais j'ai fait du code dur car cela déroute l'utilisateur.
• Il existe probablement des automètres mac, des logiciels libres et d'autres méthodes, mais il était plus rapide de les faire en python. Si vous le connaissez, faites-le moi savoir quelle que soit la langue ou la plate-forme OS (même si ce n'est pas via slack). «Même si l'icône est modifiée pour chaque rapport de survie et rapport d'erreur, lorsque le même contributeur publie en continu, il est triste que les fils ne soient pas reflétés ensemble. ..

• C'est une mesure minutieuse à gérer en fusionnant le temps de démarrage du programme à la fin du nom de l'affiche (nom d'utilisateur). Pas belle. Même si le nom de l'affiche est le même, les icônes peuvent être affichées une par une, donc je ne comprends pas les règles.

	payload_dic = {
		"icon_emoji": ':heartpulse:',
		"text": message,
		"username": args.name + "_" + str(int(ellapsed/(60*60))),
		"channel": "#general", # #Nécessaire également
	}