[PYTHON] Ermitteln Sie den Überflutungsstatus eines Flusses mit AI (ternäre Klassifizierung).
Einführung
AI überwacht die Überflutungssituation des Flusses, damit er nicht aussteigt oder sich nähert, wenn er überläuft Ich fragte mich, ob ich dich warnen könnte.
Motivation:
(Ich war frei in Corona ... ich meinte es nicht. Zuerst wollte ich zum ersten Mal Qiita posten.) Juli 2020 war die längste Regenzeit aller Zeiten.
Der Kamo-Fluss in Kyoto, den ich seit 30 Jahren beobachte, scheint jedes Mal überflutet zu sein, wenn es regnet, und die Stadt hat einen Zaun vorbereitet, damit er nicht zum Flussufer hinuntergeht, aber wenn Sie die Situation einzeln am elektronischen Schwarzen Brett anzeigen könnten, alle Ich dachte ich wäre glücklich. (Es sollte eine Katastrophenschutzkamera geben. Es tut mir leid, wenn ich es bereits getan habe. Ich denke, dass ich den Wasserstand kennen kann, daher kann es eine gute Kombination mit KI sein.) Klicken Sie hier, um die URL der Katastrophenschutzkamera von Kyoto City anzuzeigen
Fazit zuerst
Ich bemerkte, dass es schwierig war, Bilder zu sammeln, und dass unterwegs keine Ergebnisse erzielt wurden.
Ergänzung
Damit es von Menschen verwendet werden kann, die im Begriff sind, maschinelles Lernen zu beginnen Ich habe ein einfaches ternäres Klassifizierungsframework verwendet, daher ist der Code mehr als das Ergebnis Ich würde es begrüßen, wenn Sie sich darauf beziehen könnten.
Ich werde lernen, wie man Sätze und Codes schreibt und zeigt. Bitte beachten Sie, dass es nicht gut ist.
Über die Implementierung
Ich habe eine ternäre Klassifizierung mit Keras vorgenommen. Denken Sie nicht zu viel wie ein Signal ・ Sichere Situation (blau) ・ Ein Zustand, der Aufmerksamkeit erfordert (gelb) ・ Gefährlicher Zustand (rot) Ich dachte, es wäre genug, wenn es auftauchen würde.
・ Sichere Situation (blau)
・ Ein Zustand, der Aufmerksamkeit erfordert (gelb)
↑ wird von hier nachgedruckt.
・ Gefährlicher Zustand (rot)
Code
Der Code ist unten in Git aufgeführt.
https://github.com/nakamolinto/River_flood_detection
Code-Referenzquelle
Ich habe es basierend auf dem Code des APTOS-Wettbewerbs von kaggle erstellt. https://www.kaggle.com/c/aptos2019-blindness-detection/notebooks?sortBy=voteCount&group=everyone&pageSize=20&competitionId=14774
Für alle Lerndaten der Inhalte habe ich die Bilder verwendet, die auf SNS sind, wie z. B. Twitter, Was angezeigt wird, ist ein von mir aufgenommenes Bild ohne Quelle.
Implementierung
** Laden Sie die erforderlichen Bibliotheken **
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
** Daten lesen ** Dieses Mal habe ich es in drei Datenordner gelegt und gelesen.
#Sichere Situation (blau)
files=glob.glob("./images/ok/*")
dfok=pd.DataFrame(files,columns=["id_code"])
dfok["diagnosis"]=0
dfok.shape
#Zustand, der Aufmerksamkeit braucht(Gelb)
files=glob.glob("./images/bad/*")
dfbad=pd.DataFrame(files,columns=["id_code"])
dfbad["diagnosis"]=1
dfbad.shape
#Gefährlicher Zustand(rot)
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
** Aufteilen von Daten und Ändern der Größe usw. ** ** Die Bildgröße ist aufgrund von Spezifikationsproblemen auf 32 eingestellt, aber 256 * 256 ist besser, wenn Sie die GPU verwenden können. ** ** **
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
)
Überprüfen Sie die Anzahl der Daten in jeder Klasse.
** Klassendefinition **
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
** Datenauffüllung usw. **
#Chargengröße
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
#Lass uns DenseNet machen. Sie können verschiedene Modelle ausprobieren, indem Sie das Modell hier ändern.
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
** Modell bauen **
model = build_model()
model.summary()
kappa_metrics = Metrics()
** Modellzusammenfassung **
** Lernen Fühlen Sie sich frei, den Epock zu wechseln **
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)
** Überprüfen Sie das Inferenzergebnis **
test_df
↑ Das Obige ist ein Ergebnis, das überhaupt nicht vorhergesagt werden kann. .. ..
** Letzter Ausgabeteil **
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("Das Wasser der heutigen Flüsse ist sicher.")
image=cv2.imread(i[1],1)
plt.title("safe")
# plt.title(i[1])
plt.imshow(image)
elif i[0] ==1 :
print("Bitte beachten Sie, dass der Wasserstand des Flusses steigt.")
image=cv2.imread(i[1],1)
# plt.title(i[1])
plt.title("be careful")
plt.imshow(image)
else :
print("Der Fluss überschwemmt. Verlasse niemals den Fluss")
image=cv2.imread(i[1],1)
# plt.title(i[1])
plt.title("Do NOT enter")
plt.imshow(image)
** Ausgabebeispiel **
Ich gebe es vorerst aus.
Impressionen
Es ist leicht, den sicheren Zustand und den gefährlichen Zustand zu verstehen, aber es gibt nicht viele Bilder des Warnzustands. Es war schwer zu finden, ich gab den Weg auf und tat es mit Aya. Wenn Sie es richtig machen wollen, ist es schneller, es selbst zu nehmen. Ich konnte die GPU auch nicht verwenden und reduzierte die Bildgröße auf 32, aber wenn ich mit 224 * 224 studiere, kann ich möglicherweise eine richtige Schlussfolgerung ziehen.
Ich habe bereits die Überflutung des Flusses mit Sensoren usw. gesehen, daher ist es möglicherweise nicht erforderlich, KI zu verwenden. Dies kann nützlich sein, um festzustellen, ob sich nachts Personen am Flussufer befinden. Als nächstes werde ich es mit der Objekterkennung versuchen.
Wenn Sie Codefehler finden, teilen Sie uns dies bitte mit.
Ausrede
Wenn Sie genug schöne Bilder sammeln können, erhalten Sie gute Vorhersagen. Wenn jemand am Fluss beteiligt ist, probieren Sie es aus.
Referenz-URL
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 (Repost)
Der Code ist unten in Git aufgeführt. Wenn Sie Fragen zum Code haben, können Sie sich gerne an uns wenden https://github.com/nakamolinto/River_flood_detection
Schließlich
Ich mache auch Twitter. Bitte folgen Sie mir, wenn Sie möchten. https://twitter.com/pythonmachine