[PYTHON] Détection des crues des rivières avec IA (classification ternaire)
introduction
L'IA surveille la situation d'inondation de la rivière afin qu'elle ne descende pas ou ne s'approche pas si elle déborde Je me suis demandé si je pouvais vous prévenir.
motivation:
(J'étais libre à Corona ... je ne veux pas dire. Tout d'abord, je voulais publier qiita pour la première fois.) Juillet 2020 a été la plus longue saison des pluies jamais vue.
La rivière Kamo à Kyoto, que je surveille depuis 30 ans, semble être inondée à chaque fois qu'il pleut, et la ville a préparé une clôture pour qu'elle ne descende pas au bord de la rivière, mais si vous pouviez afficher la situation une par une sur le tableau d'affichage électronique, tout le monde Je pensais que j'étais heureux. (Il devrait y avoir une caméra de prévention des catastrophes, je suis désolé si je l'ai déjà fait. Je pense que je peux connaître le niveau de l'eau, donc cela peut être une bonne combinaison avec l'IA) Cliquez ici pour l'URL de la caméra de prévention des catastrophes de la ville de Kyoto
Conclusion d'abord
J'ai remarqué qu'il était difficile de collecter des images et qu'aucun résultat n'a été obtenu en cours de route.
Supplément
Pour qu'il puisse être utilisé par des personnes sur le point de commencer l'apprentissage automatique J'ai utilisé un cadre de classification ternaire simple, donc le code est plus que le résultat J'apprécierais que vous y fassiez référence.
J'étudierai comment écrire et montrer des phrases et des codes. Veuillez noter que ce n'est pas bon.
À propos de la mise en œuvre
J'ai fait une classification ternaire en utilisant Keras. Ne pense pas trop, comme un signal ・ Situation de sécurité (bleu) ・ Un état qui nécessite une attention (jaune) ・ État dangereux (rouge) J'ai pensé que ce serait suffisant s'il apparaissait.
・ Situation de sécurité (bleu)
・ Un état qui nécessite une attention (jaune)
[↑ est réimprimé à partir d'ici. ](https://www.google.com/url?sa=i&url=https%3A%2F%2F4travel.jp%2Ftravelogue%2F11377392&psig=AOvVaw0v1E4ZtZd91XODHGo9-j-Q&ust=159710259100&scanner&source=159710259100&images
・ État dangereux (rouge)
code
Le code est répertorié dans git ci-dessous.
https://github.com/nakamolinto/River_flood_detection
Source de référence de code
Je l'ai créé sur la base du code du concours APTOS de kaggle. https://www.kaggle.com/c/aptos2019-blindness-detection/notebooks?sortBy=voteCount&group=everyone&pageSize=20&competitionId=14774
Pour toutes les données d'apprentissage du contenu, j'ai utilisé les images qui sont sur SNS comme Twitter, Ce qui est affiché est une image prise par moi-même qui n'a pas de source.
la mise en oeuvre
** Chargez les bibliothèques requises **
import json
import math
import os
import cv2
from PIL import Image
import numpy as np
from keras import layers
from keras.applications import DenseNet121
from keras.callbacks import Callback, ModelCheckpoint
from keras.preprocessing.image import ImageDataGenerator
from keras.models import Sequential
from keras.optimizers import Adam
import matplotlib.pyplot as plt
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.metrics import cohen_kappa_score, accuracy_score
import scipy
import tensorflow as tf
from tqdm import tqdm
import glob
%matplotlib inline
** Lecture des données ** Cette fois, je l'ai mis dans trois dossiers de données et je l'ai lu.
#Situation sûre (bleu)
files=glob.glob("./images/ok/*")
dfok=pd.DataFrame(files,columns=["id_code"])
dfok["diagnosis"]=0
dfok.shape
#État qui a besoin d'attention(Jaune)
files=glob.glob("./images/bad/*")
dfbad=pd.DataFrame(files,columns=["id_code"])
dfbad["diagnosis"]=1
dfbad.shape
#Condition dangereuse(rouge)
files=glob.glob("./images/ng/*")
dfng=pd.DataFrame(files,columns=["id_code"])
dfng["diagnosis"]=2
dfng.shape
dfall=pd.concat([dfok,dfbad,dfng])
dfall.shape
dfall['diagnosis'].hist()
dfall['diagnosis'].value_counts()
dfall.shape
** Fractionnement des données et redimensionnement, etc. ** ** La taille de l'image est définie sur 32 en raison de problèmes de spécifications, mais 256 * 256 est préférable si vous pouvez utiliser le GPU. ** **
from sklearn.model_selection import train_test_split
train_df, test_df=train_test_split(dfall,test_size=0.20)
train_df.shape
def get_pad_width(im, new_shape, is_rgb=True):
pad_diff = new_shape - im.shape[0], new_shape - im.shape[1]
t, b = math.floor(pad_diff[0]/2), math.ceil(pad_diff[0]/2)
l, r = math.floor(pad_diff[1]/2), math.ceil(pad_diff[1]/2)
if is_rgb:
pad_width = ((t,b), (l,r), (0, 0))
else:
pad_width = ((t,b), (l,r))
return pad_width
def preprocess_image(image_path, desired_size=32):
im = Image.open(image_path)
im = im.resize((desired_size, )*2, resample=Image.LANCZOS)
return im
N = train_df.shape[0]
x_train = np.empty((N, 32, 32, 3), dtype=np.uint8)
for i, image_id in enumerate(tqdm(train_df['id_code'])):
x_train[i, :, :, :] = preprocess_image(image_id)
x_train.shape
N = test_df.shape[0]
x_test = np.empty((N, 32, 32, 3), dtype=np.uint8)
for i, image_id in enumerate(tqdm(test_df['id_code'])):
x_test[i, :, :, :] = preprocess_image(image_id)
y_train = pd.get_dummies(train_df['diagnosis']).values
print(x_train.shape)
print(y_train.shape)
print(x_test.shape)
y_train_multi = np.empty(y_train.shape, dtype=y_train.dtype)
y_train_multi[:, 2] = y_train[:, 2]
for i in range(2):
y_train_multi[:, i] = np.logical_or(y_train[:, i], y_train_multi[:, i+1])
print("Original y_train:", y_train.sum(axis=0))
print("Multilabel version:", y_train_multi.sum(axis=0))
x_train, x_val, y_train, y_val = train_test_split(
x_train, y_train_multi,
test_size=0.15,
random_state=2019
)
Vérifiez le nombre de données dans chaque classe.
** Définition de classe **
class MixupGenerator():
def __init__(self, X_train, y_train, batch_size=32, alpha=0.2, shuffle=True, datagen=None):
self.X_train = X_train
self.y_train = y_train
self.batch_size = batch_size
self.alpha = alpha
self.shuffle = shuffle
self.sample_num = len(X_train)
self.datagen = datagen
def __call__(self):
while True:
indexes = self.__get_exploration_order()
itr_num = int(len(indexes) // (self.batch_size * 2))
for i in range(itr_num):
batch_ids = indexes[i * self.batch_size * 2:(i + 1) * self.batch_size * 2]
X, y = self.__data_generation(batch_ids)
yield X, y
def __get_exploration_order(self):
indexes = np.arange(self.sample_num)
if self.shuffle:
np.random.shuffle(indexes)
return indexes
def __data_generation(self, batch_ids):
_, h, w, c = self.X_train.shape
l = np.random.beta(self.alpha, self.alpha, self.batch_size)
X_l = l.reshape(self.batch_size, 1, 1, 1)
y_l = l.reshape(self.batch_size, 1)
X1 = self.X_train[batch_ids[:self.batch_size]]
X2 = self.X_train[batch_ids[self.batch_size:]]
X = X1 * X_l + X2 * (1 - X_l)
if self.datagen:
for i in range(self.batch_size):
X[i] = self.datagen.random_transform(X[i])
X[i] = self.datagen.standardize(X[i])
if isinstance(self.y_train, list):
y = []
for y_train_ in self.y_train:
y1 = y_train_[batch_ids[:self.batch_size]]
y2 = y_train_[batch_ids[self.batch_size:]]
y.append(y1 * y_l + y2 * (1 - y_l))
else:
y1 = self.y_train[batch_ids[:self.batch_size]]
y2 = self.y_train[batch_ids[self.batch_size:]]
y = y1 * y_l + y2 * (1 - y_l)
return X, y
** Remplissage de données, etc. **
#Taille du lot
BATCH_SIZE = 16
def create_datagen():
return ImageDataGenerator(
zoom_range=0.15, # set range for random zoom
# set mode for filling points outside the input boundaries
fill_mode='constant',
cval=0., # value used for fill_mode = "constant"
horizontal_flip=True, # randomly flip images
vertical_flip=True, # randomly flip images
)
# Using original generator
data_generator = create_datagen().flow(x_train, y_train, batch_size=BATCH_SIZE, seed=2019)
# Using Mixup
mixup_generator = MixupGenerator(x_train, y_train, batch_size=BATCH_SIZE, alpha=0.2, datagen=create_datagen())()
class Metrics(Callback):
def on_train_begin(self, logs={}):
self.val_kappas = []
def on_epoch_end(self, epoch, logs={}):
X_val, y_val = self.validation_data[:2]
y_val = y_val.sum(axis=1) - 1
y_pred = self.model.predict(X_val) > 0.5
y_pred = y_pred.astype(int).sum(axis=1) - 1
_val_kappa = cohen_kappa_score(
y_val,
y_pred,
weights='quadratic'
)
self.val_kappas.append(_val_kappa)
print(f"val_kappa: {_val_kappa:.4f}")
if _val_kappa == max(self.val_kappas):
print("Validation Kappa has improved. Saving model.")
self.model.save('model.h5')
return
#Faisons DenseNet. Vous pouvez essayer différents modèles en modifiant le modèle ici.
densenet = DenseNet121(
weights="imagenet",
include_top=False,
input_shape=(32,32,3)
)
def build_model():
model = Sequential()
model.add(densenet)
model.add(layers.GlobalAveragePooling2D())
model.add(layers.Dropout(0.5))
model.add(layers.Dense(3, activation='sigmoid'))
model.compile(
loss='binary_crossentropy',
optimizer=Adam(lr=0.00005),
metrics=['accuracy']
)
return model
** Construisez le modèle **
model = build_model()
model.summary()
kappa_metrics = Metrics()
** Résumé du modèle **
** Apprendre N'hésitez pas à changer l'epock **
history = model.fit_generator(
data_generator,
steps_per_epoch=x_train.shape[0] / BATCH_SIZE,
epochs=50,
validation_data=(x_val, y_val),
callbacks=[kappa_metrics])
** Infer **
model.load_weights('model.h5')
y_val_pred = model.predict(x_val)
def compute_score_inv(threshold):
y1 = y_val_pred > threshold
y1 = y1.astype(int).sum(axis=1) - 1
y2 = y_val.sum(axis=1) - 1
score = cohen_kappa_score(y1, y2, weights='quadratic')
return 1 - score
simplex = scipy.optimize.minimize(
compute_score_inv, 0.5, method='nelder-mead'
)
best_threshold = simplex['x'][0]
y_test = model.predict(x_test) > 0.5
y_test = y_test.astype(int).sum(axis=1) - 1
test_df['prediction'] = y_test
test_df.to_csv('kamogawa_result.csv',index=False)
** Vérifiez le résultat de l'inférence **
test_df
↑ Ce qui précède est un résultat qui ne peut pas du tout être prédit. .. ..
** Dernière partie de sortie **
prediction=test_df.prediction
id=test_df.id_code
%matplotlib inline
plt.figure(figsize=(16,12))
for num,i in enumerate(zip(prediction,id)):
plt.subplot(4,2,num+1)
if i[0] == 0 :
print("Les eaux des rivières d'aujourd'hui sont sûres.")
image=cv2.imread(i[1],1)
plt.title("safe")
# plt.title(i[1])
plt.imshow(image)
elif i[0] ==1 :
print("Veuillez noter que le niveau de l'eau de la rivière monte.")
image=cv2.imread(i[1],1)
# plt.title(i[1])
plt.title("be careful")
plt.imshow(image)
else :
print("La rivière est en crue. Ne descends jamais du bord de la rivière")
image=cv2.imread(i[1],1)
# plt.title(i[1])
plt.title("Do NOT enter")
plt.imshow(image)
** Exemple de sortie **
Je l'ai sorti pour le moment.
Impressions
Il est facile de comprendre l'état de sécurité et l'état dangereux, mais il n'y a pas beaucoup d'images de l'état de prudence, C'était difficile à trouver, j'ai abandonné le chemin et l'ai fait avec Aya. Si vous voulez le faire correctement, il est plus rapide de le faire vous-même. De plus, je n'ai pas pu utiliser le GPU et j'ai réduit la taille de l'image à 32, mais si j'étudie avec 224 * 224, je pourrai peut-être faire une déduction correcte.
J'ai déjà vu l'inondation de la rivière avec des capteurs etc., donc il n'est peut-être pas nécessaire d'utiliser l'IA, Cela peut être utile pour détecter s'il y a des personnes au bord de la rivière la nuit. Ensuite, je pense que je vais essayer avec la détection d'objets.
Si vous trouvez des erreurs de code, veuillez nous en informer.
excuse
Si vous pouvez collecter suffisamment de belles images, vous obtiendrez de bonnes prédictions. Si quelqu'un est impliqué dans la rivière, essayez-le.
URL de référence
https://qiita.com/yu4u/items/078054dfb5592cbb80cc
https://www.kaggle.com/c/aptos2019-blindness-detection/notebooks?sortBy=voteCount&group=everyone&pageSize=20&competitionId=14774
http://www.qsr.mlit.go.jp/useful/n-shiryo/kikaku/kenkyu/h30/04/4_03(18).pdf
Code (republication)
Le code est répertorié dans git ci-dessous. Si vous avez des questions sur le code, n'hésitez pas à nous contacter https://github.com/nakamolinto/River_flood_detection
finalement
Je fais aussi Twitter. Suivez-moi si vous le souhaitez. https://twitter.com/pythonmachine