raphael/siamese
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

initial commit, adds notes.md for instructions, adds data, adds python code

Browse Source
This commit is contained in:
Raphael Maenle 2021-07-22 22:24:50 +02:00
commit 25fa687f7f
17 changed files with 1173 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
.devcontainer/Dockerfile Normal file
View File

@ -0,0 +1,37 @@
FROM tensorflow/tensorflow:1.13.2-gpu
## Install updates and network tool
RUN apt-get update -y && apt-get upgrade -y && apt install net-tools -y
## Install basic functions
RUN apt-get install sudo -y
## Install git
RUN apt-get install git -y
## Install python requirements
COPY requirements.txt .
RUN pip install -r requirements.txt
## Create user and group
ARG HOST_USER_UID=1000
ARG HOST_USER_GID=1000
RUN groupadd -g $HOST_USER_GID containergroup
RUN useradd -m -l -u $HOST_USER_UID -g $HOST_USER_GID containeruser
## Passwordless sudo for user
RUN usermod -aG sudo containeruser
RUN echo "containeruser ALL=(root) NOPASSWD:ALL" > /etc/sudoers.d/containeruser && \
chmod 0440 /etc/sudoers.d/containeruser
## Activate User
USER containeruser
## Set working directory
WORKDIR /home/containeruser
## Workaround for vscode bug
ENV HOME=/home/containeruser
## Keep container running forever
CMD tail -f /dev/null

12
.devcontainer/devcontainer.json Normal file
View File

@ -0,0 +1,12 @@
{
"name": "siamese",
"dockerComposeFile": "docker-compose.yml",
"workspaceMount": "/workspace",
"workspaceFolder": "/workspace",
"service": "devcontainer",
"shutdownAction": "stopCompose",
"extensions": [
"ms-python.python",
"ms-azuretools.vscode-docker"
]
}

22
.devcontainer/docker-compose.yml Normal file
View File

@ -0,0 +1,22 @@
version: '2.3'
services:
devcontainer:
build:
context: ..
dockerfile: .devcontainer/Dockerfile
args:
HOST_USER_UID: 1000
HOST_USER_GID: 1000
network_mode: host
environment:
- DISPLAY=$DISPLAY
runtime: nvidia
volumes:
- ..:/workspace
- ~/.gitconfig:/home/containeruser/.gitconfig
- ~/.ssh:/home/containeruser/.ssh
command: sleep infinity

114
.gitignore vendored Normal file
View File

@ -0,0 +1,114 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
/env
# idea files
.idea/
# model checkpoint data
checkpoint
model_checkpoint
siamese_checkpoint

21
LICENSE Normal file
View File

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2018 aspamers
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

144
README.md Normal file
View File

