机器学习实验（四）：用tensorflow实现卷积神经网络识别人类活动

声明：版权所有，转载请联系作者并注明出处  http://blog.csdn.net/u013719780?viewmode=contents

博主简介：风雪夜归子（英文名：Allen），机器学习算法攻城狮，喜爱钻研Meachine
Learning的黑科技，对Deep Learning和Artificial Intelligence充满兴趣，经常关注Kaggle数据挖掘竞赛平台，对数据、Machine Learning和Artificial Intelligence有兴趣的童鞋可以一起探讨哦，个人CSDN博客：http://blog.csdn.net/u013719780?viewmode=contents

在近几年，越来越多的用户在智能手机上安装加速度传感器等一些设备，这就为做一些应用需要收集相关的数据提供了方便。人类活动识别(human activity recognition (HAR))是其中的一个应用。对于HAR，有很多的方法可以去尝试，方法的performance很大程度上依赖于特征工程。传统的机器学习特征工程通常是手工完成（人肉工程），这需要拥有较好的专业领域知识，同时比较耗时间。神经网络特别是深度学习在object
recognition, machine translation, audio generation等取得了很大的成功，同样，深跌学习技术也可以应用到HAR上。
在本文中，我们将会看到如何将卷积神经网络技术应用到HAR问题上。

数据预处理

我们将会使用Wireless Sensor Data Mining (WISDM) lab发布的数据集Actitracker（http://www.cis.fordham.edu/wisdm/dataset.php） 这个数据集是在一个可以控制的实验环境中收集到的。数据集中包含6个活动类别，分别是jogging,
walking, ascending stairs, descending stairs, sitting and standing。 这个数据集关于activities（labels）分布如下图所示：



首先导入相应的库和函数reading, normalising and plotting数据集。

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy import stats
import tensorflow as tf

%matplotlib inline
plt.style.use('ggplot')

def read_data(file_path):
column_names = ['user-id','activity','timestamp', 'x-axis', 'y-axis', 'z-axis']
data = pd.read_csv(file_path,header = None, names = column_names)
return data

def feature_normalize(dataset):
mu = np.mean(dataset,axis = 0)
sigma = np.std(dataset,axis = 0)
return (dataset - mu)/sigma

def plot_axis(ax, x, y, title):
ax.plot(x, y)
ax.set_title(title)
ax.xaxis.set_visible(False)
ax.set_ylim([min(y) - np.std(y), max(y) + np.std(y)])
ax.set_xlim([min(x), max(x)])
ax.grid(True)

def plot_activity(activity,data):
fig, (ax0, ax1, ax2) = plt.subplots(nrows = 3, figsize = (15, 10), sharex = True)
plot_axis(ax0, data['timestamp'], data['x-axis'], 'x-axis')
plot_axis(ax1, data['timestamp'], data['y-axis'], 'y-axis')
plot_axis(ax2, data['timestamp'], data['z-axis'], 'z-axis')
plt.subplots_adjust(hspace=0.2)
fig.suptitle(activity)
plt.subplots_adjust(top=0.90)
plt.show()

首先，读取数据集，然后normalise特征x-axis、y-axis、z-axis。

dataset = read_data('/Users/youwei.tan/Desktop/WISDM_ar_v1.1/WISDM_ar_v1.1_raw.txt')
dataset['x-axis'] = feature_normalize(dataset['x-axis'])
dataset['y-axis'] = feature_normalize(dataset['y-axis'])
dataset['z-axis'] = feature_normalize(dataset['z-axis'])

接下来可视化x-axis、y-axis、z-axis与时间的关系图。

for activity in np.unique(dataset["activity"]):
subset = dataset[dataset["activity"] == activity][:180]
plot_activity(activity,subset)

数据已经处理好啦，现在需要将数据转变成卷积神经网络模型所需要的数据形式。具体实现直接看代码：

def windows(data, size):
start = 0
while start < data.count():
yield start, start + size
start += (size / 2)

