[PYTHON] Versuchen Sie, Merkmale von Sensordaten mit CNN zu extrahieren
Ich habe versucht, mit Sensordaten (1D-Vektor) zu falten.
Das Faltungs-Neuronale Netzwerk wird im Allgemeinen in der Bildverarbeitung verwendet, aber diesmal habe ich es mit einem eindimensionalen Vektor versucht, wie er in Sensordaten zu sehen ist. Der Punkt ist das gleiche 3D (RGB, X, Y) wie das Bild, indem die Datenstruktur mithilfe der Umformung konvertiert wird. Ich denke, dass Faltung für die Erkennung von Anomalien effektiv ist, da sie Merkmale extrahieren kann, selbst wenn es nur wenige Lernparameter gibt.
1. Erstellen Sie Beispieldaten
Erstellen Sie eine Sin-Welle für einen Zyklus mit numpy, Fügen Sie mit np.random.rand () Pseudorauschen hinzu Erstellen Sie 100 Elemente mit einigen Variationen. 99 wurden zum Lernen verwendet und die restlichen 1 wurden zur Verifizierung verwendet.
data=[]
for i in range(100):
data.append([np.sin(np.pi * n /50)*(1+np.random.rand())for n in range(100)])
2. Erstellen Sie ein CNN-Modell
Erstellen Sie ein Trainingsmodell mit Chainers Convolution 2D. Die Struktur ist so, dass die gewundenen Daten auf die ursprünglichen Eingabedaten in der letzten Schicht zurückgesetzt werden. Dadurch wird die Feature-Extraktion als Auto-Encoder durchgeführt.
~~ Die Aktivierungsfunktion ist auf Tanh eingestellt, da bei ReLU die Daten in der Mitte von Convlution angezeigt werden. Ich dachte, es würde schlecht aussehen, weil es bei der Visualisierung keine negative Seite gab. ~~ (Ergänzung) Die Daten zur Visualisierung werden herausgenommen, ohne die Aktivierungsfunktion zu durchlaufen. Ich fand das richtiger.
Selbst mit ReLU hatten die Sin-Wave-Daten keinen Einfluss auf die Trainingsergebnisse.
class MyChain(chainer.Chain):
def __init__(self,n_out):
super(MyChain, self).__init__()
with self.init_scope():
self.l1 = L.Convolution2D(None,2, ksize=(1,4),stride=(1,1))
self.l2 = L.Convolution2D(None,2, ksize=(1,4),stride=(1,1))
self.l3 = L.Convolution2D(None,2, ksize=(1,4),stride=(1,1))
self.l4 = L.Linear(None, n_out)
def __Call__(self,x,y):
return F.mean_squared_error(self.fwd(x),y)
def fwd(self, x):
h1 = F.tanh(F.max_pooling_2d(self.l1(x),2))
h2 = F.tanh(F.max_pooling_2d(self.l2(h1),2))
h3 = F.tanh(F.max_pooling_2d(self.l3(h2),2))
h3 = h3.reshape(h3.shape[0],-1)
return self.l4(h3)
3. Lernen
Bei Verwendung von CNN können eindimensionale Vektordaten nicht gelesen werden. Daher werden die Trainingsdaten mit Reshape konvertiert. Auch die Lehrerdaten sind die ursprünglichen Vektordaten.
TrainData = np.array(data,dtype=np.float32).reshape(100,1,1,100)
x=chainer.Variable(TrainData[:99])
for epoch in range(201):
model.zerograds()
loss=model(x,x.reshape(99,100))
loss.backward()
optimizer.update()
4. Ergebnis
Erstens das Ergebnis der Wiederherstellung aus der Eingangswellenform.
Es kann normal wiederhergestellt werden.
Schauen wir uns als Referenz die Wellenform des Faltvorgangs an. Ich versuche, die Größe der Daten auf Pseudo-Weise anzupassen.
Ehrlich gesagt verstehe ich das nicht wirklich. Man kann sagen, dass Schicht 3 eine Wellenform mit eingegrenzten Merkmalen ist. Was sind die Eigenschaften der Sin-Welle selbst? Erfassen Sie die Form eines Berges? Ich habe mehrmals versucht zu lernen, aber jedes Mal ist die Wellenform anders. Ich finde es interessant.
5. Erweiterung der Anomalieerkennung
Ich habe in der Vergangenheit eine Anomalieerkennung mit AutoEncoder gesehen. Es ist eine Methode, um den Grad der Abnormalität auszudrücken, indem die Differenz zwischen der Eingabe und der wiederhergestellten Ausgabe verwendet wird. Ebenso habe ich anomale Daten erstellt und überprüft.
Der erste ist die Phasenverschiebung Ich habe versucht, den Eingabewert um 5 Punkte zu verschieben (5/100 Zyklus).
Die Vorhersage-Wellenform liegt nahe an der ursprünglichen Phase und die Linienform ist gezackt. Es scheint, dass dies leicht als Anomalie erkannt werden kann.
Als nächstes, wenn ein Punkt wie ein Dorn hervorsteht
Das ist auch gezackt, nicht wahr? Wenn es so gezackt ist, kann es zur Erkennung von Anomalien anhand von Merkmalen wie Unterschieden verwendet werden.
6. Fazit
Dies ist mein erster Beitrag. Über diese Seite lese ich Ihre Beiträge und studiere. Ich dachte, ich würde etwas zurückgeben, also beschloss ich, es zu posten. Ich hoffe, dieser Beitrag hilft Ihnen weiter.
(Ergänzung) Tatsächlich verwendeter Beispielcode
Umgebung Python 3.6.1 Anaconda 4.4.0 (64-bit) Chainer 2.0.2
import chainer
import chainer.functions as F
import chainer.links as L
import chainer.optimizers
import numpy as np
import matplotlib.pyplot as plt
class MyChain(chainer.Chain):
def __init__(self,n_out):
super(MyChain, self).__init__()
with self.init_scope():
self.l1 = L.Convolution2D(None,2, ksize=(1,4),stride=(1,1))
self.l2 = L.Convolution2D(None,2, ksize=(1,4),stride=(1,1))
self.l3 = L.Convolution2D(None,2, ksize=(1,4),stride=(1,1))
self.l4 = L.Linear(None, n_out)
def __call__(self,x,y):
return F.mean_squared_error(self.fwd(x),y)
def fwd(self, x):
h1 = F.tanh(F.max_pooling_2d(self.l1(x),2))
h2 = F.tanh(F.max_pooling_2d(self.l2(h1),2))
h3 = F.tanh(F.max_pooling_2d(self.l3(h2),2))
h3 = h3.reshape(h3.shape[0],-1)
return self.l4(h3)
def Layaer1(self, x):
return F.max_pooling_2d(self.l1(x),2)
def Layaer2(self, x):
h1=F.tanh(F.max_pooling_2d(self.l1(x),2))
return F.max_pooling_2d(self.l2(h1),2)
def Layaer3(self, x):
h1 = F.tanh(F.max_pooling_2d(self.l1(x),2))
h2 = F.tanh(F.max_pooling_2d(self.l2(h1),2))
return F.max_pooling_2d(self.l3(h2),2)
def CreatePlotData(arr,n1):
Buf1,Buf2=[],[]
for j in range(n1):
Buf1.append(0)
Buf2.append(0)
for i in range(arr.shape[1]):
for j in range(n1):
Buf1.append(arr[0][i].real)
Buf2.append(arr[1][i].real)
return np.array(Buf1,dtype=np.float32),np.array(Buf2,dtype=np.float32)
data=[]
for i in range(100):
data.append([np.sin(np.pi * n /50)*(1+np.random.rand())for n in range(100)])
model = MyChain(100)
optimizer = chainer.optimizers.Adam()
optimizer.setup(model)
TrainData = np.array(data,dtype=np.float32).reshape(100,1,1,100)
x=chainer.Variable(TrainData[:99])
ValidationData=TrainData[99].reshape(1,1,1,100)
PlotInput = ValidationData.reshape(100)
for epoch in range(201):
model.zerograds()
loss=model(x,x.reshape(99,100))
loss.backward()
optimizer.update()
if epoch%20==0:
Layer1Arr = np.array(model.Layaer1(ValidationData).data).reshape(2,-1)
Layer1Arr1,Layer1Arr2 = CreatePlotData(Layer1Arr,2)
Layer2Arr = np.array(model.Layaer2(ValidationData).data).reshape(2,-1)
Layer2Arr1,Layer2Arr2 = CreatePlotData(Layer2Arr,4)
Layer3Arr = np.array(model.Layaer3(ValidationData).data).reshape(2,-1)
Layer3Arr1,Layer3Arr2 = CreatePlotData(Layer3Arr,8)
plt.plot(PlotInput,label='Input')
plt.plot(Layer1Arr1,label='Lalyer1-1')
plt.plot(Layer1Arr2,label='Lalyer1-2')
plt.plot(Layer2Arr1,label='Lalyer2-1')
plt.plot(Layer2Arr2,label='Lalyer2-2')
plt.plot(Layer3Arr1,label='Lalyer3-1')
plt.plot(Layer3Arr2,label='Lalyer3-2')
plt.legend()
plt.savefig(str(epoch)+'-Epoch Convolution Graph.png')
plt.close()
predict = model.fwd(ValidationData)
predict=np.array(predict.data).reshape(100)
plt.plot(predict,label='Predict')
plt.plot(PlotInput,label='Input')
plt.legend()
plt.savefig(str(epoch)+'-Epoch Validation Graph.png')
plt.close()
ErrorPlot = [PlotInput[i+5]for i in range(len(PlotInput)-5)]
for i in range(5):
ErrorPlot.append(PlotInput[i])
predict = model.fwd(chainer.Variable(np.array(ErrorPlot,dtype=np.float32)).reshape(1,1,1,100))
predict=np.array(predict.data).reshape(100)
plt.plot(predict,label='Predict')
plt.plot(ErrorPlot,label='Error Input')
plt.legend()
plt.savefig('Shift Error Input Graph.png')
plt.close()
Rnd = np.random.randint(0,99)
ErrorPlot2=np.array(PlotInput)
ErrorPlot2[Rnd]=ErrorPlot2[Rnd]+3
predict = model.fwd(chainer.Variable(np.array(ErrorPlot2,dtype=np.float32)).reshape(1,1,1,100))
predict=np.array(predict.data).reshape(100)
plt.plot(predict,label='Predict')
plt.plot(ErrorPlot2,label='Error Input')
plt.legend()
plt.savefig('Spike Error Input Graph.png')
plt.close()
Recommended Posts