@ -0,0 +1,144 @@
# Siamese Neural Network for Keras
This project provides a lightweight, easy to use and flexible siamese neural network module for use with the Keras
framework.
Siamese neural networks are used to generate embeddings that describe inter and extra class relationships.
This makes Siamese Networks like many other similarity learning algorithms suitable as a pre-training step for many
classification problems.
An example of the siamese network module being used to produce a noteworthy 99.85% validation performance on the MNIST
dataset with no data augmentation and minimal modification from the Keras example is provided.
## Installation
Create and activate a virtual environment for the project.
```sh
$ virtualenv env
$ source env/bin/activate
```
To install the module directly from GitHub:
```
$ pip install git+https://github.com/aspamers/siamese
```
The module will install keras and numpy but no back-end (like tensorflow). This is deliberate since it leaves the module
decoupled from any back-end and gives you a chance to install whatever backend you prefer.
To install tensorflow:
```
$ pip install tensorflow
```
To install tensorflow with gpu support:
```
$ pip install tensorflow-gpu
```
## To run examples
With the activated virtual environment with the installed python package run the following commands.
To run the mnist baseline example:
```
$ python mnist_example.py
```
To run the mnist siamese pretrained example:
```
$ python mnist_siamese_example.py
```
## Usage
For detailed usage examples please refer to the examples and unit test modules. If the instructions are not sufficient
feel free to make a request for improvements.
- Import the module
```python
from siamese import SiameseNetwork
```
- Load or generate some data.
```python
x_train = np.random.rand(100, 3)
y_train = np.random.randint(num_classes, size=100)
x_test = np.random.rand(30, 3)
y_test = np.random.randint(num_classes, size=30)
```
- Design a base model
```python
def create_base_model(input_shape):
model_input = Input(shape=input_shape)
embedding = Flatten()(model_input)
embedding = Dense(128)(embedding)
return Model(model_input, embedding)
```
- Design a head model
```python
def create_head_model(embedding_shape):
embedding_a = Input(shape=embedding_shape)
embedding_b = Input(shape=embedding_shape)
head = Concatenate()([embedding_a, embedding_b])
head = Dense(4)(head)
head = BatchNormalization()(head)
head = Activation(activation='sigmoid')(head)
head = Dense(1)(head)
head = BatchNormalization()(head)
head = Activation(activation='sigmoid')(head)
return Model([embedding_a, embedding_b], head)
```
- Create an instance of the SiameseNetwork class
```python
base_model = create_base_model(input_shape)
head_model = create_head_model(base_model.output_shape)
siamese_network = SiameseNetwork(base_model, head_model)
```
- Compile the model
```python
siamese_network.compile(loss='binary_crossentropy', optimizer=keras.optimizers.adam())
```
- Train the model
```python
siamese_network.fit(x_train, y_train,
validation_data=(x_test, y_test),
batch_size=64,
epochs=epochs)
```
## Development Environment
Create and activate a test virtual environment for the project.
```sh
$ virtualenv env
$ source env/bin/activate
```
Install requirements
```sh
$ pip install -r requirements.txt
```
Install the backend of your choice.
```
$ pip install tensorflow
```
Run tests
```sh
$ pytest tests/test_siamese.py
```
## Development container
To set up the vscode development container follow the instructions at the link provided:
https://github.com/aspamers/vscode-devcontainer
You will also need to install the nvidia docker gpu passthrough layer:
https://github.com/NVIDIA/nvidia-docker

1
data Symbolic link
View File

@ -0,0 +1 @@
/home/creation/files/data/

94
mnist_example.py Normal file
View File

@ -0,0 +1,94 @@
"""
This is a modified version of the Keras mnist example.
https://keras.io/examples/mnist_cnn/
Instead of using a fixed number of epochs this version continues to train
until the stop criteria is reached.
Model performance should be around 99.4% after training.
"""
from __future__ import print_function
import keras
from keras.datasets import mnist
from keras.layers import Conv2D, MaxPooling2D, BatchNormalization, Activation
from keras import backend as K
from keras.callbacks import ModelCheckpoint, EarlyStopping
from keras.models import Model
from keras.layers import Input, Flatten, Dense
batch_size = 128
num_classes = 10
epochs = 999999
# input image dimensions
img_rows, img_cols = 28, 28
# the data, split between train and test sets
(x_train, y_train), (x_test, y_test) = mnist.load_data()
if K.image_data_format() == 'channels_first':
x_train = x_train.reshape(x_train.shape[0], 1, img_rows, img_cols)
x_test = x_test.reshape(x_test.shape[0], 1, img_rows, img_cols)
input_shape = (1, img_rows, img_cols)
else:
x_train = x_train.reshape(x_train.shape[0], img_rows, img_cols, 1)
x_test = x_test.reshape(x_test.shape[0], img_rows, img_cols, 1)
input_shape = (img_rows, img_cols, 1)
x_train = x_train.astype('float32')
x_test = x_test.astype('float32')
x_train /= 255
x_test /= 255
y_train = keras.utils.to_categorical(y_train, num_classes)
y_test = keras.utils.to_categorical(y_test, num_classes)
def create_base_network(input_shape):
input = Input(shape=input_shape)
x = Conv2D(32, kernel_size=(3, 3),
input_shape=input_shape)(input)
x = BatchNormalization()(x)
x = Activation(activation='relu')(x)
x = MaxPooling2D(pool_size=(2, 2))(x)
x = Conv2D(64, kernel_size=(3, 3))(x)
x = BatchNormalization()(x)
x = Activation(activation='relu')(x)
x = MaxPooling2D(pool_size=(2, 2))(x)
x = Flatten()(x)
x = Dense(128)(x)
x = BatchNormalization()(x)
x = Activation(activation='relu')(x)
x = Dense(num_classes)(x)
x = BatchNormalization()(x)
x = Activation(activation='softmax')(x)
return Model(input, x)
model = create_base_network(input_shape)
model.compile(loss=keras.losses.categorical_crossentropy,
optimizer=keras.optimizers.adam(),
metrics=['accuracy'])
checkpoint_path = "./checkpoint"
callbacks = [
EarlyStopping(monitor='val_acc', patience=10, verbose=0),
ModelCheckpoint(checkpoint_path,
monitor='val_acc',
save_best_only=True,
verbose=0)
]
model.fit(x_train, y_train,
batch_size=batch_size,
epochs=epochs,
verbose=1,
callbacks=callbacks,
validation_data=(x_test, y_test))
model.load_weights(checkpoint_path)
score = model.evaluate(x_test, y_test, verbose=0)
print('Test loss:', score[0])
print('Test accuracy:', score[1])

