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. image.png ↓ Un avis me demande de faire quelque chose à ce sujet. image.png

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

image.png *Fujita et al., 2018*.

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.

Voici le comportement du bot lorsqu'il est configuré pour surveiller deux dossiers スクリーンショット 2020-01-15 14.37.46.jpg

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

résultat Clipboard2.jpg

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 image.png

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

image.png

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

Recommended Posts

Un programme qui notifie l'état de fonctionnement des équipements de photographie d'usine entièrement automatiques
J'ai fait un bot mou qui m'informe de la température
[Python] Un programme qui compte le nombre de vallées
[Python] Un programme qui compare les positions des kangourous.
L'histoire de l'exportation d'un programme
[Python] Un programme qui fait pivoter le contenu de la liste vers la gauche
[Python] Un programme qui calcule le nombre de segments de chocolat qui remplissent les conditions
[Python] Un programme qui calcule le nombre de chaussettes jumelées
[Python] Un programme qui arrondit le score
Création d'un bot Slack qui confirme et notifie à AWS Lambda la date d'expiration d'un certificat SSL
[Python] Un programme qui calcule le nombre de mises à jour des enregistrements les plus élevés et les plus faibles
Programme pour rechercher la même image
Un programme shell qui affiche une séquence de Fibonacci
[Python] Un programme qui trouve le nombre d'étapes le plus court dans un jeu qui traverse les nuages
Programme qui résume les données csv de l’historique des transactions de l’action SBI Securities [Python3]
Une histoire qui réduit l'effort de fonctionnement / maintenance
Créez un BOT qui raccourcit l'URL Discord
#Une fonction qui renvoie le code de caractère d'une chaîne de caractères
Programme Shell qui devient aho par multiples de 3
LINE Bot qui vous informe des stocks d'intérêt
Générer cette forme du fond d'une bouteille pour animaux de compagnie
Une histoire qui a analysé la livraison de Nico Nama.
Script Python qui peut vérifier l'état du serveur à partir du navigateur
[Python] Un programme pour trouver le nombre de pommes et d'oranges qui peuvent être récoltées