def segment_signal(data,window_size = 90):
segments = np.empty((0,window_size,3))
labels = np.empty((0))
for (start, end) in windows(data["timestamp"], window_size):
x = data["x-axis"][start:end]
y = data["y-axis"][start:end]
z = data["z-axis"][start:end]
if(len(dataset["timestamp"][start:end]) == window_size):
segments = np.vstack([segments,np.dstack([x,y,z])])
labels = np.append(labels,stats.mode(data["activity"][start:end])[0][0])
return segments, labels

segments, labels = segment_signal(dataset)
labels = np.asarray(pd.get_dummies(labels), dtype = np.int8)
reshaped_segments = segments.reshape(len(segments), 1,90, 3)

现在的数据已经是我们所期待的数据形式了，为了后面做交叉验证，需要将数据集分割为训练集和测试集。

train_test_split = np.random.rand(len(reshaped_segments)) < 0.70
train_x = reshaped_segments[train_test_split]
train_y = labels[train_test_split]
test_x = reshaped_segments[~train_test_split]
test_y = labels[~train_test_split]

卷积神经网络模型

CNN模型的结构如下图所示：



下面直接上代码：

input_height = 1
input_width = 90
num_labels = 6
num_channels = 3

batch_size = 10
kernel_size = 60
depth = 60
num_hidden = 1000

learning_rate = 0.0001
training_epochs = 5

total_batchs = reshaped_segments.shape[0] // batch_size

def weight_variable(shape):
initial = tf.truncated_normal(shape, stddev = 0.1)
return tf.Variable(initial)

def bias_variable(shape):
initial = tf.constant(0.0, shape = shape)
return tf.Variable(initial)

def depthwise_conv2d(x, W):
return tf.nn.depthwise_conv2d(x,W, [1, 1, 1, 1], padding='VALID')

def apply_depthwise_conv(x,kernel_size,num_channels,depth):
weights = weight_variable([1, kernel_size, num_channels, depth])
biases = bias_variable([depth * num_channels])
return tf.nn.relu(tf.add(depthwise_conv2d(x, weights),biases))

def apply_max_pool(x,kernel_size,stride_size):
return tf.nn.max_pool(x, ksize=[1, 1, kernel_size, 1],
strides=[1, 1, stride_size, 1], padding='VALID')

X = tf.placeholder(tf.float32, shape=[None,input_height,input_width,num_channels])
Y = tf.placeholder(tf.float32, shape=[None,num_labels])

c = apply_depthwise_conv(X,kernel_size,num_channels,depth)
p = apply_max_pool(c,20,2)
c = apply_depthwise_conv(p,6,depth*num_channels,depth//10)

shape = c.get_shape().as_list()
c_flat = tf.reshape(c, [-1, shape[1] * shape[2] * shape[3]])

f_weights_l1 = weight_variable([shape[1] * shape[2] * depth * num_channels * (depth//10), num_hidden])
f_biases_l1 = bias_variable([num_hidden])
f = tf.nn.tanh(tf.add(tf.matmul(c_flat, f_weights_l1),f_biases_l1))

out_weights = weight_variable([num_hidden, num_labels])
out_biases = bias_variable([num_labels])
y_ = tf.nn.softmax(tf.matmul(f, out_weights) + out_biases)

loss = -tf.reduce_sum(Y * tf.log(y_))
optimizer = tf.train.GradientDescentOptimizer(learning_rate = learning_rate).minimize(loss)

correct_prediction = tf.equal(tf.argmax(y_,1), tf.argmax(Y,1))
accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))

cost_history = np.empty(shape=[1],dtype=float)

with tf.Session() as session:
tf.initialize_all_variables().run()
for epoch in range(training_epochs):
for b in range(total_batchs):
offset = (b * batch_size) % (train_y.shape[0] - batch_size)
batch_x = train_x[offset:(offset + batch_size), :, :, :]
batch_y = train_y[offset:(offset + batch_size), :]
_, c = session.run([optimizer, loss],feed_dict={X: batch_x, Y : batch_y})
cost_history = np.append(cost_history,c)
print "Epoch: ",epoch," Training Loss: ",c," Training Accuracy: ",
session.run(accuracy, feed_dict={X: train_x, Y: train_y})

print "Testing Accuracy:", session.run(accuracy, feed_dict={X: test_x, Y: test_y})