299
mnist_siamese_example.py Normal file
View File

@ -0,0 +1,299 @@
"""
This is a modified version of the Keras mnist example.
https://keras.io/examples/mnist_cnn/
Instead of using a fixed number of epochs this version continues to train until a stop criteria is reached.
A siamese neural network is used to pre-train an embedding for the network. The resulting embedding is then extended
with a softmax output layer for categorical predictions.
Model performance should be around 99.84% after training. The resulting model is identical in structure to the one in
the example yet shows considerable improvement in relative error confirming that the embedding learned by the siamese
network is useful.
"""
from __future__ import print_function
import tensorflow.keras as keras
from tensorflow.keras.datasets import mnist
from tensorflow.keras.layers import Conv2D, MaxPooling2D, BatchNormalization, Activation, Concatenate
from tensorflow.keras import backend as K
from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Flatten, Dense
from siamese import SiameseNetwork
import os, math, numpy as np
from PIL import Image
import pdb
batch_size = 128
num_classes = 131
epochs = 999999
# input image dimensions
img_rows, img_cols = 28, 28
def createTrainingData():
base_dir = '../towards/data/fruits-360/Training/'
train_test_split = 0.7
no_of_files_in_each_class = 80
#Read all the folders in the directory
folder_list = os.listdir(base_dir)
print( len(folder_list), "categories found in the dataset")
#Declare training array
cat_list = []
x = []
names = []
y = []
y_label = 0
#Using just 5 images per category
for folder_name in folder_list:
files_list = os.listdir(os.path.join(base_dir, folder_name))
temp=[]
for file_name in files_list[:no_of_files_in_each_class]:
temp.append(len(x))
x.append(np.asarray(Image.open(os.path.join(base_dir, folder_name, file_name)).convert('RGB').resize((img_rows, img_cols))))
names.append(folder_name + "/" + file_name)
y.append(y_label)
y_label+=1
cat_list.append(temp)
cat_list = np.asarray(cat_list)
x = np.asarray(x)/255.0
y = np.asarray(y)
print('X, Y shape',x.shape, y.shape, cat_list.shape)
#Training Split
x_train, y_train, cat_train, x_val, y_val, cat_test = [], [], [], [], [], []
train_split = math.floor((train_test_split) * no_of_files_in_each_class)
test_split = math.floor((1-train_test_split) * no_of_files_in_each_class)
train_count = 0
test_count = 0
for i in range(len(x)-1):
if i % no_of_files_in_each_class == 0:
cat_train.append([])
cat_test.append([])
class_train_count = 1
class_test_count = 1
if i % math.floor(1/train_test_split) == 0 and class_test_count < test_split:
x_val.append(x[i])
y_val.append(y[i])
cat_test[-1].append(test_count)
test_count += 1
class_test_count += 1
elif class_train_count < train_split:
x_train.append(x[i])
y_train.append(y[i])
cat_train[-1].append(train_count)
train_count += 1
class_train_count += 1
x_val = np.array(x_val)
y_val = np.array(y_val)
x_train = np.array(x_train)
y_train = np.array(y_train)
cat_train = np.array(cat_train)
cat_test = np.array(cat_test)
print('X&Y shape of training data :',x_train.shape, 'and',
y_train.shape, cat_train.shape)
print('X&Y shape of testing data :' , x_val.shape, 'and',
y_val.shape, cat_test.shape)
return (x_train, y_train), (x_val, y_val), cat_train
# the data, split between train and test sets
# (x_train, y_train), (x_test, y_test) = mnist.load_data()
# channels = 1
(x_train, y_train), (x_test, y_test), cat_train = createTrainingData()
channels = 3
if K.image_data_format() == 'channels_first':
x_train = x_train.reshape(x_train.shape[0], channels, img_rows, img_cols)
x_test = x_test.reshape(x_test.shape[0], channels, img_rows, img_cols)
input_shape = (channels, img_rows, img_cols)
else:
x_train = x_train.reshape(x_train.shape[0], img_rows, img_cols, channels)
x_test = x_test.reshape(x_test.shape[0], img_rows, img_cols, channels)
input_shape = (img_rows, img_cols, channels)
x_train = x_train.astype('float32')
x_test = x_test.astype('float32')
x_train /= 255
x_test /= 255
pdb.set_trace()
def create_own_base_model(input_shape):
model_input = Input(shape=input_shape)
embedding = Conv2D(32, kernel_size=(10, 10), input_shape=input_shape)(model_input)
embedding = MaxPooling2D(pool_size=(2, 2))(embedding)
embedding = Conv2D(64, kernel_size=(7, 7))(embedding)
embedding = MaxPooling2D(pool_size=(2, 2))(embedding)
embedding = Conv2D(128, kernel_size=(4, 4))(embedding)
embedding = MaxPooling2D(pool_size=(2, 2))(embedding)
embedding = Conv2D(256, kernel_size=(4, 4))(embedding)
embedding = MaxPooling2D(pool_size=(2, 2))(embedding)
embedding = Flatten()(embedding)
embedding = Dense(4096, activation='sigmoid')(embedding)
embedding = BatchNormalization()(embedding)
embedding = Activation(activation='relu')(embedding)
return Model(model_input, embedding)
def create_base_model(input_shape):
model_input = Input(shape=input_shape)
embedding = Conv2D(32, kernel_size=(3, 3), input_shape=input_shape)(model_input)
embedding = BatchNormalization()(embedding)
embedding = Activation(activation='relu')(embedding)
embedding = MaxPooling2D(pool_size=(2, 2))(embedding)
embedding = Conv2D(64, kernel_size=(3, 3))(embedding)
embedding = BatchNormalization()(embedding)
embedding = Activation(activation='relu')(embedding)
embedding = MaxPooling2D(pool_size=(2, 2))(embedding)
embedding = Flatten()(embedding)
embedding = Dense(128)(embedding)
embedding = BatchNormalization()(embedding)
embedding = Activation(activation='relu')(embedding)
return Model(model_input, embedding)
def create_head_model(embedding_shape):
embedding_a = Input(shape=embedding_shape[1:])
embedding_b = Input(shape=embedding_shape[1:])
head = Concatenate()([embedding_a, embedding_b])
head = Dense(8)(head)
head = BatchNormalization()(head)
head = Activation(activation='sigmoid')(head)
head = Dense(1)(head)
head = BatchNormalization()(head)
head = Activation(activation='sigmoid')(head)
return Model([embedding_a, embedding_b], head)
def get_batch(x_train, y_train, x_test, y_test, cat_train, batch_size=64):
temp_x = x_train
temp_cat_list = cat_train
start=0
batch_x=[]
batch_y = np.zeros(batch_size)
batch_y[int(batch_size/2):] = 1
np.random.shuffle(batch_y)
class_list = np.random.randint(start, len(cat_train), batch_size)
batch_x.append(np.zeros((batch_size, 100, 100, 3)))
batch_x.append(np.zeros((batch_size, 100, 100, 3)))
for i in range(0, batch_size):
batch_x[0][i] = temp_x[np.random.choice(temp_cat_list[class_list[i]])]
#If train_y has 0 pick from the same class, else pick from any other class
if batch_y[i]==0:
r = np.random.choice(temp_cat_list[class_list[i]])
batch_x[1][i] = temp_x[r]
else:
temp_list = np.append(temp_cat_list[:class_list[i]].flatten(), temp_cat_list[class_list[i]+1:].flatten())
batch_x[1][i] = temp_x[np.random.choice(temp_list)]
return(batch_x, batch_y)
num_classes = 131
epochs = 2000
base_model = create_base_model(input_shape)
head_model = create_head_model(base_model.output_shape)
siamese_network = SiameseNetwork(base_model, head_model)
siamese_network.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])
siamese_checkpoint_path = "./siamese_checkpoint"
siamese_callbacks = [
# EarlyStopping(monitor='val_accuracy', patience=10, verbose=0),
ModelCheckpoint(siamese_checkpoint_path, monitor='val_accuracy', save_best_only=True, verbose=0)
]
# batch_size = 64
# for epoch in range(1, epochs):
# batch_x, batch_y = get_batch(x_train, y_train, x_test, y_test, cat_train, train_size, batch_size)
# loss = siamese_network.train_on_batch(batch_x, batch_y)
# print('Epoch:', epoch, ', Loss:', loss)
siamese_network.fit(x_train, y_train,
validation_data=(x_test, y_test),
batch_size=45,
epochs=epochs,
callbacks=siamese_callbacks)
# try:
# siamese_network = keras.models.load_model(siamese_checkpoint_path)
# except Exception as e:
# print(e)
# print("!!!!!!")
# siamese_network.load_weights(siamese_checkpoint_path)
embedding = base_model.outputs[-1]
y_train = keras.utils.to_categorical(y_train)
y_test = keras.utils.to_categorical(y_test)
# Add softmax layer to the pre-trained embedding network
embedding = Dense(num_classes)(embedding)
embedding = BatchNormalization()(embedding)
embedding = Activation(activation='sigmoid')(embedding)
model = Model(base_model.inputs[0], embedding)
model.compile(loss=keras.losses.binary_crossentropy,
optimizer=keras.optimizers.Adam(),
metrics=['accuracy'])
model_checkpoint_path = "./model_checkpoint"
model__callbacks = [
# EarlyStopping(monitor='val_accuracy', patience=10, verbose=0),
ModelCheckpoint(model_checkpoint_path, monitor='val_accuracy', save_best_only=True, verbose=0)
]
# for e in range(1, epochs):
# batch_x, batch_y = get_batch(x_train, y_train, x_test, y_test, cat_train, train_size, batch_size)
# loss = model.train_on_batch(batch_x, batch_y)
# print('Epoch:', epoch, ', Loss:', loss)
model.fit(x_train, y_train,
batch_size=128,
epochs=epochs,
callbacks=model__callbacks,
validation_data=(x_test, y_test))
# try:
# model = keras.models.load_model(model_checkpoint_path)
# except Exception as e:
# print(e)
# print("!!!!!!")
# model.load_weights(model_checkpoint_path)
score = model.evaluate(x_test, y_test, verbose=0)
print('Test loss:', score[0])
print('Test accuracy:', score[1])

BIN
model.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 44 KiB

BIN
mymodel.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 146 KiB

39
notes.md Normal file
View File

@ -0,0 +1,39 @@
the steps taken so far, which lead to a successfull detection of an image
- train the model defined in mnist_siamese_example, which uses the 'siamese.py' model to
create a siamese keras model.
- in this mnist siamese example, the data collection has been updated form the mnist drawing
sample to the fruit sample. Lots of work went into setting the arrays up correctly, because the
example from towards data science did not correctly seperate the classes. He had originally used
91 classes for teching and the rest for testing, where I now use images of every class for
teaching _and_ training.
- The images were shrunken down to 28 x 28 so the model defined in the siamese example could be used
without adaption
- in this example, there is two teachings going on, once he trains the siamese model (which is saved under
'siamese_checkpoint' and then he reteaches a new model based on this one, with some additonal layers ontop
I'm not yet sure what these do [todo] but 'I'll figure it out.
- after you've successfully trained the model, it's now saved to 'model_checkpoint' or 'siamese_checkpoint'
- The following steps can be used to classify two images:
Note, that it was so far only tested using images in a 'pdb' shell from the mnist_siamese_example script
```
import tensorflow.keras as keras
from PIL import image
model = keras.models.load_model('./siamese_checkpoint')
image1 = np.asarray(Image.open('../towards/data/fruits-360/Training/Avocado/r_254_100.jpg').convert('RGB').resize((28, 28))) / 255 / 255
image2 = np.asarray(Image.open('../towards/data/fruits-360/Training/Avocado/r_250_100.jpg').convert('RGB').resize((28, 28))) / 255 / 255
# note that the double division through 255 is only because the model bas taught with this double division, depends on
# the input numbers of course
output = model.predict([np.array([image2]), np.array([image1])])
# Note here, that the cast to np.array is nencessary - otherwise the input vector is malformed
print(output)
```

4
requirements.txt Normal file
View File

@ -0,0 +1,4 @@
keras==2.2.4
numpy==1.16.4
pytest==4.6.4
pep8==1.7.1

15
setup.py Normal file
View File

@ -0,0 +1,15 @@
from setuptools import setup
setup(
name='siamese',
version='0.1',
packages=[''],
url='https://github.com/aspamers/siamese',
license='MIT',
author='Abram Spamers',
author_email='aspamers@gmail.com',
install_requires=[
'keras', 'numpy',
],
description='An easy to use Keras Siamese Neural Network implementation'
)

291
siamese.py Normal file
View File

@ -0,0 +1,291 @@
"""
Siamese neural network module.
"""
import random, math
import numpy as np
from tensorflow.keras.layers import Input
from tensorflow.keras.models import Model
import pdb
class SiameseNetwork:
"""
A simple and lightweight siamese neural network implementation.
The SiameseNetwork class requires the base and head model to be defined via the constructor. The class exposes
public methods that allow it to behave similarly to a regular Keras model by passing kwargs through to the
underlying keras model object where possible. This allows Keras features like callbacks and metrics to be used.
"""
def __init__(self, base_model, head_model):
"""
Construct the siamese model class with the following structure.
-------------------------------------------------------------------
input1 -> base_model |
--> embedding --> head_model --> binary output
input2 -> base_model |
-------------------------------------------------------------------
:param base_model: The embedding model.
* Input shape must be equal to that of data.
:param head_model: The discriminator model.
* Input shape must be equal to that of embedding
* Output shape must be equal to 1..
"""
# Set essential parameters
self.base_model = base_model
self.head_model = head_model
# Get input shape from base model
self.input_shape = self.base_model.input_shape[1:]
# Initialize siamese model
self.siamese_model = None
self.__initialize_siamese_model()
def compile(self, *args, **kwargs):
"""
Configures the model for training.
Passes all arguments to the underlying Keras model compile function.
"""
self.siamese_model.compile(*args, **kwargs)
def train_on_batch(self, *args, **kwargs):
return self.siamese_model.train_on_batch(args[0], args[1])
def fit(self, *args, **kwargs):
"""
Trains the model on data generated batch-by-batch using the siamese network generator function.
Redirects arguments to the fit_generator function.
"""
x_train = args[0]
y_train = args[1]
x_test, y_test = kwargs.pop('validation_data')
batch_size = kwargs.pop('batch_size')
train_generator = self.__pair_generator(x_train, y_train, batch_size)
train_steps = math.floor(max(len(x_train) / batch_size, 1))
test_generator = self.__pair_generator(x_test, y_test, batch_size)
test_steps = math.floor(max(len(x_test) / batch_size, 1))
pdb.set_trace()
self.siamese_model.fit(train_generator,
steps_per_epoch=train_steps,
validation_data=test_generator,
validation_steps=test_steps, **kwargs)
def fit_generator(self, x_train, y_train, x_test, y_test, batch_size, *args, **kwargs):
"""
Trains the model on data generated batch-by-batch using the siamese network generator function.
:param x_train: Training input data.
:param y_train: Training output data.
:param x_test: Validation input data.
:param y_test: Validation output data.
:param batch_size: Number of pairs to generate per batch.
"""
train_generator = self.__pair_generator(x_train, y_train, batch_size)
train_steps = max(len(x_train) / batch_size, 1)
test_generator = self.__pair_generator(x_test, y_test, batch_size)
test_steps = max(len(x_test) / batch_size, 1)
self.siamese_model.fit_generator(train_generator,
steps_per_epoch=train_steps,
validation_data=test_generator,
validation_steps=test_steps,
*args, **kwargs)
def load_weights(self, checkpoint_path):
"""
Load siamese model weights. This also affects the reference to the base and head models.
:param checkpoint_path: Path to the checkpoint file.
"""
self.siamese_model.load_weights(checkpoint_path)
def evaluate(self, *args, **kwargs):
"""
Evaluate the siamese network with the same generator that is used to train it. Passes arguments through to the
underlying Keras function so that callbacks etc can be used.
Redirects arguments to the evaluate_generator function.
:return: A tuple of scores
"""
x = args[0]
y = args[1]
batch_size = kwargs.pop('batch_size')
generator = self.__pair_generator(x, y, batch_size)
steps = len(x) / batch_size
return self.siamese_model.evaluate_generator(generator, steps=steps, **kwargs)
def evaluate_generator(self, x, y, batch_size, *args, **kwargs):
"""
Evaluate the siamese network with the same generator that is used to train it. Passes arguments through to the
underlying Keras function so that callbacks etc can be used.
:param x: Input data
:param y: Class labels
:param batch_size: Number of pairs to generate per batch.
:return: A tuple of scores
"""
generator = self.__pair_generator(x, y, batch_size=batch_size)
steps = len(x) / batch_size
return self.siamese_model.evaluate_generator(generator, steps=steps, *args, **kwargs)
def __initialize_siamese_model(self):
"""
Create the siamese model structure using the supplied base and head model.
"""
input_a = Input(shape=self.input_shape)
input_b = Input(shape=self.input_shape)
processed_a = self.base_model(input_a)
processed_b = self.base_model(input_b)
head = self.head_model([processed_a, processed_b])
self.siamese_model = Model([input_a, input_b], head)
def __create_pairs(self, x, class_indices, batch_size, num_classes):
"""
Create a numpy array of positive and negative pairs and their associated labels.
:param x: Input data
:param class_indices: A python list of lists that contains each of the indices in the input data that belong
to each class. It is used to find and access elements in the input data that belong to a desired class.
* Example usage:
* element_index = class_indices[class][index]
* element = x[element_index]
:param batch_size: The number of pair samples to create.
:param num_classes: number of classes in the supplied input data
:return: A tuple of (Numpy array of pairs, Numpy array of labels)
"""
num_pairs = batch_size / 2
positive_pairs, positive_labels = self.__create_positive_pairs(x, class_indices, num_pairs, num_classes)
negative_pairs, negative_labels = self.__create_negative_pairs(x, class_indices, num_pairs, num_classes)
return np.array(positive_pairs + negative_pairs), np.array(positive_labels + negative_labels)
def __create_positive_pairs(self, x, class_indices, num_positive_pairs, num_classes):
"""
Create a list of positive pairs and labels. A positive pair is defined as two input samples of the same class.
:param x: Input data
:param class_indices: A python list of lists that contains each of the indices in the input data that belong
to each class. It is used to find and access elements in the input data that belong to a desired class.
* Example usage:
* element_index = class_indices[class][index]
* element = x[element_index]
:param num_positive_pairs: The number of positive pair samples to create.
:param num_classes: number of classes in the supplied input data
:return: A tuple of (python list of positive pairs, python list of positive labels)
"""
positive_pairs = []
positive_labels = []
for _ in range(int(num_positive_pairs)):
class_1 = random.randint(0, num_classes - 1)
num_elements = len(class_indices[class_1])
if num_elements == 0:
return [], []
index_1, index_2 = self.__randint_unequal(0, num_elements - 1)
element_index_1, element_index_2 = class_indices[class_1][index_1], class_indices[class_1][index_2]
positive_pairs.append([x[element_index_1], x[element_index_2]])
positive_labels.append([1.0])
return positive_pairs, positive_labels
def __create_negative_pairs(self, x, class_indices, num_negative_pairs, num_classes):
"""
Create a list of negative pairs and labels. A negative pair is defined as two input samples of different class.
:param x: Input data
:param class_indices: A python list of lists that contains each of the indices in the input data that belong
to each class. It is used to find and access elements in the input data that belong to a desired class.
* Example usage:
* element_index = class_indices[class][index]
* element = x[element_index]
:param num_negative_pairs: The number of negative pair samples to create.
:param num_classes: number of classes in the supplied input data
:return: A tuple of (python list of negative pairs, python list of negative labels)
"""
negative_pairs = []
negative_labels = []
if num_classes == 0:
return [], []
for _ in range(int(num_negative_pairs)):
cls_1, cls_2 = self.__randint_unequal(0, num_classes - 1)
try:
index_1 = random.randint(0, len(class_indices[cls_1]) - 1)
index_2 = random.randint(0, len(class_indices[cls_2]) - 1)
except Exception as e:
print(e)
pdb.set_trace()
element_index_1, element_index_2 = class_indices[cls_1][index_1], class_indices[cls_2][index_2]
negative_pairs.append([x[element_index_1], x[element_index_2]])
negative_labels.append([0.0])
return negative_pairs, negative_labels
def __pair_generator(self, x, y, batch_size):
"""
Creates a python generator that produces pairs from the original input data.
:param x: Input data
:param y: Integer class labels
:param batch_size: The number of pair samples to create per batch.
:return:
"""
class_indices, num_classes = self.__get_class_indices(y)
while True:
pairs, labels = self.__create_pairs(x, class_indices, batch_size, num_classes)
# The siamese network expects two inputs and one output. Split the pairs into a list of inputs.
yield [pairs[:, 0], pairs[:, 1]], labels
def __get_class_indices(self, y):
"""
Create a python list of lists that contains each of the indices in the input data that belong
to each class. It is used to find and access elements in the input data that belong to a desired class.
* Example usage:
* element_index = class_indices[class][index]
* element = x[element_index]
:param y: Integer class labels
:return: Python list of lists
"""
num_classes = np.max(y) + 1
return [np.where(y == i)[0] for i in range(num_classes)], num_classes
@staticmethod
def __randint_unequal(lower, upper):
"""
Get two random integers that are not equal.
Note: In some cases (such as there being only one sample of a class) there may be an endless loop here. This
will only happen on fairly exotic datasets though. May have to address in future.
:param lower: Lower limit inclusive of the random integer.
:param upper: Upper limit inclusive of the random integer. Need to use -1 for random indices.
:return: Tuple of (integer, integer)
"""
int_1 = random.randint(lower, upper)
int_2 = random.randint(lower, upper)
tries = 0
while int_1 == int_2:
tries += 1
if tries > 10:
break
int_1 = random.randint(lower, upper)
int_2 = random.randint(lower, upper)
return int_1, int_2

0
tests/__init__.py Normal file
View File

80
tests/test_siamese.py Normal file
View File

@ -0,0 +1,80 @@
"""
Tests for the siamese neural network module
"""
import numpy as np
import keras
from keras import Model, Input
from keras.layers import Concatenate, Dense, BatchNormalization, Activation
from siamese import SiameseNetwork
def test_siamese():
"""
Test that all components the siamese network work correctly by executing a
training run against generated data.
"""
num_classes = 5
input_shape = (3,)
epochs = 1000
# Generate some data
x_train = np.random.rand(100, 3)
y_train = np.random.randint(num_classes, size=100)
x_test = np.random.rand(30, 3)
y_test = np.random.randint(num_classes, size=30)
# Define base and head model
def create_base_model(input_shape):
model_input = Input(shape=input_shape)
embedding = Dense(4)(model_input)
embedding = BatchNormalization()(embedding)
embedding = Activation(activation='relu')(embedding)
return Model(model_input, embedding)
def create_head_model(embedding_shape):
embedding_a = Input(shape=embedding_shape)
embedding_b = Input(shape=embedding_shape)
head = Concatenate()([embedding_a, embedding_b])
head = Dense(4)(head)
head = BatchNormalization()(head)
head = Activation(activation='sigmoid')(head)
head = Dense(1)(head)
head = BatchNormalization()(head)
head = Activation(activation='sigmoid')(head)
return Model([embedding_a, embedding_b], head)
# Create siamese neural network
base_model = create_base_model(input_shape)
head_model = create_head_model(base_model.output_shape)
siamese_network = SiameseNetwork(base_model, head_model)
# Prepare siamese network for training
siamese_network.compile(loss='binary_crossentropy',
optimizer=keras.optimizers.adam())
# Evaluate network before training to establish a baseline
score_before = siamese_network.evaluate_generator(
x_train, y_train, batch_size=64
)
# Train network
siamese_network.fit(x_train, y_train,
validation_data=(x_test, y_test),
batch_size=64,
epochs=epochs)
# Evaluate network
score_after = siamese_network.evaluate(x_train, y_train, batch_size=64)
# Ensure that the training loss score improved as a result of the training
assert(score_before > score_after)