67 Commits

Author SHA1 Message Date
Christoph
94fe4435a8 Bump version: 0.4.4 → 0.4.5 2021-05-27 09:58:25 +02:00
Alexander Engelsberger
c204bc8e1f integrate reviews from ChristophRaab:master 2021-05-27 09:43:02 +02:00
Alexander Engelsberger
00615ae837 refactored gtlvq from ChristophRaab:master 2021-05-27 09:40:42 +02:00
Jensun Ravichandran
9f5f0d12dd [BUGFIX] Parse dictionary distribution appropirately 2021-05-25 20:52:39 +02:00
Jensun Ravichandran
8a291f7bfb Overload distribution argument in component initializers
The component initializers behave differently based on the type of the
`distribution` argument. If it is a Python
[list](https://docs.python.org/3/tutorial/datastructures.html), it is assumed
that there are as many entries in this list as there are classes, and the number
at each location of this list describes the number of prototypes to be used for
that particular class. So, `[1, 1, 1]` implies that we have three classes with
one prototype per class. If it is a Python
[tuple](https://docs.python.org/3/tutorial/datastructures.html), it a shorthand
of `(num_classes, prototypes_per_class)` is assumed. If it is a Python
[dictionary](https://docs.python.org/3/tutorial/datastructures.html), the
key-value pairs describe the class label and the number of prototypes for that
class respectively. So, `{0: 2, 1: 2, 2: 2}` implies that we have three classes
with labels `{1, 2, 3}`, each equipped with two prototypes.
2021-05-25 20:05:29 +02:00
Alexander Engelsberger
21e3e3b82d Cache pip in CI 2021-05-25 16:43:48 +02:00
Alexander Engelsberger
a6bd6e130a Add subpackages into prototorch namespace. 2021-05-25 16:40:53 +02:00
Alexander Engelsberger
fcdfa52892 Ignore artiifacts folder 2021-05-25 16:40:34 +02:00
Alexander Engelsberger
73e6fe384e Use 'num_' in all variable names 2021-05-25 15:57:05 +02:00
Alexander Engelsberger
aff7a385a3 Use dict for distribution
This change allows the use of LightningCLI.
2021-05-21 17:10:02 +02:00
Jensun Ravichandran
1e23ba05fa Add test_components 2021-05-21 16:22:02 +02:00
Alexander Engelsberger
ee30d4da5b [BUGFIX] Initializers can handle Dataloaders now 2021-05-21 16:00:20 +02:00
Alexander Engelsberger
14508f0600 [DOC] Small improvements 2021-05-21 15:59:44 +02:00
Jensun Ravichandran
e3f8828da4 Accept dataloaders for component initialization 2021-05-21 11:59:57 +02:00
Jensun Ravichandran
30adbf705c Update dependencies 2021-05-20 11:44:53 +02:00
Jensun Ravichandran
ee42fd68b1 NumpyDataset now has data and targets properties 2021-05-18 19:38:40 +02:00
Jensun Ravichandran
736d9a6349 Rename PositionAwareInitializer to DataAwareInitializer
Also, add the aliases `Zeros` and `Ones`.
2021-05-18 19:37:25 +02:00
Alexander Engelsberger
0055e15bc1 [DOC] Fix iris data dimension 2021-05-18 18:57:03 +02:00
Alexander Engelsberger
b2e1df7308 Improve dataset documentation. 2021-05-18 18:54:43 +02:00
Jensun Ravichandran
b935e9caf3 Update _get_dp_dm 2021-05-18 13:09:11 +02:00
Jensun Ravichandran
503ef0e05f Cleanup components 2021-05-17 16:58:57 +02:00
Jensun Ravichandran
dc6248413c Apply transformations in component initializers 2021-05-17 16:58:22 +02:00
Jensun Ravichandran
e73b70ceb7 Minor aesthetic change 2021-05-17 16:57:41 +02:00
Jensun Ravichandran
639198e774 Update Iris dataset 2021-05-17 16:57:13 +02:00
Alexander Engelsberger
768d969f89 Device agnostic initialization of components. 2021-05-13 15:21:04 +02:00
Alexander Engelsberger
aec422c277 Remove copy paste error from documentation. 2021-05-13 11:56:38 +02:00
Jensun Ravichandran
6c14170de6 [BUGFIX] Fix typo 2021-05-12 16:31:22 +02:00
Jensun Ravichandran
36a330aa66 Update component initializers 2021-05-12 16:28:55 +02:00
Jensun Ravichandran
acd4ac6a86 Flatten tensors before computing distances 2021-05-12 16:28:34 +02:00
Jensun Ravichandran
abe64cfe8f Merge pull request #3 from dmoebius-dm/dev
Removed wrong parameter.
2021-05-12 16:23:27 +02:00
Danny
caae95d01d Removed wrong parameter. 2021-05-12 16:00:01 +02:00
Alexander Engelsberger
088429a16a Bump version: 0.4.3 → 0.4.4 2021-05-11 17:17:49 +02:00
Jensun Ravichandran
b6145223c8 [HOTFIX] Add missing iris.py and fix knnc bug 2021-05-11 17:20:48 +02:00
Alexander Engelsberger
09256956f3 Bump version: 0.4.2 → 0.4.3 2021-05-11 17:04:08 +02:00
Jensun Ravichandran
0ca90fdcee Merge branch 'dev' of github.com:si-cim/prototorch into dev 2021-05-11 17:07:04 +02:00
Jensun Ravichandran
be21412f8a Add thin wrapper for the Iris dataset 2021-05-11 17:06:41 +02:00
Jensun Ravichandran
ae6bc47f87 [BUGFIX] Fix knnc 2021-05-11 17:06:27 +02:00
Jensun Ravichandran
7bb93f027a Support for unequal prototype distributions 2021-05-11 16:11:11 +02:00
Alexander Engelsberger
bc20acd63b Bump version: 0.4.1 → 0.4.2 2021-05-11 16:08:37 +02:00
Alexander Engelsberger
a864cf5d4d Bump version: 0.4.0 → 0.4.1 2021-05-11 13:37:54 +02:00
Alexander Engelsberger
2175f524e8 Update bug report issues template. 2021-05-11 13:35:38 +02:00
Alexander Engelsberger
c1c21e92df Add LVQ 1 and LVQ 2.1 loss functions. 2021-05-11 13:25:10 +02:00
Alexander Engelsberger
2b676ee06e Fix travis.yml. 2021-05-10 17:15:05 +02:00
Jensun Ravichandran
dda2f1d779 Clean-up CI setup 2021-05-10 16:37:43 +02:00
Alexander Engelsberger
3a8388e24f Version 0.4.0 2021-05-10 15:13:58 +02:00
Alexander Engelsberger
a9eef8ae6d Bump version: 0.3.1 → 0.4.0 2021-05-10 15:10:07 +02:00
Alexander Engelsberger
ac3091d8da Update Bumpversion config 2021-05-10 15:09:38 +02:00
Jensun Ravichandran
ce3991de94 Accept torch datasets to initialize components 2021-05-07 15:19:22 +02:00
Jensun Ravichandran
47b4b9bcb1 Expose prototorch.datasets 2021-05-07 15:18:33 +02:00
Alexander Engelsberger
19475d7e2b Update Tecator dataset storage id. 2021-05-06 18:42:36 +02:00
Jensun Ravichandran
269eb8ba25 Update unittests to reflect recent changes 2021-05-04 21:17:07 +02:00
Jensun Ravichandran
b06ded683d Update functions/activations.py 2021-05-04 20:55:49 +02:00
Jensun Ravichandran
466e9bde6b Refactor functions/losses.py 2021-05-04 20:36:48 +02:00
Alexander Engelsberger
fc7d64aaea Use Github Default Issue Templates 2021-05-04 11:20:15 +02:00
Jensun Ravichandran
9a7d3192c0 [BUG] GLVQ training is unstable
GLVQ training is unstable when prototypes are initialized exactly to datapoints
without small shifts. Perhaps because of zero distances?
2021-04-29 19:25:28 +02:00
Jensun Ravichandran
e686adbea1 Add spiral dataset 2021-04-29 19:15:35 +02:00
Jensun Ravichandran
b7d53aa5f1 Update initializers 2021-04-29 19:15:27 +02:00
Jensun Ravichandran
9b663477fd Update components 2021-04-29 18:06:26 +02:00
Jensun Ravichandran
a70166280a Update readme 2021-04-29 14:31:36 +02:00
Jensun Ravichandran
a083c4b276 Merge pull request #2 from si-cim/new-components
Create Component and initializer classes.
2021-04-29 13:25:58 +02:00
Alexander Engelsberger
40751aa50a Create Component and initializer classes. 2021-04-26 20:49:50 +02:00
Alexander Engelsberger
7c30ffe2c7 Automatic Formatting. 2021-04-23 17:25:23 +02:00
Alexander Engelsberger
e1d56595c1 Add NumpyDataset. 2021-04-23 17:24:59 +02:00
Alexander Engelsberger
4540c8848e Add neural gas energy function as loss. 2021-04-23 17:24:59 +02:00
Alexander Engelsberger
c88f288d12 Copy utilities for visualization from Protoflow. 2021-04-23 17:24:59 +02:00
Jensun Ravichandran
e2918dffed Add euclidean_distance_v2 2021-04-22 16:55:50 +02:00
Jensun Ravichandran
7d9dfc27ee Add similarities file 2021-04-22 13:12:19 +02:00
47 changed files with 1591 additions and 551 deletions

View File

@@ -1,20 +1,11 @@
[bumpversion]
current_version = 0.3.0-dev0
current_version = 0.4.5
commit = True
tag = True
parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(\-(?P<release>[a-z]+)(?P<build>\d+))?
parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)
serialize =
{major}.{minor}.{patch}-{release}{build}
{major}.{minor}.{patch}
[bumpversion:part:release]
optional_value = prod
first_value = dev
values =
dev
rc
prod
[bumpversion:file:setup.py]
[bumpversion:file:./prototorch/__init__.py]

31
.github/ISSUE_TEMPLATE/bug_report.md vendored Normal file
View File

@@ -0,0 +1,31 @@
---
name: Bug report
about: Create a report to help us improve
title: ''
labels: ''
assignees: ''
---
**Describe the bug**
A clear and concise description of what the bug is.
**To Reproduce**
Steps to reproduce the behavior:
1. Install Prototorch by running '...'
2. Run script '...'
3. See errors
**Expected behavior**
A clear and concise description of what you expected to happen.
**Screenshots**
If applicable, add screenshots to help explain your problem.
**Desktop (please complete the following information):**
- OS: [e.g. Ubuntu 20.10]
- Prototorch Version: [e.g. v0.4.0]
- Python Version: [e.g. 3.9.5]
**Additional context**
Add any other context about the problem here.

View File

@@ -0,0 +1,20 @@
---
name: Feature request
about: Suggest an idea for this project
title: ''
labels: ''
assignees: ''
---
**Is your feature request related to a problem? Please describe.**
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
**Describe the solution you'd like**
A clear and concise description of what you want to happen.
**Describe alternatives you've considered**
A clear and concise description of any alternative solutions or features you've considered.
**Additional context**
Add any other context or screenshots about the feature request here.

View File

@@ -23,10 +23,7 @@ jobs:
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install .
- name: Install extras
run: |
pip install -r requirements.txt
pip install .[all]
- name: Lint with flake8
run: |
pip install flake8

3
.gitignore vendored
View File

@@ -154,4 +154,5 @@ scratch*
# End of https://www.gitignore.io/api/visualstudiocode
.vscode/
reports
reports
artifacts

View File

@@ -4,11 +4,11 @@ language: python
python: 3.8
cache:
directories:
- "$HOME/.cache/pip"
- "./tests/artifacts"
# - "$HOME/.prototorch/datasets"
- "$HOME/datasets"
install:
- pip install . --progress-bar off
- pip install -r requirements.txt
- pip install .[all] --progress-bar off
# Generate code coverage report
script:
@@ -25,8 +25,8 @@ deploy:
password:
secure: rVQNCxKIuiEtMz4zLSsjdt6spG7cf3miKN5eqjxZfcELALHxAV4w/+CideQObOn3u9emmxb87R9XWKcogqK2MXqnuIcY4mWg7HUqaip1bhz/4YiVXjFILcG6itjX9IUF1DrtjKKRk6xryucSZcEB7yTcXz1hQTb768KWlLlKOVTRNwr7j07eyeafexz/L2ANQCqfOZgS4b0k2AMeDBRPykPULtyeneEFlb6MJZ2MxeqtTNVK4b/6VsQSZwQ9jGJNGWonn5Y287gHmzvEcymSJogTe2taxGBWawPnOsibws9v88DEAHdsEvYdnqEE3hFl0R5La2Lkjd8CjNUYegxioQ57i3WNS3iksq10ZLMCbH29lb9YPG7r6Y8z9H85735kV2gKLdf+o7SPS03TRgjSZKN6pn4pLG0VWkxC6l8VfLuJnRNTHX4g6oLQwOWIBbxybn9Zw/yLjAXAJNgBHt5v86H6Jfi1Va4AhEV6itkoH9IM3/uDhrE/mmorqyVled/CPNtBWNTyoDevLNxMUDnbuhH0JzLki+VOjKnTxEfq12JB8X9faFG5BjvU9oGjPPewrp5DGGzg6KDra7dikciWUxE1eTFFDhMyG1CFGcjKlDvlAGHyI6Kih35egGUeq+N/pitr2330ftM9Dm4rWpOTxPyCI89bXKssx/MgmLG7kSM=
on:
tags: true
skip_existing: true
tags: true
skip_existing: true
# The password is encrypted with:
# `cd prototorch && travis encrypt your-pypi-api-token --add deploy.password`

View File

@@ -31,15 +31,15 @@ To also install the extras, use
pip install -U prototorch[all]
```
*Note: If you're using [ZSH](https://www.zsh.org/), the square brackets `[ ]`
have to be escaped like so: `\[\]`, making the install command `pip install -U
prototorch\[all\]`.*
*Note: If you're using [ZSH](https://www.zsh.org/) (which is also the default
shell on MacOS now), the square brackets `[ ]` have to be escaped like so:
`\[\]`, making the install command `pip install -U prototorch\[all\]`.*
To install the bleeding-edge features and improvements:
```bash
git clone https://github.com/si-cim/prototorch.git
git checkout dev
cd prototorch
git checkout dev
pip install -e .[all]
```

View File

@@ -1,13 +1,24 @@
.. ProtoFlow API Reference
.. ProtoTorch API Reference
ProtoFlow API Reference
ProtoTorch API Reference
======================================
Datasets
--------------------------------------
Common Datasets
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. automodule:: prototorch.datasets
:members:
:undoc-members:
Abstract Datasets
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Abstract Datasets are used to build your own datasets.
.. autoclass:: prototorch.datasets.abstract.NumpyDataset
:members:
Functions
--------------------------------------

View File

@@ -12,9 +12,8 @@
#
import os
import sys
sys.path.insert(0, os.path.abspath("../../"))
import sphinx_rtd_theme
sys.path.insert(0, os.path.abspath("../../"))
# -- Project information -----------------------------------------------------
@@ -24,7 +23,7 @@ author = "Jensun Ravichandran"
# The full version, including alpha/beta/rc tags
#
release = "0.3.0-dev0"
release = "0.4.5"
# -- General configuration ---------------------------------------------------
@@ -47,6 +46,7 @@ extensions = [
"sphinx.ext.viewcode",
"sphinx_rtd_theme",
"sphinxcontrib.katex",
'sphinx_autodoc_typehints',
]
# katex_prerender = True
@@ -128,15 +128,12 @@ latex_elements = {
# The paper size ("letterpaper" or "a4paper").
#
# "papersize": "letterpaper",
# The font size ("10pt", "11pt" or "12pt").
#
# "pointsize": "10pt",
# Additional stuff for the LaTeX preamble.
#
# "preamble": "",
# Latex figure (float) alignment
#
# "figure_align": "htbp",
@@ -146,15 +143,21 @@ latex_elements = {
# (source start file, target name, title,
# author, documentclass [howto, manual, or own class]).
latex_documents = [
(master_doc, "prototorch.tex", "ProtoTorch Documentation",
"Jensun Ravichandran", "manual"),
(
master_doc,
"prototorch.tex",
"ProtoTorch Documentation",
"Jensun Ravichandran",
"manual",
),
]
# -- Options for manual page output ---------------------------------------
# One entry per manual page. List of tuples
# (source start file, name, description, authors, manual section).
man_pages = [(master_doc, "ProtoTorch", "ProtoTorch Documentation", [author], 1)]
man_pages = [(master_doc, "ProtoTorch", "ProtoTorch Documentation", [author],
1)]
# -- Options for Texinfo output -------------------------------------------
@@ -162,15 +165,24 @@ man_pages = [(master_doc, "ProtoTorch", "ProtoTorch Documentation", [author], 1)
# (source start file, target name, title, author,
# dir menu entry, description, category)
texinfo_documents = [
(master_doc, "prototorch", "ProtoTorch Documentation", author, "prototorch",
"Prototype-based machine learning in PyTorch.",
"Miscellaneous"),
(
master_doc,
"prototorch",
"ProtoTorch Documentation",
author,
"prototorch",
"Prototype-based machine learning in PyTorch.",
"Miscellaneous",
),
]
# Example configuration for intersphinx: refer to the Python standard library.
intersphinx_mapping = {
"python": ("https://docs.python.org/", None),
"numpy": ("https://docs.scipy.org/doc/numpy/", None),
"torch": ('https://pytorch.org/docs/stable/', None),
"pytorch_lightning":
("https://pytorch-lightning.readthedocs.io/en/stable/", None),
}
# -- Options for Epub output ----------------------------------------------

View File

@@ -27,9 +27,10 @@ class Model(torch.nn.Module):
self.proto_layer = Prototypes1D(
input_dim=2,
prototypes_per_class=3,
nclasses=3,
num_classes=3,
prototype_initializer="stratified_random",
data=[x_train, y_train])
data=[x_train, y_train],
)
def forward(self, x):
protos = self.proto_layer.prototypes
@@ -61,8 +62,10 @@ for epoch in range(70):
with torch.no_grad():
pred = wtac(dis, plabels)
correct = pred.eq(y_in.view_as(pred)).sum().item()
acc = 100. * correct / len(x_train)
print(f"Epoch: {epoch + 1:03d} Loss: {loss.item():05.02f} Acc: {acc:05.02f}%")
acc = 100.0 * correct / len(x_train)
print(
f"Epoch: {epoch + 1:03d} Loss: {loss.item():05.02f} Acc: {acc:05.02f}%"
)
# Take a gradient descent step
optimizer.zero_grad()
@@ -83,13 +86,15 @@ for epoch in range(70):
ax.set_ylabel("Data dimension 2")
cmap = "viridis"
ax.scatter(x_train[:, 0], x_train[:, 1], c=y_train, edgecolor="k")
ax.scatter(protos[:, 0],
protos[:, 1],
c=plabels,
cmap=cmap,
edgecolor="k",
marker="D",
s=50)
ax.scatter(
protos[:, 0],
protos[:, 1],
c=plabels,
cmap=cmap,
edgecolor="k",
marker="D",
s=50,
)
# Paint decision regions
x = np.vstack((x_train, protos))

View File

@@ -2,13 +2,12 @@
import matplotlib.pyplot as plt
import torch
from torch.utils.data import DataLoader
from prototorch.datasets.tecator import Tecator
from prototorch.functions.distances import sed
from prototorch.modules import Prototypes1D
from prototorch.modules.losses import GLVQLoss
from prototorch.utils.colors import get_legend_handles
from torch.utils.data import DataLoader
# Prepare the dataset and dataloader
train_data = Tecator(root="./artifacts", train=True)
@@ -20,11 +19,13 @@ class Model(torch.nn.Module):
"""GMLVQ model as a siamese network."""
super().__init__()
x, y = train_data.data, train_data.targets
self.p1 = Prototypes1D(input_dim=100,
prototypes_per_class=2,
nclasses=2,
prototype_initializer="stratified_random",
data=[x, y])
self.p1 = Prototypes1D(
input_dim=100,
prototypes_per_class=2,
num_classes=2,
prototype_initializer="stratified_random",
data=[x, y],
)
self.omega = torch.nn.Linear(in_features=100,
out_features=100,
bias=False)

View File

@@ -12,13 +12,13 @@ import numpy as np
import torch
import torch.nn as nn
import torchvision
from torchvision import transforms
from prototorch.modules.losses import GLVQLoss
from prototorch.functions.helper import calculate_prototype_accuracy
from prototorch.modules.losses import GLVQLoss
from prototorch.modules.models import GTLVQ
from torchvision import transforms
# Parameters and options
n_epochs = 50
num_epochs = 50
batch_size_train = 64
batch_size_test = 1000
learning_rate = 0.1
@@ -26,32 +26,40 @@ momentum = 0.5
log_interval = 10
cuda = "cuda:1"
random_seed = 1
device = torch.device(cuda if torch.cuda.is_available() else 'cpu')
device = torch.device(cuda if torch.cuda.is_available() else "cpu")
# Configures reproducability
torch.manual_seed(random_seed)
np.random.seed(random_seed)
# Prepare and preprocess the data
train_loader = torch.utils.data.DataLoader(torchvision.datasets.MNIST(
'./files/',
train=True,
download=True,
transform=torchvision.transforms.Compose(
[transforms.ToTensor(),
transforms.Normalize((0.1307, ), (0.3081, ))])),
batch_size=batch_size_train,
shuffle=True)
train_loader = torch.utils.data.DataLoader(
torchvision.datasets.MNIST(
"./files/",
train=True,
download=True,
transform=torchvision.transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.1307, ), (0.3081, ))
]),
),
batch_size=batch_size_train,
shuffle=True,
)
test_loader = torch.utils.data.DataLoader(torchvision.datasets.MNIST(
'./files/',
train=False,
download=True,
transform=torchvision.transforms.Compose(
[transforms.ToTensor(),
transforms.Normalize((0.1307, ), (0.3081, ))])),
batch_size=batch_size_test,
shuffle=True)
test_loader = torch.utils.data.DataLoader(
torchvision.datasets.MNIST(
"./files/",
train=False,
download=True,
transform=torchvision.transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.1307, ), (0.3081, ))
]),
),
batch_size=batch_size_test,
shuffle=True,
)
# Define the GLVQ model plus appropriate feature extractor
@@ -67,25 +75,34 @@ class CNNGTLVQ(torch.nn.Module):
):
super(CNNGTLVQ, self).__init__()
#Feature Extractor - Simple CNN
self.fe = nn.Sequential(nn.Conv2d(1, 32, 3, 1), nn.ReLU(),
nn.Conv2d(32, 64, 3, 1), nn.ReLU(),
nn.MaxPool2d(2), nn.Dropout(0.25),
nn.Flatten(), nn.Linear(9216, bottleneck_dim),
nn.Dropout(0.5), nn.LeakyReLU(),
nn.LayerNorm(bottleneck_dim))
# Feature Extractor - Simple CNN
self.fe = nn.Sequential(
nn.Conv2d(1, 32, 3, 1),
nn.ReLU(),
nn.Conv2d(32, 64, 3, 1),
nn.ReLU(),
nn.MaxPool2d(2),
nn.Dropout(0.25),
nn.Flatten(),
nn.Linear(9216, bottleneck_dim),
nn.Dropout(0.5),
nn.LeakyReLU(),
nn.LayerNorm(bottleneck_dim),
)
# Forward pass of subspace and prototype initialization data through feature extractor
subspace_data = self.fe(subspace_data)
prototype_data[0] = self.fe(prototype_data[0])
# Initialization of GTLVQ
self.gtlvq = GTLVQ(num_classes,
subspace_data,
prototype_data,
tangent_projection_type=tangent_projection_type,
feature_dim=bottleneck_dim,
prototypes_per_class=prototypes_per_class)
self.gtlvq = GTLVQ(
num_classes,
subspace_data,
prototype_data,
tangent_projection_type=tangent_projection_type,
feature_dim=bottleneck_dim,
prototypes_per_class=prototypes_per_class,
)
def forward(self, x):
# Feature Extraction
@@ -103,23 +120,27 @@ subspace_data = torch.cat(
prototype_data = next(iter(train_loader))
# Build the CNN GTLVQ model
model = CNNGTLVQ(10,
subspace_data,
prototype_data,
tangent_projection_type="local",
bottleneck_dim=128).to(device)
model = CNNGTLVQ(
10,
subspace_data,
prototype_data,
tangent_projection_type="local",
bottleneck_dim=128,
).to(device)
# Optimize using SGD optimizer from `torch.optim`
optimizer = torch.optim.Adam([{
'params': model.fe.parameters()
}, {
'params': model.gtlvq.parameters()
}],
lr=learning_rate)
criterion = GLVQLoss(squashing='sigmoid_beta', beta=10)
optimizer = torch.optim.Adam(
[{
"params": model.fe.parameters()
}, {
"params": model.gtlvq.parameters()
}],
lr=learning_rate,
)
criterion = GLVQLoss(squashing="sigmoid_beta", beta=10)
# Training loop
for epoch in range(n_epochs):
for epoch in range(num_epochs):
for batch_idx, (x_train, y_train) in enumerate(train_loader):
model.train()
x_train, y_train = x_train.to(device), y_train.to(device)
@@ -139,8 +160,8 @@ for epoch in range(n_epochs):
if batch_idx % log_interval == 0:
acc = calculate_prototype_accuracy(distances, y_train, plabels)
print(
f'Epoch: {epoch + 1:02d}/{n_epochs:02d} Epoch Progress: {100. * batch_idx / len(train_loader):02.02f} % Loss: {loss.item():02.02f} \
Train Acc: {acc.item():02.02f}')
f"Epoch: {epoch + 1:02d}/{num_epochs:02d} Epoch Progress: {100. * batch_idx / len(train_loader):02.02f} % Loss: {loss.item():02.02f} \
Train Acc: {acc.item():02.02f}")
# Test
with torch.no_grad():
@@ -154,9 +175,9 @@ for epoch in range(n_epochs):
i = torch.argmin(test_distances, 1)
correct += torch.sum(y_test == test_plabels[i])
total += y_test.size(0)
print('Accuracy of the network on the test images: %d %%' %
print("Accuracy of the network on the test images: %d %%" %
(torch.true_divide(correct, total) * 100))
# Save the model
PATH = './glvq_mnist_model.pth'
PATH = "./glvq_mnist_model.pth"
torch.save(model.state_dict(), PATH)

View File

@@ -22,10 +22,12 @@ class Model(torch.nn.Module):
def __init__(self):
"""Local-GMLVQ model."""
super().__init__()
self.p1 = Prototypes1D(input_dim=2,
prototype_distribution=[1, 2, 2],
prototype_initializer="stratified_random",
data=[x_train, y_train])
self.p1 = Prototypes1D(
input_dim=2,
prototype_distribution=[1, 2, 2],
prototype_initializer="stratified_random",
data=[x_train, y_train],
)
omegas = torch.zeros(5, 2, 2)
self.omegas = torch.nn.Parameter(omegas)
eye_(self.omegas)
@@ -76,14 +78,16 @@ for epoch in range(100):
ax.set_xlabel("Data dimension 1")
ax.set_ylabel("Data dimension 2")
cmap = "viridis"
ax.scatter(x_train[:, 0], x_train[:, 1], c=y_train, edgecolor='k')
ax.scatter(protos[:, 0],
protos[:, 1],
c=plabels,
cmap=cmap,
edgecolor='k',
marker='D',
s=50)
ax.scatter(x_train[:, 0], x_train[:, 1], c=y_train, edgecolor="k")
ax.scatter(
protos[:, 0],
protos[:, 1],
c=plabels,
cmap=cmap,
edgecolor="k",
marker="D",
s=50,
)
# Paint decision regions
x = np.vstack((x_train, protos))

View File

@@ -0,0 +1,65 @@
"""This example script shows the usage of the new components architecture.
Serialization/deserialization also works as expected.
"""
# DATASET
import torch
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
x_train, y_train = load_iris(return_X_y=True)
x_train = x_train[:, [0, 2]]
scaler.fit(x_train)
x_train = scaler.transform(x_train)
x_train = torch.Tensor(x_train)
y_train = torch.Tensor(y_train)
num_classes = len(torch.unique(y_train))
# CREATE NEW COMPONENTS
from prototorch.components import *
from prototorch.components.initializers import *
unsupervised = Components(6, SelectionInitializer(x_train))
print(unsupervised())
prototypes = LabeledComponents(
(3, 2), StratifiedSelectionInitializer(x_train, y_train))
print(prototypes())
components = ReasoningComponents(
(3, 6), StratifiedSelectionInitializer(x_train, y_train))
print(components())
# TEST SERIALIZATION
import io
save = io.BytesIO()
torch.save(unsupervised, save)
save.seek(0)
serialized_unsupervised = torch.load(save)
assert torch.all(unsupervised.components == serialized_unsupervised.components
), "Serialization of Components failed."
save = io.BytesIO()
torch.save(prototypes, save)
save.seek(0)
serialized_prototypes = torch.load(save)
assert torch.all(prototypes.components == serialized_prototypes.components
), "Serialization of Components failed."
assert torch.all(prototypes.component_labels == serialized_prototypes.
component_labels), "Serialization of Components failed."
save = io.BytesIO()
torch.save(components, save)
save.seek(0)
serialized_components = torch.load(save)
assert torch.all(components.components == serialized_components.components
), "Serialization of Components failed."
assert torch.all(components.reasonings == serialized_components.reasonings
), "Serialization of Components failed."

View File

@@ -1,31 +1,32 @@
"""ProtoTorch package."""
# #############################################
# Core Setup
# #############################################
__version__ = "0.3.0-dev0"
import pkgutil
from prototorch import datasets, functions, modules
import pkg_resources
from . import components, datasets, functions, modules, utils
from .datasets import *
# Core Setup
__version__ = "0.4.5"
__all_core__ = [
"datasets",
"functions",
"modules",
"components",
"utils",
]
# #############################################
# Plugin Loader
# #############################################
import pkgutil
import pkg_resources
__path__ = pkgutil.extend_path(__path__, __name__)
def discover_plugins():
return {
entry_point.name: entry_point.load()
for entry_point in pkg_resources.iter_entry_points("prototorch.plugins")
for entry_point in pkg_resources.iter_entry_points(
"prototorch.plugins")
}
@@ -33,14 +34,12 @@ discovered_plugins = discover_plugins()
locals().update(discovered_plugins)
# Generate combines __version__ and __all__
version_plugins = "\n".join(
[
"- " + name + ": v" + plugin.__version__
for name, plugin in discovered_plugins.items()
]
)
version_plugins = "\n".join([
"- " + name + ": v" + plugin.__version__
for name, plugin in discovered_plugins.items()
])
if version_plugins != "":
version_plugins = "\nPlugins: \n" + version_plugins
version = "core: v" + __version__ + version_plugins
__all__ = __all_core__ + list(discovered_plugins.keys())
__all__ = __all_core__ + list(discovered_plugins.keys())

View File

@@ -0,0 +1,2 @@
from prototorch.components.components import *
from prototorch.components.initializers import *

View File

@@ -0,0 +1,161 @@
"""ProtoTorch components modules."""
import warnings
import torch
from prototorch.components.initializers import (ClassAwareInitializer,
ComponentsInitializer,
CustomLabelsInitializer,
EqualLabelsInitializer,
UnequalLabelsInitializer,
ZeroReasoningsInitializer)
from torch.nn.parameter import Parameter
class Components(torch.nn.Module):
"""Components is a set of learnable Tensors."""
def __init__(self,
num_components=None,
initializer=None,
*,
initialized_components=None):
super().__init__()
self.num_components = num_components
# Ignore all initialization settings if initialized_components is given.
if initialized_components is not None:
self.register_parameter("_components",
Parameter(initialized_components))
if num_components is not None or initializer is not None:
wmsg = "Arguments ignored while initializing Components"
warnings.warn(wmsg)
else:
self._initialize_components(initializer)
def _precheck_initializer(self, initializer):
if not isinstance(initializer, ComponentsInitializer):
emsg = f"`initializer` has to be some subtype of " \
f"{ComponentsInitializer}. " \
f"You have provided: {initializer=} instead."
raise TypeError(emsg)
def _initialize_components(self, initializer):
self._precheck_initializer(initializer)
_components = initializer.generate(self.num_components)
self.register_parameter("_components", Parameter(_components))
@property
def components(self):
"""Tensor containing the component tensors."""
return self._components.detach()
def forward(self):
return self._components
def extra_repr(self):
return f"components.shape: {tuple(self._components.shape)}"
class LabeledComponents(Components):
"""LabeledComponents generate a set of components and a set of labels.
Every Component has a label assigned.
"""
def __init__(self,
distribution=None,
initializer=None,
*,
initialized_components=None):
if initialized_components is not None:
components, component_labels = initialized_components
super().__init__(initialized_components=components)
self._labels = component_labels
else:
_labels = self._initialize_labels(distribution)
super().__init__(len(_labels), initializer=initializer)
self.register_buffer("_labels", _labels)
def _initialize_components(self, initializer):
if isinstance(initializer, ClassAwareInitializer):
self._precheck_initializer(initializer)
_components = initializer.generate(self.num_components,
self.distribution)
self.register_parameter("_components", Parameter(_components))
else:
super()._initialize_components(initializer)
def _initialize_labels(self, distribution):
if type(distribution) == dict:
if "num_classes" in distribution.keys():
labels = EqualLabelsInitializer(
distribution["num_classes"],
distribution["prototypes_per_class"])
else:
labels = CustomLabelsInitializer(distribution)
elif type(distribution) == tuple:
num_classes, prototypes_per_class = distribution
labels = EqualLabelsInitializer(num_classes, prototypes_per_class)
elif type(distribution) == list:
labels = UnequalLabelsInitializer(distribution)
self.distribution = labels.distribution
return labels.generate()
@property
def component_labels(self):
"""Tensor containing the component tensors."""
return self._labels.detach()
def forward(self):
return super().forward(), self._labels
class ReasoningComponents(Components):
"""ReasoningComponents generate a set of components and a set of reasoning matrices.
Every Component has a reasoning matrix assigned.
A reasoning matrix is a Nx2 matrix, where N is the number of Classes. The
first element is called positive reasoning :math:`p`, the second negative
reasoning :math:`n`. A components can reason in favour (positive) of a
class, against (negative) a class or not at all (neutral).
It holds that :math:`0 \leq n \leq 1`, :math:`0 \leq p \leq 1` and :math:`0
\leq n+p \leq 1`. Therefore :math:`n` and :math:`p` are two elements of a
three element probability distribution.
"""
def __init__(self,
reasonings=None,
initializer=None,
*,
initialized_components=None):
if initialized_components is not None:
components, reasonings = initialized_components
super().__init__(initialized_components=components)
self.register_parameter("_reasonings", reasonings)
else:
self._initialize_reasonings(reasonings)
super().__init__(len(self._reasonings), initializer=initializer)
def _initialize_reasonings(self, reasonings):
if type(reasonings) == tuple:
num_classes, num_components = reasonings
reasonings = ZeroReasoningsInitializer(num_classes, num_components)
_reasonings = reasonings.generate()
self.register_parameter("_reasonings", _reasonings)
@property
def reasonings(self):
"""Returns Reasoning Matrix.
Dimension NxCx2
"""
return self._reasonings.detach()
def forward(self):
return super().forward(), self._reasonings

View File

@@ -0,0 +1,228 @@
"""ProtoTroch Initializers."""
import warnings
from collections.abc import Iterable
from itertools import chain
import torch
from torch.utils.data import DataLoader, Dataset
def parse_data_arg(data_arg):
if isinstance(data_arg, Dataset):
data_arg = DataLoader(data_arg, batch_size=len(data_arg))
if isinstance(data_arg, DataLoader):
data = torch.tensor([])
labels = torch.tensor([])
for x, y in data_arg:
data = torch.cat([data, x])
labels = torch.cat([labels, y])
else:
data, labels = data_arg
if not isinstance(data, torch.Tensor):
wmsg = f"Converting data to {torch.Tensor}."
warnings.warn(wmsg)
data = torch.Tensor(data)
if not isinstance(labels, torch.Tensor):
wmsg = f"Converting labels to {torch.Tensor}."
warnings.warn(wmsg)
labels = torch.Tensor(labels)
return data, labels
# Components
class ComponentsInitializer(object):
def generate(self, number_of_components):
raise NotImplementedError("Subclasses should implement this!")
class DimensionAwareInitializer(ComponentsInitializer):
def __init__(self, c_dims):
super().__init__()
if isinstance(c_dims, Iterable):
self.components_dims = tuple(c_dims)
else:
self.components_dims = (c_dims, )
class OnesInitializer(DimensionAwareInitializer):
def generate(self, length):
gen_dims = (length, ) + self.components_dims
return torch.ones(gen_dims)
class ZerosInitializer(DimensionAwareInitializer):
def generate(self, length):
gen_dims = (length, ) + self.components_dims
return torch.zeros(gen_dims)
class UniformInitializer(DimensionAwareInitializer):
def __init__(self, c_dims, min=0.0, max=1.0):
super().__init__(c_dims)
self.min = min
self.max = max
def generate(self, length):
gen_dims = (length, ) + self.components_dims
return torch.ones(gen_dims).uniform_(self.min, self.max)
class DataAwareInitializer(ComponentsInitializer):
def __init__(self, data):
super().__init__()
self.data = data
class SelectionInitializer(DataAwareInitializer):
def generate(self, length):
indices = torch.LongTensor(length).random_(0, len(self.data))
return self.data[indices]
class MeanInitializer(DataAwareInitializer):
def generate(self, length):
mean = torch.mean(self.data, dim=0)
repeat_dim = [length] + [1] * len(mean.shape)
return mean.repeat(repeat_dim)
class ClassAwareInitializer(ComponentsInitializer):
def __init__(self, data, transform=torch.nn.Identity()):
super().__init__()
data, labels = parse_data_arg(data)
self.data = data
self.labels = labels
self.transform = transform
self.clabels = torch.unique(self.labels)
self.num_classes = len(self.clabels)
def _get_samples_from_initializer(self, length, dist):
if not dist:
per_class = length // self.num_classes
dist = self.num_classes * [per_class]
if type(dist) == dict:
dist = dist.values()
samples_list = [
init.generate(n) for init, n in zip(self.initializers, dist)
]
out = torch.vstack(samples_list)
with torch.no_grad():
out = self.transform(out)
return out
def __del__(self):
del self.data
del self.labels
class StratifiedMeanInitializer(ClassAwareInitializer):
def __init__(self, data, **kwargs):
super().__init__(data, **kwargs)
self.initializers = []
for clabel in self.clabels:
class_data = self.data[self.labels == clabel]
class_initializer = MeanInitializer(class_data)
self.initializers.append(class_initializer)
def generate(self, length, dist=[]):
samples = self._get_samples_from_initializer(length, dist)
return samples
class StratifiedSelectionInitializer(ClassAwareInitializer):
def __init__(self, data, noise=None, **kwargs):
super().__init__(data, **kwargs)
self.noise = noise
self.initializers = []
for clabel in self.clabels:
class_data = self.data[self.labels == clabel]
class_initializer = SelectionInitializer(class_data)
self.initializers.append(class_initializer)
def add_noise_v1(self, x):
return x + self.noise
def add_noise_v2(self, x):
"""Shifts some dimensions of the data randomly."""
n1 = torch.rand_like(x)
n2 = torch.rand_like(x)
mask = torch.bernoulli(n1) - torch.bernoulli(n2)
return x + (self.noise * mask)
def generate(self, length, dist=[]):
samples = self._get_samples_from_initializer(length, dist)
if self.noise is not None:
samples = self.add_noise_v1(samples)
return samples
# Labels
class LabelsInitializer:
def generate(self):
raise NotImplementedError("Subclasses should implement this!")
class UnequalLabelsInitializer(LabelsInitializer):
def __init__(self, dist):
self.dist = dist
@property
def distribution(self):
return self.dist
def generate(self, clabels=None, dist=None):
if not clabels:
clabels = range(len(self.dist))
if not dist:
dist = self.dist
labels = list(chain(*[[i] * n for i, n in zip(clabels, dist)]))
return torch.LongTensor(labels)
class EqualLabelsInitializer(LabelsInitializer):
def __init__(self, classes, per_class):
self.classes = classes
self.per_class = per_class
@property
def distribution(self):
return self.classes * [self.per_class]
def generate(self):
return torch.arange(self.classes).repeat(self.per_class, 1).T.flatten()
class CustomLabelsInitializer(UnequalLabelsInitializer):
def generate(self):
clabels = list(self.dist.keys())
dist = list(self.dist.values())
return super().generate(clabels, dist)
# Reasonings
class ReasoningsInitializer:
def generate(self, length):
raise NotImplementedError("Subclasses should implement this!")
class ZeroReasoningsInitializer(ReasoningsInitializer):
def __init__(self, classes, length):
self.classes = classes
self.length = length
def generate(self):
return torch.zeros((self.length, self.classes, 2))
# Aliases
SSI = StratifiedSampleInitializer = StratifiedSelectionInitializer
SMI = StratifiedMeanInitializer
Random = RandomInitializer = UniformInitializer
Zeros = ZerosInitializer
Ones = OnesInitializer

View File

@@ -1,7 +1,8 @@
"""ProtoTorch datasets."""
from .abstract import NumpyDataset
from .iris import Iris
from .spiral import Spiral
from .tecator import Tecator
__all__ = [
'Tecator',
]
__all__ = ['Iris', 'Spiral', 'Tecator']

View File

@@ -12,6 +12,15 @@ import os
import torch
class NumpyDataset(torch.utils.data.TensorDataset):
"""Create a PyTorch TensorDataset from NumPy arrays."""
def __init__(self, data, targets):
self.data = data
self.targets = targets
tensors = [torch.Tensor(data), torch.Tensor(targets)]
super().__init__(*tensors)
class Dataset(torch.utils.data.Dataset):
"""Abstract dataset class to be inherited."""
@@ -44,15 +53,13 @@ class ProtoDataset(Dataset):
self._download()
if not self._check_exists():
raise RuntimeError(
"Dataset not found. " "You can use download=True to download it"
)
raise RuntimeError("Dataset not found. "
"You can use download=True to download it")
data_file = self.training_file if self.train else self.test_file
self.data, self.targets = torch.load(
os.path.join(self.processed_folder, data_file)
)
os.path.join(self.processed_folder, data_file))
@property
def raw_folder(self):
@@ -68,8 +75,9 @@ class ProtoDataset(Dataset):
def _check_exists(self):
return os.path.exists(
os.path.join(self.processed_folder, self.training_file)
) and os.path.exists(os.path.join(self.processed_folder, self.test_file))
os.path.join(
self.processed_folder, self.training_file)) and os.path.exists(
os.path.join(self.processed_folder, self.test_file))
def __repr__(self):
head = "Dataset " + self.__class__.__name__

View File

@@ -0,0 +1,40 @@
"""Thin wrapper for the Iris classification dataset from sklearn.
URL:
https://scikit-learn.org/stable/modules/generated/sklearn.datasets.load_iris.html
"""
from typing import Sequence
from prototorch.datasets.abstract import NumpyDataset
from sklearn.datasets import load_iris
class Iris(NumpyDataset):
"""
Iris Dataset by Ronald Fisher introduced in 1936.
The dataset contains four measurements from flowers of three species of iris.
.. list-table:: Iris
:header-rows: 1
* - dimensions
- classes
- training size
- validation size
- test size
* - 4
- 3
- 150
- 0
- 0
:param dims: select a subset of dimensions
"""
def __init__(self, dims: Sequence[int] = None):
x, y = load_iris(return_X_y=True)
if dims:
x = x[:, dims]
super().__init__(x, y)

View File

@@ -0,0 +1,57 @@
"""Spiral dataset for binary classification."""
import numpy as np
import torch
def make_spiral(num_samples=500, noise=0.3):
"""Generates the Spiral Dataset.
For use in Prototorch use `prototorch.datasets.Spiral` instead.
"""
def get_samples(n, delta_t):
points = []
for i in range(n):
r = i / num_samples * 5
t = 1.75 * i / n * 2 * np.pi + delta_t
x = r * np.sin(t) + np.random.rand(1) * noise
y = r * np.cos(t) + np.random.rand(1) * noise
points.append([x, y])
return points
n = num_samples // 2
positive = get_samples(n=n, delta_t=0)
negative = get_samples(n=n, delta_t=np.pi)
x = np.concatenate(
[np.array(positive).reshape(n, -1),
np.array(negative).reshape(n, -1)],
axis=0)
y = np.concatenate([np.zeros(n), np.ones(n)])
return x, y
class Spiral(torch.utils.data.TensorDataset):
"""Spiral dataset for binary classification.
This datasets consists of two spirals of two different classes.
.. list-table:: Spiral
:header-rows: 1
* - dimensions
- classes
- training size
- validation size
- test size
* - 2
- 2
- num_samples
- 0
- 0
:param num_samples: number of random samples
:param noise: noise added to the spirals
"""
def __init__(self, num_samples: int = 500, noise: float = 0.3):
x, y = make_spiral(num_samples, noise)
super().__init__(torch.Tensor(x), torch.LongTensor(y))

View File

@@ -40,19 +40,34 @@ import os
import numpy as np
import torch
from torchvision.datasets.utils import download_file_from_google_drive
from prototorch.datasets.abstract import ProtoDataset
from torchvision.datasets.utils import download_file_from_google_drive
class Tecator(ProtoDataset):
"""
`Tecator Dataset <http://lib.stat.cmu.edu/datasets/tecator>`__
for classification.
`Tecator Dataset <http://lib.stat.cmu.edu/datasets/tecator>`__ for classification.
The dataset contains wavelength measurements of meat.
.. list-table:: Tecator
:header-rows: 1
* - dimensions
- classes
- training size
- validation size
- test size
* - 100
- 2
- 129
- 43
- 43
"""
_resources = [
("1MMuUK8V41IgNpnPDbg3E-QAL6wlErTk0", "ba5607c580d0f91bb27dc29d13c2f8df"),
("1P9WIYnyxFPh6f1vqAbnKfK8oYmUgyV83",
"ba5607c580d0f91bb27dc29d13c2f8df"),
] # (google_storage_id, md5hash)
classes = ["0 - low_fat", "1 - high_fat"]
@@ -74,15 +89,15 @@ class Tecator(ProtoDataset):
print("Downloading...")
for fileid, md5 in self._resources:
filename = "tecator.npz"
download_file_from_google_drive(
fileid, root=self.raw_folder, filename=filename, md5=md5
)
download_file_from_google_drive(fileid,
root=self.raw_folder,
filename=filename,
md5=md5)
if self.verbose:
print("Processing...")
with np.load(
os.path.join(self.raw_folder, "tecator.npz"), allow_pickle=False
) as f:
with np.load(os.path.join(self.raw_folder, "tecator.npz"),
allow_pickle=False) as f:
x_train, y_train = f["x_train"], f["y_train"]
x_test, y_test = f["x_test"], f["y_test"]
training_set = [
@@ -94,9 +109,11 @@ class Tecator(ProtoDataset):
torch.tensor(y_test),
]
with open(os.path.join(self.processed_folder, self.training_file), "wb") as f:
with open(os.path.join(self.processed_folder, self.training_file),
"wb") as f:
torch.save(training_set, f)
with open(os.path.join(self.processed_folder, self.test_file), "wb") as f:
with open(os.path.join(self.processed_folder, self.test_file),
"wb") as f:
torch.save(test_set, f)
if self.verbose:

View File

@@ -4,9 +4,9 @@ from .activations import identity, sigmoid_beta, swish_beta
from .competitions import knnc, wtac
__all__ = [
'identity',
'sigmoid_beta',
'swish_beta',
'knnc',
'wtac',
"identity",
"sigmoid_beta",
"swish_beta",
"knnc",
"wtac",
]

View File

@@ -16,40 +16,43 @@ def register_activation(function):
@register_activation
# @torch.jit.script
def identity(x, beta=torch.tensor(0)):
def identity(x, beta=0.0):
"""Identity activation function.
Definition:
:math:`f(x) = x`
Keyword Arguments:
beta (`float`): Ignored.
"""
return x
@register_activation
# @torch.jit.script
def sigmoid_beta(x, beta=torch.tensor(10)):
def sigmoid_beta(x, beta=10.0):
r"""Sigmoid activation function with scaling.
Definition:
:math:`f(x) = \frac{1}{1 + e^{-\beta x}}`
Keyword Arguments:
beta (`torch.tensor`): Scaling parameter :math:`\beta`
beta (`float`): Scaling parameter :math:`\beta`
"""
out = torch.reciprocal(1.0 + torch.exp(-int(beta.item()) * x))
out = 1.0 / (1.0 + torch.exp(-1.0 * beta * x))
return out
@register_activation
# @torch.jit.script
def swish_beta(x, beta=torch.tensor(10)):
def swish_beta(x, beta=10.0):
r"""Swish activation function with scaling.
Definition:
:math:`f(x) = \frac{x}{1 + e^{-\beta x}}`
Keyword Arguments:
beta (`torch.tensor`): Scaling parameter :math:`\beta`
beta (`float`): Scaling parameter :math:`\beta`
"""
out = x * sigmoid_beta(x, beta=beta)
return out
@@ -61,4 +64,4 @@ def get_activation(funcname):
return funcname
if funcname in ACTIVATIONS:
return ACTIVATIONS.get(funcname)
raise NameError(f'Activation {funcname} was not found.')
raise NameError(f"Activation {funcname} was not found.")

View File

@@ -3,23 +3,22 @@
import torch
# @torch.jit.script
def stratified_min(distances, labels):
clabels = torch.unique(labels, dim=0)
nclasses = clabels.size()[0]
if distances.size()[1] == nclasses:
num_classes = clabels.size()[0]
if distances.size()[1] == num_classes:
# skip if only one prototype per class
return distances
batch_size = distances.size()[0]
winning_distances = torch.zeros(nclasses, batch_size)
inf = torch.full_like(distances.T, fill_value=float('inf'))
winning_distances = torch.zeros(num_classes, batch_size)
inf = torch.full_like(distances.T, fill_value=float("inf"))
# distances_to_wpluses = torch.where(matcher, distances, inf)
for i, cl in enumerate(clabels):
# cdists = distances.T[labels == cl]
matcher = torch.eq(labels.unsqueeze(dim=1), cl)
if labels.ndim == 2:
# if the labels are one-hot vectors
matcher = torch.eq(torch.sum(matcher, dim=-1), nclasses)
matcher = torch.eq(torch.sum(matcher, dim=-1), num_classes)
cdists = torch.where(matcher, distances.T, inf).T
winning_distances[i] = torch.min(cdists, dim=1,
keepdim=True).values.squeeze()
@@ -31,15 +30,15 @@ def stratified_min(distances, labels):
return winning_distances.T # return with `batch_size` first
# @torch.jit.script
def wtac(distances, labels):
winning_indices = torch.min(distances, dim=1).indices
winning_labels = labels[winning_indices].squeeze()
return winning_labels
# @torch.jit.script
def knnc(distances, labels, k):
winning_indices = torch.topk(-distances, k=k.item(), dim=1).indices
winning_labels = labels[winning_indices].squeeze()
def knnc(distances, labels, k=1):
winning_indices = torch.topk(-distances, k=k, dim=1).indices
# winning_labels = torch.mode(labels[winning_indices].squeeze(),
# dim=1).values
winning_labels = torch.mode(labels[winning_indices], dim=1).values
return winning_labels

View File

@@ -1,12 +1,9 @@
"""ProtoTorch distance functions."""
import torch
from prototorch.functions.helper import (
equal_int_shape,
_int_and_mixed_shape,
_check_shapes,
)
import numpy as np
import torch
from prototorch.functions.helper import (_check_shapes, _int_and_mixed_shape,
equal_int_shape, get_flat)
def squared_euclidean_distance(x, y):
@@ -14,12 +11,10 @@ def squared_euclidean_distance(x, y):
Compute :math:`{\langle \bm x - \bm y \rangle}_2`
:param `torch.tensor` x: Two dimensional vector
:param `torch.tensor` y: Two dimensional vector
**Alias:**
``prototorch.functions.distances.sed``
"""
x, y = get_flat(x, y)
expanded_x = x.unsqueeze(dim=1)
batchwise_difference = y - expanded_x
differences_raised = torch.pow(batchwise_difference, 2)
@@ -32,30 +27,40 @@ def euclidean_distance(x, y):
Compute :math:`\sqrt{{\langle \bm x - \bm y \rangle}_2}`
:param `torch.tensor` x: Input Tensor of shape :math:`X \times N`
:param `torch.tensor` y: Input Tensor of shape :math:`Y \times N`
:returns: Distance Tensor of shape :math:`X \times Y`
:rtype: `torch.tensor`
"""
x, y = get_flat(x, y)
distances_raised = squared_euclidean_distance(x, y)
distances = torch.sqrt(distances_raised)
return distances
def euclidean_distance_v2(x, y):
x, y = get_flat(x, y)
diff = y - x.unsqueeze(1)
pairwise_distances = (diff @ diff.permute((0, 2, 1))).sqrt()
# Passing `dim1=-2` and `dim2=-1` to `diagonal()` takes the
# batch diagonal. See:
# https://pytorch.org/docs/stable/generated/torch.diagonal.html
distances = torch.diagonal(pairwise_distances, dim1=-2, dim2=-1)
# print(f"{diff.shape=}") # (nx, ny, ndim)
# print(f"{pairwise_distances.shape=}") # (nx, ny, ny)
# print(f"{distances.shape=}") # (nx, ny)
return distances
def lpnorm_distance(x, y, p):
r"""
Calculates the lp-norm between :math:`\bm x` and :math:`\bm y`.
r"""Calculate the lp-norm between :math:`\bm x` and :math:`\bm y`.
Also known as Minkowski distance.
Compute :math:`{\| \bm x - \bm y \|}_p`.
Calls ``torch.cdist``
:param `torch.tensor` x: Two dimensional vector
:param `torch.tensor` y: Two dimensional vector
:param p: p parameter of the lp norm
"""
x, y = get_flat(x, y)
distances = torch.cdist(x, y, p=p)
return distances
@@ -65,10 +70,9 @@ def omega_distance(x, y, omega):
Compute :math:`{\| \Omega \bm x - \Omega \bm y \|}_p`
:param `torch.tensor` x: Two dimensional vector
:param `torch.tensor` y: Two dimensional vector
:param `torch.tensor` omega: Two dimensional matrix
"""
x, y = get_flat(x, y)
projected_x = x @ omega
projected_y = y @ omega
distances = squared_euclidean_distance(projected_x, projected_y)
@@ -80,15 +84,14 @@ def lomega_distance(x, y, omegas):
Compute :math:`{\| \Omega_k \bm x - \Omega_k \bm y_k \|}_p`
:param `torch.tensor` x: Two dimensional vector
:param `torch.tensor` y: Two dimensional vector
:param `torch.tensor` omegas: Three dimensional matrix
"""
x, y = get_flat(x, y)
projected_x = x @ omegas
projected_y = torch.diagonal(y @ omegas).T
expanded_y = torch.unsqueeze(projected_y, dim=1)
batchwise_difference = expanded_y - projected_x
differences_squared = batchwise_difference ** 2
differences_squared = batchwise_difference**2
distances = torch.sum(differences_squared, dim=2)
distances = distances.permute(1, 0)
return distances
@@ -107,26 +110,18 @@ def euclidean_distance_matrix(x, y, squared=False, epsilon=1e-10):
for tensor in [x, y]:
if tensor.ndim != 2:
raise ValueError(
"The tensor dimension must be two. You provide: tensor.ndim="
+ str(tensor.ndim)
+ "."
)
"The tensor dimension must be two. You provide: tensor.ndim=" +
str(tensor.ndim) + ".")
if not equal_int_shape([tuple(x.shape)[1]], [tuple(y.shape)[1]]):
raise ValueError(
"The vector shape must be equivalent in both tensors. You provide: tuple(y.shape)[1]="
+ str(tuple(x.shape)[1])
+ " and tuple(y.shape)(y)[1]="
+ str(tuple(y.shape)[1])
+ "."
)
+ str(tuple(x.shape)[1]) + " and tuple(y.shape)(y)[1]=" +
str(tuple(y.shape)[1]) + ".")
y = torch.transpose(y)
diss = (
torch.sum(x ** 2, axis=1, keepdims=True)
- 2 * torch.dot(x, y)
+ torch.sum(y ** 2, axis=0, keepdims=True)
)
diss = (torch.sum(x**2, axis=1, keepdims=True) - 2 * torch.dot(x, y) +
torch.sum(y**2, axis=0, keepdims=True))
if not squared:
if epsilon == 0:
@@ -173,19 +168,18 @@ def tangent_distance(signals, protos, subspaces, squared=False, epsilon=1e-10):
if subspaces.ndim == 2:
# clean solution without map if the matrix_scope is global
projectors = torch.eye(subspace_int_shape[-2]) - torch.dot(
subspaces, torch.transpose(subspaces)
)
subspaces, torch.transpose(subspaces))
projected_signals = torch.dot(signals, projectors)
projected_protos = torch.dot(protos, projectors)
diss = euclidean_distance_matrix(
projected_signals, projected_protos, squared=squared, epsilon=epsilon
)
diss = euclidean_distance_matrix(projected_signals,
projected_protos,
squared=squared,
epsilon=epsilon)
diss = torch.reshape(
diss, [signal_shape[0], signal_shape[2], proto_shape[0]]
)
diss, [signal_shape[0], signal_shape[2], proto_shape[0]])
return torch.permute(diss, [0, 2, 1])
@@ -193,21 +187,18 @@ def tangent_distance(signals, protos, subspaces, squared=False, epsilon=1e-10):
# no solution without map possible --> memory efficient but slow!
projectors = torch.eye(subspace_int_shape[-2]) - torch.bmm(
subspaces, subspaces
) # K.batch_dot(subspaces, subspaces, [2, 2])
subspaces,
subspaces) # K.batch_dot(subspaces, subspaces, [2, 2])
projected_protos = (
protos @ subspaces
).T # K.batch_dot(projectors, protos, [1, 1]))
projected_protos = (protos @ subspaces
).T # K.batch_dot(projectors, protos, [1, 1]))
def projected_norm(projector):
return torch.sum(torch.dot(signals, projector) ** 2, axis=1)
return torch.sum(torch.dot(signals, projector)**2, axis=1)
diss = (
torch.transpose(map(projected_norm, projectors))
- 2 * torch.dot(signals, projected_protos)
+ torch.sum(projected_protos ** 2, axis=0, keepdims=True)
)
diss = (torch.transpose(map(projected_norm, projectors)) -
2 * torch.dot(signals, projected_protos) +
torch.sum(projected_protos**2, axis=0, keepdims=True))
if not squared:
if epsilon == 0:
@@ -216,8 +207,7 @@ def tangent_distance(signals, protos, subspaces, squared=False, epsilon=1e-10):
diss = torch.sqrt(torch.max(diss, epsilon))
diss = torch.reshape(
diss, [signal_shape[0], signal_shape[2], proto_shape[0]]
)
diss, [signal_shape[0], signal_shape[2], proto_shape[0]])
return torch.permute(diss, [0, 2, 1])
@@ -233,12 +223,12 @@ def tangent_distance(signals, protos, subspaces, squared=False, epsilon=1e-10):
# Scope: Tangentspace Projections
diff = torch.reshape(
diff, (signal_shape[0] * signal_shape[2], signal_shape[1], -1)
)
diff, (signal_shape[0] * signal_shape[2], signal_shape[1], -1))
projected_diff = diff @ projectors
projected_diff = torch.reshape(
projected_diff,
(signal_shape[0], signal_shape[2], signal_shape[1]) + signal_shape[3:],
(signal_shape[0], signal_shape[2], signal_shape[1]) +
signal_shape[3:],
)
diss = torch.norm(projected_diff, 2, dim=-1)
@@ -251,13 +241,13 @@ def tangent_distance(signals, protos, subspaces, squared=False, epsilon=1e-10):
# Scope: Tangentspace Projections
diff = torch.reshape(
diff, (signal_shape[0] * signal_shape[2], signal_shape[1], -1)
)
diff, (signal_shape[0] * signal_shape[2], signal_shape[1], -1))
diff = diff.permute([1, 0, 2])
projected_diff = torch.bmm(diff, projectors)
projected_diff = torch.reshape(
projected_diff,
(signal_shape[1], signal_shape[0], signal_shape[2]) + signal_shape[3:],
(signal_shape[1], signal_shape[0], signal_shape[2]) +
signal_shape[3:],
)
diss = torch.norm(projected_diff, 2, dim=-1)

View File

@@ -1,6 +1,11 @@
import torch
def get_flat(*args):
rv = [x.view(x.size(0), -1) for x in args]
return rv
def calculate_prototype_accuracy(y_pred, y_true, plabels):
"""Computes the accuracy of a prototype based model.
via Winner-Takes-All rule.
@@ -23,7 +28,7 @@ def predict_label(y_pred, plabels):
def mixed_shape(inputs):
if not torch.is_tensor(inputs):
raise ValueError('Input must be a tensor.')
raise ValueError("Input must be a tensor.")
else:
int_shape = list(inputs.shape)
# sometimes int_shape returns mixed integer types
@@ -39,11 +44,11 @@ def mixed_shape(inputs):
def equal_int_shape(shape_1, shape_2):
if not isinstance(shape_1,
(tuple, list)) or not isinstance(shape_2, (tuple, list)):
raise ValueError('Input shapes must list or tuple.')
raise ValueError("Input shapes must list or tuple.")
for shape in [shape_1, shape_2]:
if not all([isinstance(x, int) or x is None for x in shape]):
raise ValueError(
'Input shapes must be list or tuple of int and None values.')
"Input shapes must be list or tuple of int and None values.")
if len(shape_1) != len(shape_2):
return False

View File

@@ -15,59 +15,59 @@ def register_initializer(function):
def labels_from(distribution, one_hot=True):
"""Takes a distribution tensor and returns a labels tensor."""
nclasses = distribution.shape[0]
llist = [[i] * n for i, n in zip(range(nclasses), distribution)]
num_classes = distribution.shape[0]
llist = [[i] * n for i, n in zip(range(num_classes), distribution)]
# labels = [l for cl in llist for l in cl] # flatten the list of lists
flat_llist = list(chain(*llist)) # flatten label list with itertools.chain
plabels = torch.tensor(flat_llist, requires_grad=False)
if one_hot:
return torch.eye(nclasses)[plabels]
return torch.eye(num_classes)[plabels]
return plabels
@register_initializer
def ones(x_train, y_train, prototype_distribution, one_hot=True):
nprotos = torch.sum(prototype_distribution)
protos = torch.ones(nprotos, *x_train.shape[1:])
num_protos = torch.sum(prototype_distribution)
protos = torch.ones(num_protos, *x_train.shape[1:])
plabels = labels_from(prototype_distribution, one_hot)
return protos, plabels
@register_initializer
def zeros(x_train, y_train, prototype_distribution, one_hot=True):
nprotos = torch.sum(prototype_distribution)
protos = torch.zeros(nprotos, *x_train.shape[1:])
num_protos = torch.sum(prototype_distribution)
protos = torch.zeros(num_protos, *x_train.shape[1:])
plabels = labels_from(prototype_distribution, one_hot)
return protos, plabels
@register_initializer
def rand(x_train, y_train, prototype_distribution, one_hot=True):
nprotos = torch.sum(prototype_distribution)
protos = torch.rand(nprotos, *x_train.shape[1:])
num_protos = torch.sum(prototype_distribution)
protos = torch.rand(num_protos, *x_train.shape[1:])
plabels = labels_from(prototype_distribution, one_hot)
return protos, plabels
@register_initializer
def randn(x_train, y_train, prototype_distribution, one_hot=True):
nprotos = torch.sum(prototype_distribution)
protos = torch.randn(nprotos, *x_train.shape[1:])
num_protos = torch.sum(prototype_distribution)
protos = torch.randn(num_protos, *x_train.shape[1:])
plabels = labels_from(prototype_distribution, one_hot)
return protos, plabels
@register_initializer
def stratified_mean(x_train, y_train, prototype_distribution, one_hot=True):
nprotos = torch.sum(prototype_distribution)
num_protos = torch.sum(prototype_distribution)
pdim = x_train.shape[1]
protos = torch.empty(nprotos, pdim)
protos = torch.empty(num_protos, pdim)
plabels = labels_from(prototype_distribution, one_hot)
for i, label in enumerate(plabels):
matcher = torch.eq(label.unsqueeze(dim=0), y_train)
if one_hot:
nclasses = y_train.size()[1]
matcher = torch.eq(torch.sum(matcher, dim=-1), nclasses)
num_classes = y_train.size()[1]
matcher = torch.eq(torch.sum(matcher, dim=-1), num_classes)
xl = x_train[matcher]
mean_xl = torch.mean(xl, dim=0)
protos[i] = mean_xl
@@ -81,15 +81,15 @@ def stratified_random(x_train,
prototype_distribution,
one_hot=True,
epsilon=1e-7):
nprotos = torch.sum(prototype_distribution)
num_protos = torch.sum(prototype_distribution)
pdim = x_train.shape[1]
protos = torch.empty(nprotos, pdim)
protos = torch.empty(num_protos, pdim)
plabels = labels_from(prototype_distribution, one_hot)
for i, label in enumerate(plabels):
matcher = torch.eq(label.unsqueeze(dim=0), y_train)
if one_hot:
nclasses = y_train.size()[1]
matcher = torch.eq(torch.sum(matcher, dim=-1), nclasses)
num_classes = y_train.size()[1]
matcher = torch.eq(torch.sum(matcher, dim=-1), num_classes)
xl = x_train[matcher]
rand_index = torch.zeros(1).long().random_(0, xl.shape[0] - 1)
random_xl = xl[rand_index]
@@ -104,4 +104,4 @@ def get_initializer(funcname):
return funcname
if funcname in INITIALIZERS:
return INITIALIZERS.get(funcname)
raise NameError(f'Initializer {funcname} was not found.')
raise NameError(f"Initializer {funcname} was not found.")

View File

@@ -3,20 +3,29 @@
import torch
def _get_dp_dm(distances, targets, plabels):
matcher = torch.eq(targets.unsqueeze(dim=1), plabels)
if plabels.ndim == 2:
def _get_matcher(targets, labels):
"""Returns a boolean tensor."""
matcher = torch.eq(targets.unsqueeze(dim=1), labels)
if labels.ndim == 2:
# if the labels are one-hot vectors
nclasses = targets.size()[1]
matcher = torch.eq(torch.sum(matcher, dim=-1), nclasses)
num_classes = targets.size()[1]
matcher = torch.eq(torch.sum(matcher, dim=-1), num_classes)
return matcher
def _get_dp_dm(distances, targets, plabels, with_indices=False):
"""Returns the d+ and d- values for a batch of distances."""
matcher = _get_matcher(targets, plabels)
not_matcher = torch.bitwise_not(matcher)
inf = torch.full_like(distances, fill_value=float('inf'))
inf = torch.full_like(distances, fill_value=float("inf"))
d_matching = torch.where(matcher, distances, inf)
d_unmatching = torch.where(not_matcher, distances, inf)
dp = torch.min(d_matching, dim=1, keepdim=True).values
dm = torch.min(d_unmatching, dim=1, keepdim=True).values
return dp, dm
dp = torch.min(d_matching, dim=-1, keepdim=True)
dm = torch.min(d_unmatching, dim=-1, keepdim=True)
if with_indices:
return dp, dm
return dp.values, dm.values
def glvq_loss(distances, target_labels, prototype_labels):
@@ -24,3 +33,27 @@ def glvq_loss(distances, target_labels, prototype_labels):
dp, dm = _get_dp_dm(distances, target_labels, prototype_labels)
mu = (dp - dm) / (dp + dm)
return mu
def lvq1_loss(distances, target_labels, prototype_labels):
"""LVQ1 loss function with support for one-hot labels.
See Section 4 [Sado&Yamada]
https://papers.nips.cc/paper/1995/file/9c3b1830513cc3b8fc4b76635d32e692-Paper.pdf
"""
dp, dm = _get_dp_dm(distances, target_labels, prototype_labels)
mu = dp
mu[dp > dm] = -dm[dp > dm]
return mu
def lvq21_loss(distances, target_labels, prototype_labels):
"""LVQ2.1 loss function with support for one-hot labels.
See Section 4 [Sado&Yamada]
https://papers.nips.cc/paper/1995/file/9c3b1830513cc3b8fc4b76635d32e692-Paper.pdf
"""
dp, dm = _get_dp_dm(distances, target_labels, prototype_labels)
mu = dp - dm
return mu

View File

@@ -1,7 +1,5 @@
# -*- coding: utf-8 -*-
from __future__ import print_function
from __future__ import absolute_import
from __future__ import division
from __future__ import absolute_import, division, print_function
import torch

View File

@@ -0,0 +1,18 @@
"""ProtoTorch similarity functions."""
import torch
def cosine_similarity(x, y):
"""Compute the cosine similarity between :math:`x` and :math:`y`.
Expected dimension of x is 2.
Expected dimension of y is 2.
"""
norm_x = x.pow(2).sum(1).sqrt()
norm_y = y.pow(2).sum(1).sqrt()
norm_mat = norm_x.unsqueeze(-1) @ norm_y.unsqueeze(-1).T
epsilon = torch.finfo(norm_mat.dtype).eps
norm_mat.clamp_(min=epsilon)
similarities = (x @ y.T) / norm_mat
return similarities

View File

@@ -3,5 +3,5 @@
from .prototypes import Prototypes1D
__all__ = [
'Prototypes1D',
"Prototypes1D",
]

View File

@@ -7,7 +7,7 @@ from prototorch.functions.losses import glvq_loss
class GLVQLoss(torch.nn.Module):
def __init__(self, margin=0.0, squashing='identity', beta=10, **kwargs):
def __init__(self, margin=0.0, squashing="identity", beta=10, **kwargs):
super().__init__(**kwargs)
self.margin = margin
self.squashing = get_activation(squashing)
@@ -18,3 +18,23 @@ class GLVQLoss(torch.nn.Module):
mu = glvq_loss(distances, targets, prototype_labels=plabels)
batch_loss = self.squashing(mu + self.margin, beta=self.beta)
return torch.sum(batch_loss, dim=0)
class NeuralGasEnergy(torch.nn.Module):
def __init__(self, lm):
super().__init__()
self.lm = lm
def forward(self, d):
order = torch.argsort(d, dim=1)
ranks = torch.argsort(order, dim=1)
cost = torch.sum(self._nghood_fn(ranks, self.lm) * d)
return cost, order
def extra_repr(self):
return f"lambda: {self.lm}"
@staticmethod
def _nghood_fn(rankings, lm):
return torch.exp(-rankings / lm)

View File

@@ -1,10 +1,8 @@
from torch import nn
import torch
from prototorch.modules.prototypes import Prototypes1D
from prototorch.functions.distances import tangent_distance, euclidean_distance_matrix
from prototorch.functions.distances import euclidean_distance_matrix
from prototorch.functions.normalization import orthogonalization
from prototorch.functions.helper import _check_shapes, _int_and_mixed_shape
from prototorch.modules.prototypes import Prototypes1D
from torch import nn
class GTLVQ(nn.Module):
r""" Generalized Tangent Learning Vector Quantization
@@ -71,50 +69,42 @@ class GTLVQ(nn.Module):
subspace_data=None,
prototype_data=None,
subspace_size=256,
tangent_projection_type='local',
tangent_projection_type="local",
prototypes_per_class=2,
feature_dim=256,
):
super(GTLVQ, self).__init__()
self.num_protos = num_classes * prototypes_per_class
self.num_protos_class = prototypes_per_class
self.subspace_size = feature_dim if subspace_size is None else subspace_size
self.feature_dim = feature_dim
self.num_classes = num_classes
self.cls = Prototypes1D(
input_dim=feature_dim,
prototypes_per_class=prototypes_per_class,
nclasses=num_classes,
prototype_initializer="stratified_mean",
data=prototype_data,
)
if subspace_data is None:
raise ValueError('Init Data must be specified!')
raise ValueError("Init Data must be specified!")
self.tpt = tangent_projection_type
with torch.no_grad():
if self.tpt == 'local' or self.tpt == 'local_proj':
self.init_local_subspace(subspace_data)
elif self.tpt == 'global':
if self.tpt == "local":
self.init_local_subspace(subspace_data, subspace_size,
self.num_protos)
elif self.tpt == "global":
self.init_gobal_subspace(subspace_data, subspace_size)
else:
self.subspaces = None
# Hypothesis-Margin-Classifier
self.cls = Prototypes1D(input_dim=feature_dim,
prototypes_per_class=prototypes_per_class,
nclasses=num_classes,
prototype_initializer='stratified_mean',
data=prototype_data)
def forward(self, x):
# Tangent Projection
if self.tpt == 'local_proj':
x_conform = x.unsqueeze(1).repeat_interleave(self.num_protos,
1).unsqueeze(2)
dis, proj_x = self.local_tangent_projection(x_conform)
proj_x = proj_x.reshape(x.shape[0] * self.num_protos,
self.feature_dim)
return proj_x, dis
elif self.tpt == "local":
x_conform = x.unsqueeze(1).repeat_interleave(self.num_protos,
1).unsqueeze(2)
dis = tangent_distance(x_conform, self.cls.prototypes,
self.subspaces)
if self.tpt == "local":
dis = self.local_tangent_distances(x)
elif self.tpt == "gloabl":
dis = self.global_tangent_distances(x)
else:
@@ -127,51 +117,39 @@ class GTLVQ(nn.Module):
_, _, v = torch.svd(data)
subspace = (torch.eye(v.shape[0]) - (v @ v.T)).T
subspaces = subspace[:, :num_subspaces]
self.subspaces = torch.nn.Parameter(
subspaces).clone().detach().requires_grad_(True)
self.subspaces = nn.Parameter(subspaces, requires_grad=True)
def init_local_subspace(self, data):
_, _, v = torch.svd(data)
inital_projector = (torch.eye(v.shape[0]) - (v @ v.T)).T
subspaces = inital_projector.unsqueeze(0).repeat_interleave(
self.num_protos, 0)
self.subspaces = torch.nn.Parameter(
subspaces).clone().detach().requires_grad_(True)
def init_local_subspace(self, data,num_subspaces,num_protos):
data = data - torch.mean(data,dim=0)
_,_,v = torch.svd(data,some=False)
v = v[:,:num_subspaces]
subspaces = v.unsqueeze(0).repeat_interleave(num_protos,0)
self.subspaces = nn.Parameter(subspaces,requires_grad=True)
def global_tangent_distances(self, x):
# Tangent Projection
x, projected_prototypes = x @ self.subspaces, self.cls.prototypes @ self.subspaces
x, projected_prototypes = (
x @ self.subspaces,
self.cls.prototypes @ self.subspaces,
)
# Euclidean Distance
return euclidean_distance_matrix(x, projected_prototypes)
def local_tangent_projection(self,
signals):
# Note: subspaces is always assumed as transposed and must be orthogonal!
# shape(signals): batch x proto_number x channels x dim1 x dim2 x ... x dimN
# shape(protos): proto_number x dim1 x dim2 x ... x dimN
# shape(subspaces): (optional [proto_number]) x prod(dim1 * dim2 * ... * dimN) x prod(projected_atom_shape)
# subspace should be orthogonalized
# Origin Source Code
# Origin Author:
protos = self.cls.prototypes
subspaces = self.subspaces
signal_shape, signal_int_shape = _int_and_mixed_shape(signals)
_, proto_int_shape = _int_and_mixed_shape(protos)
def local_tangent_distances(self, x):
# check if the shapes are correct
_check_shapes(signal_int_shape, proto_int_shape)
# Tangent Data Projections
projected_protos = torch.bmm(protos.unsqueeze(1), subspaces).squeeze(1)
data = signals.squeeze(2).permute([1, 0, 2])
projected_data = torch.bmm(data, subspaces)
projected_data = projected_data.permute([1, 0, 2]).unsqueeze(1)
diff = projected_data - projected_protos
projected_diff = torch.reshape(
diff, (signal_shape[1], signal_shape[0], signal_shape[2]) +
signal_shape[3:])
diss = torch.norm(projected_diff, 2, dim=-1)
return diss.permute([1, 0, 2]).squeeze(-1), projected_data.squeeze(1)
# Tangent Distance
x = x.unsqueeze(1).expand(x.size(0), self.cls.prototypes.size(0),
x.size(-1))
protos = self.cls.prototypes.unsqueeze(0).expand(
x.size(0), self.cls.prototypes.size(0), x.size(-1))
projectors = torch.eye(
self.subspaces.shape[-2], device=x.device) - torch.bmm(
self.subspaces, self.subspaces.permute([0, 2, 1]))
diff = (x - protos)
diff = diff.permute([1, 0, 2])
diff = torch.bmm(diff, projectors)
diff = torch.norm(diff,2,dim=-1).T
return diff
def get_parameters(self):
return {
@@ -183,8 +161,7 @@ class GTLVQ(nn.Module):
def orthogonalize_subspace(self):
if self.subspaces is not None:
with torch.no_grad():
ortho_subpsaces = orthogonalization(
self.subspaces
) if self.tpt == 'global' else torch.nn.init.orthogonal_(
self.subspaces)
ortho_subpsaces = (orthogonalization(self.subspaces)
if self.tpt == "global" else
torch.nn.init.orthogonal_(self.subspaces))
self.subspaces.copy_(ortho_subpsaces)

View File

@@ -3,7 +3,6 @@
import warnings
import torch
from prototorch.functions.initializers import get_initializer
@@ -29,14 +28,19 @@ class Prototypes1D(_Prototypes):
TODO Complete this doc-string.
"""
def __init__(self,
prototypes_per_class=1,
prototype_initializer="ones",
prototype_distribution=None,
data=None,
dtype=torch.float32,
one_hot_labels=False,
**kwargs):
def __init__(
self,
prototypes_per_class=1,
prototype_initializer="ones",
prototype_distribution=None,
data=None,
dtype=torch.float32,
one_hot_labels=False,
**kwargs,
):
warnings.warn(
PendingDeprecationWarning(
"Prototypes1D will be replaced in future versions."))
# Convert tensors to python lists before processing
if prototype_distribution is not None:
@@ -48,13 +52,13 @@ class Prototypes1D(_Prototypes):
raise NameError("`input_dim` required if "
"no `data` is provided.")
if prototype_distribution:
kwargs_nclasses = sum(prototype_distribution)
kwargs_num_classes = sum(prototype_distribution)
else:
if "nclasses" not in kwargs:
if "num_classes" not in kwargs:
raise NameError("`prototype_distribution` required if "
"both `data` and `nclasses` are not "
"both `data` and `num_classes` are not "
"provided.")
kwargs_nclasses = kwargs.pop("nclasses")
kwargs_num_classes = kwargs.pop("num_classes")
input_dim = kwargs.pop("input_dim")
if prototype_initializer in [
"stratified_mean", "stratified_random"
@@ -63,18 +67,18 @@ class Prototypes1D(_Prototypes):
f"`prototype_initializer`: `{prototype_initializer}` "
"requires `data`, but `data` is not provided. "
"Using randomly generated data instead.")
x_train = torch.rand(kwargs_nclasses, input_dim)
y_train = torch.arange(kwargs_nclasses)
x_train = torch.rand(kwargs_num_classes, input_dim)
y_train = torch.arange(kwargs_num_classes)
if one_hot_labels:
y_train = torch.eye(kwargs_nclasses)[y_train]
y_train = torch.eye(kwargs_num_classes)[y_train]
data = [x_train, y_train]
x_train, y_train = data
x_train = torch.as_tensor(x_train).type(dtype)
y_train = torch.as_tensor(y_train).type(torch.int)
nclasses = torch.unique(y_train, dim=-1).shape[-1]
num_classes = torch.unique(y_train, dim=-1).shape[-1]
if nclasses == 1:
if num_classes == 1:
warnings.warn("Are you sure about having one class only?")
if x_train.ndim != 2:
@@ -100,19 +104,20 @@ class Prototypes1D(_Prototypes):
"not match data dimension "
f"`data[0].shape[1]`={x_train.shape[1]}")
# Verify the number of classes if `nclasses` is provided
if "nclasses" in kwargs:
kwargs_nclasses = kwargs.pop("nclasses")
if kwargs_nclasses != nclasses:
raise ValueError(f"Provided `nclasses={kwargs_nclasses}` does "
"not match data labels "
"`torch.unique(data[1]).shape[0]`"
f"={nclasses}")
# Verify the number of classes if `num_classes` is provided
if "num_classes" in kwargs:
kwargs_num_classes = kwargs.pop("num_classes")
if kwargs_num_classes != num_classes:
raise ValueError(
f"Provided `num_classes={kwargs_num_classes}` does "
"not match data labels "
"`torch.unique(data[1]).shape[0]`"
f"={num_classes}")
super().__init__(**kwargs)
if not prototype_distribution:
prototype_distribution = [prototypes_per_class] * nclasses
prototype_distribution = [prototypes_per_class] * num_classes
with torch.no_grad():
self.prototype_distribution = torch.tensor(prototype_distribution)

View File

@@ -1 +0,0 @@
from .colors import color_scheme, get_legend_handles

View File

@@ -0,0 +1,46 @@
"""Easy matplotlib animation. From https://github.com/jwkvam/celluloid."""
from collections import defaultdict
from typing import Dict, List
from matplotlib.animation import ArtistAnimation
from matplotlib.artist import Artist
from matplotlib.figure import Figure
__version__ = "0.2.0"
class Camera:
"""Make animations easier."""
def __init__(self, figure: Figure) -> None:
"""Create camera from matplotlib figure."""
self._figure = figure
# need to keep track off artists for each axis
self._offsets: Dict[str, Dict[int, int]] = {
k: defaultdict(int)
for k in
["collections", "patches", "lines", "texts", "artists", "images"]
}
self._photos: List[List[Artist]] = []
def snap(self) -> List[Artist]:
"""Capture current state of the figure."""
frame_artists: List[Artist] = []
for i, axis in enumerate(self._figure.axes):
if axis.legend_ is not None:
axis.add_artist(axis.legend_)
for name in self._offsets:
new_artists = getattr(axis, name)[self._offsets[name][i]:]
frame_artists += new_artists
self._offsets[name][i] += len(new_artists)
self._photos.append(frame_artists)
return frame_artists
def animate(self, *args, **kwargs) -> ArtistAnimation:
"""Animate the snapshots taken.
Uses matplotlib.animation.ArtistAnimation
Returns
-------
ArtistAnimation
"""
return ArtistAnimation(self._figure, self._photos, *args, **kwargs)

View File

@@ -1,13 +1,14 @@
"""ProtoFlow color utilities."""
from matplotlib import cm
from matplotlib.colors import Normalize
from matplotlib.colors import to_hex
from matplotlib.colors import to_rgb
import matplotlib.lines as mlines
from matplotlib import cm
from matplotlib.colors import Normalize, to_hex, to_rgb
def color_scheme(n, cmap="viridis", form="hex", tikz=False,
def color_scheme(n,
cmap="viridis",
form="hex",
tikz=False,
zero_indexed=False):
"""Return *n* colors from the color scheme.
@@ -57,13 +58,16 @@ def get_legend_handles(labels, marker="dots", zero_indexed=False):
zero_indexed=zero_indexed)
for label, color in zip(labels, colors.values()):
if marker == "dots":
handle = mlines.Line2D([], [],
color="white",
markerfacecolor=color,
marker="o",
markersize=10,
markeredgecolor="k",
label=label)
handle = mlines.Line2D(
[],
[],
color="white",
markerfacecolor=color,
marker="o",
markersize=10,
markeredgecolor="k",
label=label,
)
else:
handle = mlines.Line2D([], [],
color=color,

243
prototorch/utils/utils.py Normal file
View File

@@ -0,0 +1,243 @@
"""Utilities that provide various small functionalities."""
import os
import pickle
import sys
from time import time
import matplotlib.pyplot as plt
import numpy as np
def progressbar(title, value, end, bar_width=20):
percent = float(value) / end
arrow = "=" * int(round(percent * bar_width) - 1) + ">"
spaces = "." * (bar_width - len(arrow))
sys.stdout.write("\r{}: [{}] {}%".format(title, arrow + spaces,
int(round(percent * 100))))
sys.stdout.flush()
if percent == 1.0:
print()
def prettify_string(inputs, start="", sep=" ", end="\n"):
outputs = start + " ".join(inputs.split()) + end
return outputs
def pretty_print(inputs):
print(prettify_string(inputs))
def writelog(self, *logs, logdir="./logs", logfile="run.txt"):
f = os.path.join(logdir, logfile)
with open(f, "a+") as fh:
for log in logs:
fh.write(log)
fh.write("\n")
def start_tensorboard(self, logdir="./logs"):
cmd = f"tensorboard --logdir={logdir} --port=6006"
os.system(cmd)
def make_directory(save_dir):
if not os.path.exists(save_dir):
print(f"Making directory {save_dir}.")
os.mkdir(save_dir)
def make_gif(filenames, duration, output_file=None):
try:
import imageio
except ModuleNotFoundError as e:
print("Please install Protoflow with [other] extra requirements.")
raise (e)
images = list()
for filename in filenames:
images.append(imageio.imread(filename))
if not output_file:
output_file = f"makegif.gif"
if images:
imageio.mimwrite(output_file, images, duration=duration)
def gif_from_dir(directory,
duration,
prefix="",
output_file=None,
verbose=True):
images = os.listdir(directory)
if verbose:
print(f"Making gif from {len(images)} images under {directory}.")
filenames = list()
# Sort images
images = sorted(
images,
key=lambda img: int(os.path.splitext(img)[0].replace(prefix, "")))
for image in images:
fname = os.path.join(directory, image)
filenames.append(fname)
if not output_file:
output_file = os.path.join(directory, "makegif.gif")
make_gif(filenames=filenames, duration=duration, output_file=output_file)
def accuracy_score(y_true, y_pred):
accuracy = np.sum(y_true == y_pred)
normalized_acc = accuracy / float(len(y_true))
return normalized_acc
def predict_and_score(clf,
x_test,
y_test,
verbose=False,
title="Test accuracy"):
y_pred = clf.predict(x_test)
accuracy = np.sum(y_test == y_pred)
normalized_acc = accuracy / float(len(y_test))
if verbose:
print(f"{title}: {normalized_acc * 100:06.04f}%")
return normalized_acc
def remove_nan_rows(arr):
"""Remove all rows with `nan` values in `arr`."""
mask = np.isnan(arr).any(axis=1)
return arr[~mask]
def remove_nan_cols(arr):
"""Remove all columns with `nan` values in `arr`."""
mask = np.isnan(arr).any(axis=0)
return arr[~mask]
def replace_in(arr, replacement_dict, inplace=False):
"""Replace the keys found in `arr` with the values from
the `replacement_dict`.
"""
if inplace:
new_arr = arr
else:
import copy
new_arr = copy.deepcopy(arr)
for k, v in replacement_dict.items():
new_arr[arr == k] = v
return new_arr
def train_test_split(data, train=0.7, val=0.15, shuffle=None, return_xy=False):
"""Split a classification dataset in such a way so as to
preserve the class distribution in subsamples of the dataset.
"""
if train + val > 1.0:
raise ValueError("Invalid split values for train and val.")
Y = data[:, -1]
labels = set(Y)
hist = dict()
for l in labels:
data_l = data[Y == l]
nl = len(data_l)
nl_train = int(nl * train)
nl_val = int(nl * val)
nl_test = nl - (nl_train + nl_val)
hist[l] = (nl_train, nl_val, nl_test)
train_data = list()
val_data = list()
test_data = list()
for l, (nl_train, nl_val, nl_test) in hist.items():
data_l = data[Y == l]
if shuffle:
np.random.shuffle(data_l)
train_l = data_l[:nl_train]
val_l = data_l[nl_train:nl_train + nl_val]
test_l = data_l[nl_train + nl_val:nl_train + nl_val + nl_test]
train_data.append(train_l)
val_data.append(val_l)
test_data.append(test_l)
def _squash(data_list):
data = np.array(data_list[0])
for item in data_list[1:]:
data = np.vstack((data, np.array(item)))
return data
train_data = _squash(train_data)
if val_data:
val_data = _squash(val_data)
if test_data:
test_data = _squash(test_data)
if return_xy:
x_train = train_data[:, :-1]
y_train = train_data[:, -1]
x_val = val_data[:, :-1]
y_val = val_data[:, -1]
x_test = test_data[:, :-1]
y_test = test_data[:, -1]
return (x_train, y_train), (x_val, y_val), (x_test, y_test)
return train_data, val_data, test_data
def class_histogram(data, title="Untitled"):
plt.figure(title)
plt.clf()
plt.title(title)
dist, counts = np.unique(data[:, -1], return_counts=True)
plt.bar(dist, counts)
plt.xticks(dist)
print("Call matplotlib.pyplot.show() to see the plot.")
def ntimer(n=10):
"""Wraps a function which wraps another function to time it."""
if n < 1:
raise (Exception(f"Invalid n = {n} given."))
def timer(func):
"""Wraps `func` with a timer and returns the wrapped `func`."""
def wrapper(*args, **kwargs):
rv = None
before = time()
for _ in range(n):
rv = func(*args, **kwargs)
after = time()
elapsed = after - before
print(f"Elapsed: {elapsed*1e3:02.02f} ms")
return rv
return wrapper
return timer
def memoize(verbose=True):
"""Wraps a function which wraps another function that memoizes."""
def memoizer(func):
"""Memoize (cache) return values of `func`.
Wraps `func` and returns the wrapped `func` so that `func`
is executed when the results are not available in the cache.
"""
cache = {}
def wrapper(*args, **kwargs):
t = (pickle.dumps(args), pickle.dumps(kwargs))
if t not in cache:
if verbose:
print(f"Adding NEW rv {func.__name__}{args}{kwargs} "
"to cache.")
cache[t] = func(*args, **kwargs)
else:
if verbose:
print(f"Using OLD rv {func.__name__}{args}{kwargs} "
"from cache.")
return cache[t]
return wrapper
return memoizer

View File

@@ -1,5 +0,0 @@
matplotlib==3.1.2
pytest==5.3.4
requests==2.22.0
codecov==2.0.22
tqdm==4.44.1

View File

@@ -1,15 +1,14 @@
"""
_____ _ _______ _
| __ \ | | |__ __| | |
_____ _ _______ _
| __ \ | | |__ __| | |
| |__) | __ ___ | |_ ___ | | ___ _ __ ___| |__
| ___/ '__/ _ \| __/ _ \| |/ _ \| '__/ __| '_ \
| ___/ '__/ _ \| __/ _ \| |/ _ \| '__/ __| '_ \
| | | | | (_) | || (_) | | (_) | | | (__| | | |
|_| |_| \___/ \__\___/|_|\___/|_| \___|_| |_|
ProtoTorch Core Package
"""
from setuptools import setup
from setuptools import find_packages
from setuptools import find_packages, setup
PROJECT_URL = "https://github.com/si-cim/prototorch"
DOWNLOAD_URL = "https://github.com/si-cim/prototorch.git"
@@ -21,28 +20,30 @@ INSTALL_REQUIRES = [
"torch>=1.3.1",
"torchvision>=0.5.0",
"numpy>=1.9.1",
]
DOCS = [
"recommonmark",
"sphinx",
"sphinx_rtd_theme",
"sphinxcontrib-katex",
"sklearn",
]
DATASETS = [
"requests",
"tqdm",
]
DEV = ["bumpversion"]
DOCS = [
"recommonmark",
"sphinx",
"sphinx_rtd_theme",
"sphinxcontrib-katex",
"sphinx-autodoc-typehints",
]
EXAMPLES = [
"sklearn",
"matplotlib",
"torchinfo",
]
TESTS = ["pytest"]
ALL = DOCS + DATASETS + EXAMPLES + TESTS
TESTS = ["codecov", "pytest"]
ALL = DATASETS + DEV + DOCS + EXAMPLES + TESTS
setup(
name="prototorch",
version="0.3.0-dev0",
version="0.4.5",
description="Highly extensible, GPU-supported "
"Learning Vector Quantization (LVQ) toolbox "
"built using PyTorch and its nn API.",
@@ -72,6 +73,7 @@ setup(
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Operating System :: OS Independent",
"Topic :: Scientific/Engineering :: Artificial Intelligence",
"Topic :: Software Development :: Libraries",

25
tests/test_components.py Normal file
View File

@@ -0,0 +1,25 @@
"""ProtoTorch components test suite."""
import prototorch as pt
import torch
def test_labcomps_zeros_init():
protos = torch.zeros(3, 2)
c = pt.components.LabeledComponents(
distribution=[1, 1, 1],
initializer=pt.components.Zeros(2),
)
assert (c.components == protos).any() == True
def test_labcomps_warmstart():
protos = torch.randn(3, 2)
plabels = torch.tensor([1, 2, 3])
c = pt.components.LabeledComponents(
distribution=[1, 1, 1],
initializer=None,
initialized_components=[protos, plabels],
)
assert (c.components == protos).any() == True
assert (c.component_labels == plabels).any() == True

View File

@@ -12,26 +12,26 @@ from prototorch.datasets import abstract, tecator
class TestAbstract(unittest.TestCase):
def test_getitem(self):
with self.assertRaises(NotImplementedError):
abstract.Dataset('./artifacts')[0]
abstract.Dataset("./artifacts")[0]
def test_len(self):
with self.assertRaises(NotImplementedError):
len(abstract.Dataset('./artifacts'))
len(abstract.Dataset("./artifacts"))
class TestProtoDataset(unittest.TestCase):
def test_getitem(self):
with self.assertRaises(NotImplementedError):
abstract.ProtoDataset('./artifacts')[0]
abstract.ProtoDataset("./artifacts")[0]
def test_download(self):
with self.assertRaises(NotImplementedError):
abstract.ProtoDataset('./artifacts').download()
abstract.ProtoDataset("./artifacts").download()
class TestTecator(unittest.TestCase):
def setUp(self):
self.artifacts_dir = './artifacts/Tecator'
self.artifacts_dir = "./artifacts/Tecator"
self._remove_artifacts()
def _remove_artifacts(self):
@@ -39,23 +39,23 @@ class TestTecator(unittest.TestCase):
shutil.rmtree(self.artifacts_dir)
def test_download_false(self):
rootdir = self.artifacts_dir.rpartition('/')[0]
rootdir = self.artifacts_dir.rpartition("/")[0]
self._remove_artifacts()
with self.assertRaises(RuntimeError):
_ = tecator.Tecator(rootdir, download=False)
def test_download_caching(self):
rootdir = self.artifacts_dir.rpartition('/')[0]
rootdir = self.artifacts_dir.rpartition("/")[0]
_ = tecator.Tecator(rootdir, download=True, verbose=False)
_ = tecator.Tecator(rootdir, download=False, verbose=False)
def test_repr(self):
rootdir = self.artifacts_dir.rpartition('/')[0]
rootdir = self.artifacts_dir.rpartition("/")[0]
train = tecator.Tecator(rootdir, download=True, verbose=True)
self.assertTrue('Split: Train' in train.__repr__())
self.assertTrue("Split: Train" in train.__repr__())
def test_download_train(self):
rootdir = self.artifacts_dir.rpartition('/')[0]
rootdir = self.artifacts_dir.rpartition("/")[0]
train = tecator.Tecator(root=rootdir,
train=True,
download=True,
@@ -67,7 +67,7 @@ class TestTecator(unittest.TestCase):
self.assertEqual(x_train.shape[1], 100)
def test_download_test(self):
rootdir = self.artifacts_dir.rpartition('/')[0]
rootdir = self.artifacts_dir.rpartition("/")[0]
test = tecator.Tecator(root=rootdir, train=False, verbose=False)
x_test, y_test = test.data, test.targets
self.assertEqual(x_test.shape[0], 71)
@@ -75,19 +75,19 @@ class TestTecator(unittest.TestCase):
self.assertEqual(x_test.shape[1], 100)
def test_class_to_idx(self):
rootdir = self.artifacts_dir.rpartition('/')[0]
rootdir = self.artifacts_dir.rpartition("/")[0]
test = tecator.Tecator(root=rootdir, train=False, verbose=False)
_ = test.class_to_idx
def test_getitem(self):
rootdir = self.artifacts_dir.rpartition('/')[0]
rootdir = self.artifacts_dir.rpartition("/")[0]
test = tecator.Tecator(root=rootdir, train=False, verbose=False)
x, y = test[0]
self.assertEqual(x.shape[0], 100)
self.assertIsInstance(y, int)
def test_loadable_with_dataloader(self):
rootdir = self.artifacts_dir.rpartition('/')[0]
rootdir = self.artifacts_dir.rpartition("/")[0]
test = tecator.Tecator(root=rootdir, train=False, verbose=False)
_ = torch.utils.data.DataLoader(test, batch_size=64, shuffle=True)

View File

@@ -4,14 +4,13 @@ import unittest
import numpy as np
import torch
from prototorch.functions import (activations, competitions, distances,
initializers, losses)
class TestActivations(unittest.TestCase):
def setUp(self):
self.flist = ['identity', 'sigmoid_beta', 'swish_beta']
self.flist = ["identity", "sigmoid_beta", "swish_beta"]
self.x = torch.randn(1024, 1)
def test_registry(self):
@@ -39,7 +38,7 @@ class TestActivations(unittest.TestCase):
self.assertEqual(1, f(1))
def test_unknown_deserialization(self):
for funcname in ['blubb', 'foobar']:
for funcname in ["blubb", "foobar"]:
with self.assertRaises(NameError):
_ = activations.get_activation(funcname)
@@ -52,7 +51,7 @@ class TestActivations(unittest.TestCase):
self.assertIsNone(mismatch)
def test_sigmoid_beta1(self):
actual = activations.sigmoid_beta(self.x, beta=torch.tensor(1))
actual = activations.sigmoid_beta(self.x, beta=1.0)
desired = torch.sigmoid(self.x)
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
@@ -60,7 +59,7 @@ class TestActivations(unittest.TestCase):
self.assertIsNone(mismatch)
def test_swish_beta1(self):
actual = activations.swish_beta(self.x, beta=torch.tensor(1))
actual = activations.swish_beta(self.x, beta=1.0)
desired = self.x * torch.sigmoid(self.x)
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
@@ -76,7 +75,7 @@ class TestCompetitions(unittest.TestCase):
pass
def test_wtac(self):
d = torch.tensor([[2., 3., 1.99, 3.01], [2., 3., 2.01, 3.]])
d = torch.tensor([[2.0, 3.0, 1.99, 3.01], [2.0, 3.0, 2.01, 3.0]])
labels = torch.tensor([0, 1, 2, 3])
actual = competitions.wtac(d, labels)
desired = torch.tensor([2, 0])
@@ -86,7 +85,7 @@ class TestCompetitions(unittest.TestCase):
self.assertIsNone(mismatch)
def test_wtac_unequal_dist(self):
d = torch.tensor([[2., 3., 4.], [2., 3., 1.]])
d = torch.tensor([[2.0, 3.0, 4.0], [2.0, 3.0, 1.0]])
labels = torch.tensor([0, 1, 1])
actual = competitions.wtac(d, labels)
desired = torch.tensor([0, 1])
@@ -96,7 +95,7 @@ class TestCompetitions(unittest.TestCase):
self.assertIsNone(mismatch)
def test_wtac_one_hot(self):
d = torch.tensor([[1.99, 3.01], [3., 2.01]])
d = torch.tensor([[1.99, 3.01], [3.0, 2.01]])
labels = torch.tensor([[0, 1], [1, 0]])
actual = competitions.wtac(d, labels)
desired = torch.tensor([[0, 1], [1, 0]])
@@ -106,40 +105,40 @@ class TestCompetitions(unittest.TestCase):
self.assertIsNone(mismatch)
def test_stratified_min(self):
d = torch.tensor([[1., 0., 2., 3.], [9., 8., 0, 1]])
d = torch.tensor([[1.0, 0.0, 2.0, 3.0], [9.0, 8.0, 0, 1]])
labels = torch.tensor([0, 0, 1, 2])
actual = competitions.stratified_min(d, labels)
desired = torch.tensor([[0., 2., 3.], [8., 0., 1.]])
desired = torch.tensor([[0.0, 2.0, 3.0], [8.0, 0.0, 1.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
self.assertIsNone(mismatch)
def test_stratified_min_one_hot(self):
d = torch.tensor([[1., 0., 2., 3.], [9., 8., 0, 1]])
d = torch.tensor([[1.0, 0.0, 2.0, 3.0], [9.0, 8.0, 0, 1]])
labels = torch.tensor([0, 0, 1, 2])
labels = torch.eye(3)[labels]
actual = competitions.stratified_min(d, labels)
desired = torch.tensor([[0., 2., 3.], [8., 0., 1.]])
desired = torch.tensor([[0.0, 2.0, 3.0], [8.0, 0.0, 1.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
self.assertIsNone(mismatch)
def test_stratified_min_simple(self):
d = torch.tensor([[0., 2., 3.], [8., 0, 1]])
d = torch.tensor([[0.0, 2.0, 3.0], [8.0, 0, 1]])
labels = torch.tensor([0, 1, 2])
actual = competitions.stratified_min(d, labels)
desired = torch.tensor([[0., 2., 3.], [8., 0., 1.]])
desired = torch.tensor([[0.0, 2.0, 3.0], [8.0, 0.0, 1.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
self.assertIsNone(mismatch)
def test_knnc_k1(self):
d = torch.tensor([[2., 3., 1.99, 3.01], [2., 3., 2.01, 3.]])
d = torch.tensor([[2.0, 3.0, 1.99, 3.01], [2.0, 3.0, 2.01, 3.0]])
labels = torch.tensor([0, 1, 2, 3])
actual = competitions.knnc(d, labels, k=torch.tensor([1]))
actual = competitions.knnc(d, labels, k=1)
desired = torch.tensor([2, 0])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
@@ -194,12 +193,12 @@ class TestDistances(unittest.TestCase):
desired = torch.empty(self.nx, self.ny)
for i in range(self.nx):
for j in range(self.ny):
desired[i][j] = torch.nn.functional.pairwise_distance(
desired[i][j] = (torch.nn.functional.pairwise_distance(
self.x[i].reshape(1, -1),
self.y[j].reshape(1, -1),
p=2,
keepdim=False,
)**2
)**2)
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=2)
@@ -254,14 +253,14 @@ class TestDistances(unittest.TestCase):
self.assertIsNone(mismatch)
def test_lpnorm_pinf(self):
actual = distances.lpnorm_distance(self.x, self.y, p=float('inf'))
actual = distances.lpnorm_distance(self.x, self.y, p=float("inf"))
desired = torch.empty(self.nx, self.ny)
for i in range(self.nx):
for j in range(self.ny):
desired[i][j] = torch.nn.functional.pairwise_distance(
self.x[i].reshape(1, -1),
self.y[j].reshape(1, -1),
p=float('inf'),
p=float("inf"),
keepdim=False,
)
mismatch = np.testing.assert_array_almost_equal(actual,
@@ -275,12 +274,12 @@ class TestDistances(unittest.TestCase):
desired = torch.empty(self.nx, self.ny)
for i in range(self.nx):
for j in range(self.ny):
desired[i][j] = torch.nn.functional.pairwise_distance(
desired[i][j] = (torch.nn.functional.pairwise_distance(
self.x[i].reshape(1, -1),
self.y[j].reshape(1, -1),
p=2,
keepdim=False,
)**2
)**2)
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=2)
@@ -293,12 +292,12 @@ class TestDistances(unittest.TestCase):
desired = torch.empty(self.nx, self.ny)
for i in range(self.nx):
for j in range(self.ny):
desired[i][j] = torch.nn.functional.pairwise_distance(
desired[i][j] = (torch.nn.functional.pairwise_distance(
self.x[i].reshape(1, -1),
self.y[j].reshape(1, -1),
p=2,
keepdim=False,
)**2
)**2)
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=2)
@@ -311,8 +310,12 @@ class TestDistances(unittest.TestCase):
class TestInitializers(unittest.TestCase):
def setUp(self):
self.flist = [
'zeros', 'ones', 'rand', 'randn', 'stratified_mean',
'stratified_random'
"zeros",
"ones",
"rand",
"randn",
"stratified_mean",
"stratified_random",
]
self.x = torch.tensor(
[[0, -1, -2], [10, 11, 12], [0, 0, 0], [2, 2, 2]],
@@ -340,7 +343,7 @@ class TestInitializers(unittest.TestCase):
self.assertEqual(1, f(1))
def test_unknown_deserialization(self):
for funcname in ['blubb', 'foobar']:
for funcname in ["blubb", "foobar"]:
with self.assertRaises(NameError):
_ = initializers.get_initializer(funcname)
@@ -383,7 +386,7 @@ class TestInitializers(unittest.TestCase):
def test_stratified_mean_equal1(self):
pdist = torch.tensor([1, 1])
actual, _ = initializers.stratified_mean(self.x, self.y, pdist, False)
desired = torch.tensor([[5., 5., 5.], [1., 1., 1.]])
desired = torch.tensor([[5.0, 5.0, 5.0], [1.0, 1.0, 1.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
@@ -393,7 +396,7 @@ class TestInitializers(unittest.TestCase):
pdist = torch.tensor([1, 1])
actual, _ = initializers.stratified_random(self.x, self.y, pdist,
False)
desired = torch.tensor([[0., -1., -2.], [0., 0., 0.]])
desired = torch.tensor([[0.0, -1.0, -2.0], [0.0, 0.0, 0.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
@@ -402,8 +405,8 @@ class TestInitializers(unittest.TestCase):
def test_stratified_mean_equal2(self):
pdist = torch.tensor([2, 2])
actual, _ = initializers.stratified_mean(self.x, self.y, pdist, False)
desired = torch.tensor([[5., 5., 5.], [5., 5., 5.], [1., 1., 1.],
[1., 1., 1.]])
desired = torch.tensor([[5.0, 5.0, 5.0], [5.0, 5.0, 5.0],
[1.0, 1.0, 1.0], [1.0, 1.0, 1.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
@@ -413,8 +416,8 @@ class TestInitializers(unittest.TestCase):
pdist = torch.tensor([2, 2])
actual, _ = initializers.stratified_random(self.x, self.y, pdist,
False)
desired = torch.tensor([[0., -1., -2.], [0., -1., -2.], [0., 0., 0.],
[0., 0., 0.]])
desired = torch.tensor([[0.0, -1.0, -2.0], [0.0, -1.0, -2.0],
[0.0, 0.0, 0.0], [0.0, 0.0, 0.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
@@ -423,8 +426,8 @@ class TestInitializers(unittest.TestCase):
def test_stratified_mean_unequal(self):
pdist = torch.tensor([1, 3])
actual, _ = initializers.stratified_mean(self.x, self.y, pdist, False)
desired = torch.tensor([[5., 5., 5.], [1., 1., 1.], [1., 1., 1.],
[1., 1., 1.]])
desired = torch.tensor([[5.0, 5.0, 5.0], [1.0, 1.0, 1.0],
[1.0, 1.0, 1.0], [1.0, 1.0, 1.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
@@ -434,8 +437,8 @@ class TestInitializers(unittest.TestCase):
pdist = torch.tensor([1, 3])
actual, _ = initializers.stratified_random(self.x, self.y, pdist,
False)
desired = torch.tensor([[0., -1., -2.], [0., 0., 0.], [0., 0., 0.],
[0., 0., 0.]])
desired = torch.tensor([[0.0, -1.0, -2.0], [0.0, 0.0, 0.0],
[0.0, 0.0, 0.0], [0.0, 0.0, 0.0]])
mismatch = np.testing.assert_array_almost_equal(actual,
desired,
decimal=5)
@@ -444,8 +447,8 @@ class TestInitializers(unittest.TestCase):
def test_stratified_mean_unequal_one_hot(self):
pdist = torch.tensor([1, 3])
y = torch.eye(2)[self.y]
desired1 = torch.tensor([[5., 5., 5.], [1., 1., 1.], [1., 1., 1.],
[1., 1., 1.]])
desired1 = torch.tensor([[5.0, 5.0, 5.0], [1.0, 1.0, 1.0],
[1.0, 1.0, 1.0], [1.0, 1.0, 1.0]])
actual1, actual2 = initializers.stratified_mean(self.x, y, pdist)
desired2 = torch.tensor([[1, 0], [0, 1], [0, 1], [0, 1]])
mismatch = np.testing.assert_array_almost_equal(actual1,
@@ -460,8 +463,8 @@ class TestInitializers(unittest.TestCase):
pdist = torch.tensor([1, 3])
y = torch.eye(2)[self.y]
actual1, actual2 = initializers.stratified_random(self.x, y, pdist)
desired1 = torch.tensor([[0., -1., -2.], [0., 0., 0.], [0., 0., 0.],
[0., 0., 0.]])
desired1 = torch.tensor([[0.0, -1.0, -2.0], [0.0, 0.0, 0.0],
[0.0, 0.0, 0.0], [0.0, 0.0, 0.0]])
desired2 = torch.tensor([[1, 0], [0, 1], [0, 1], [0, 1]])
mismatch = np.testing.assert_array_almost_equal(actual1,
desired1,

View File

@@ -4,7 +4,6 @@ import unittest
import numpy as np
import torch
from prototorch.modules import losses, prototypes
@@ -18,21 +17,23 @@ class TestPrototypes(unittest.TestCase):
def test_prototypes1d_init_without_input_dim(self):
with self.assertRaises(NameError):
_ = prototypes.Prototypes1D(nclasses=2)
_ = prototypes.Prototypes1D(num_classes=2)
def test_prototypes1d_init_without_nclasses(self):
def test_prototypes1d_init_without_num_classes(self):
with self.assertRaises(NameError):
_ = prototypes.Prototypes1D(input_dim=1)
def test_prototypes1d_init_with_nclasses_1(self):
def test_prototypes1d_init_with_num_classes_1(self):
with self.assertWarns(UserWarning):
_ = prototypes.Prototypes1D(nclasses=1, input_dim=1)
_ = prototypes.Prototypes1D(num_classes=1, input_dim=1)
def test_prototypes1d_init_without_pdist(self):
p1 = prototypes.Prototypes1D(input_dim=6,
nclasses=2,
prototypes_per_class=4,
prototype_initializer='ones')
p1 = prototypes.Prototypes1D(
input_dim=6,
num_classes=2,
prototypes_per_class=4,
prototype_initializer="ones",
)
protos = p1.prototypes
actual = protos.detach().numpy()
desired = torch.ones(8, 6)
@@ -45,7 +46,7 @@ class TestPrototypes(unittest.TestCase):
pdist = [2, 2]
p1 = prototypes.Prototypes1D(input_dim=3,
prototype_distribution=pdist,
prototype_initializer='zeros')
prototype_initializer="zeros")
protos = p1.prototypes
actual = protos.detach().numpy()
desired = torch.zeros(4, 3)
@@ -58,16 +59,17 @@ class TestPrototypes(unittest.TestCase):
with self.assertWarns(UserWarning):
_ = prototypes.Prototypes1D(
input_dim=3,
nclasses=2,
num_classes=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=None)
prototype_initializer="stratified_mean",
data=None,
)
def test_prototypes1d_init_torch_pdist(self):
pdist = torch.tensor([2, 2])
p1 = prototypes.Prototypes1D(input_dim=3,
prototype_distribution=pdist,
prototype_initializer='zeros')
prototype_initializer="zeros")
protos = p1.prototypes
actual = protos.detach().numpy()
desired = torch.zeros(4, 3)
@@ -77,24 +79,30 @@ class TestPrototypes(unittest.TestCase):
self.assertIsNone(mismatch)
def test_prototypes1d_init_without_inputdim_with_data(self):
_ = prototypes.Prototypes1D(nclasses=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=[[[1.], [0.]], [1, 0]])
_ = prototypes.Prototypes1D(
num_classes=2,
prototypes_per_class=1,
prototype_initializer="stratified_mean",
data=[[[1.0], [0.0]], [1, 0]],
)
def test_prototypes1d_init_with_int_data(self):
_ = prototypes.Prototypes1D(nclasses=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=[[[1], [0]], [1, 0]])
_ = prototypes.Prototypes1D(
num_classes=2,
prototypes_per_class=1,
prototype_initializer="stratified_mean",
data=[[[1], [0]], [1, 0]],
)
def test_prototypes1d_init_one_hot_without_data(self):
_ = prototypes.Prototypes1D(input_dim=1,
nclasses=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=None,
one_hot_labels=True)
_ = prototypes.Prototypes1D(
input_dim=1,
num_classes=2,
prototypes_per_class=1,
prototype_initializer="stratified_mean",
data=None,
one_hot_labels=True,
)
def test_prototypes1d_init_one_hot_labels_false(self):
"""Test if ValueError is raised when `one_hot_labels` is set to `False`
@@ -103,11 +111,12 @@ class TestPrototypes(unittest.TestCase):
with self.assertRaises(ValueError):
_ = prototypes.Prototypes1D(
input_dim=1,
nclasses=2,
num_classes=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=([[0.], [1.]], [[0, 1], [1, 0]]),
one_hot_labels=False)
prototype_initializer="stratified_mean",
data=([[0.0], [1.0]], [[0, 1], [1, 0]]),
one_hot_labels=False,
)
def test_prototypes1d_init_1d_y_data_one_hot_labels_true(self):
"""Test if ValueError is raised when `one_hot_labels` is set to `True`
@@ -116,11 +125,12 @@ class TestPrototypes(unittest.TestCase):
with self.assertRaises(ValueError):
_ = prototypes.Prototypes1D(
input_dim=1,
nclasses=2,
num_classes=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=([[0.], [1.]], [0, 1]),
one_hot_labels=True)
prototype_initializer="stratified_mean",
data=([[0.0], [1.0]], [0, 1]),
one_hot_labels=True,
)
def test_prototypes1d_init_one_hot_labels_true(self):
"""Test if ValueError is raised when `one_hot_labels` is set to `True`
@@ -130,53 +140,57 @@ class TestPrototypes(unittest.TestCase):
with self.assertRaises(ValueError):
_ = prototypes.Prototypes1D(
input_dim=1,
nclasses=2,
num_classes=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=([[0.], [1.]], [[0], [1]]),
one_hot_labels=True)
prototype_initializer="stratified_mean",
data=([[0.0], [1.0]], [[0], [1]]),
one_hot_labels=True,
)
def test_prototypes1d_init_with_int_dtype(self):
with self.assertRaises(RuntimeError):
_ = prototypes.Prototypes1D(
nclasses=2,
num_classes=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
prototype_initializer="stratified_mean",
data=[[[1], [0]], [1, 0]],
dtype=torch.int32)
dtype=torch.int32,
)
def test_prototypes1d_inputndim_with_data(self):
with self.assertRaises(ValueError):
_ = prototypes.Prototypes1D(input_dim=1,
nclasses=1,
num_classes=1,
prototypes_per_class=1,
data=[[1.], [1]])
data=[[1.0], [1]])
def test_prototypes1d_inputdim_with_data(self):
with self.assertRaises(ValueError):
_ = prototypes.Prototypes1D(
input_dim=2,
nclasses=2,
num_classes=2,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=[[[1.], [0.]], [1, 0]])
prototype_initializer="stratified_mean",
data=[[[1.0], [0.0]], [1, 0]],
)
def test_prototypes1d_nclasses_with_data(self):
"""Test ValueError raise if provided `nclasses` is not the same
def test_prototypes1d_num_classes_with_data(self):
"""Test ValueError raise if provided `num_classes` is not the same
as the one computed from the provided `data`.
"""
with self.assertRaises(ValueError):
_ = prototypes.Prototypes1D(
input_dim=1,
nclasses=1,
num_classes=1,
prototypes_per_class=1,
prototype_initializer='stratified_mean',
data=[[[1.], [2.]], [1, 2]])
prototype_initializer="stratified_mean",
data=[[[1.0], [2.0]], [1, 2]],
)
def test_prototypes1d_init_with_ppc(self):
p1 = prototypes.Prototypes1D(data=[self.x, self.y],
prototypes_per_class=2,
prototype_initializer='zeros')
prototype_initializer="zeros")
protos = p1.prototypes
actual = protos.detach().numpy()
desired = torch.zeros(4, 3)
@@ -186,9 +200,11 @@ class TestPrototypes(unittest.TestCase):
self.assertIsNone(mismatch)
def test_prototypes1d_init_with_pdist(self):
p1 = prototypes.Prototypes1D(data=[self.x, self.y],
prototype_distribution=[6, 9],
prototype_initializer='zeros')
p1 = prototypes.Prototypes1D(
data=[self.x, self.y],
prototype_distribution=[6, 9],
prototype_initializer="zeros",
)
protos = p1.prototypes
actual = protos.detach().numpy()
desired = torch.zeros(15, 3)
@@ -201,10 +217,12 @@ class TestPrototypes(unittest.TestCase):
def my_initializer(*args, **kwargs):
return torch.full((2, 99), 99.0), torch.tensor([0, 1])
p1 = prototypes.Prototypes1D(input_dim=99,
nclasses=2,
prototypes_per_class=1,
prototype_initializer=my_initializer)
p1 = prototypes.Prototypes1D(
input_dim=99,
num_classes=2,
prototypes_per_class=1,
prototype_initializer=my_initializer,
)
protos = p1.prototypes
actual = protos.detach().numpy()
desired = 99 * torch.ones(2, 99)
@@ -231,7 +249,7 @@ class TestPrototypes(unittest.TestCase):
def test_prototypes1d_validate_extra_repr_not_empty(self):
p1 = prototypes.Prototypes1D(input_dim=0, prototype_distribution=[0])
rep = p1.extra_repr()
self.assertNotEqual(rep, '')
self.assertNotEqual(rep, "")
def tearDown(self):
del self.x, self.y, self.gen
@@ -243,11 +261,11 @@ class TestLosses(unittest.TestCase):
pass
def test_glvqloss_init(self):
_ = losses.GLVQLoss(0, 'swish_beta', beta=20)
_ = losses.GLVQLoss(0, "swish_beta", beta=20)
def test_glvqloss_forward_1ppc(self):
criterion = losses.GLVQLoss(margin=0,
squashing='sigmoid_beta',
squashing="sigmoid_beta",
beta=100)
d = torch.stack([torch.ones(100), torch.zeros(100)], dim=1)
labels = torch.tensor([0, 1])
@@ -259,7 +277,7 @@ class TestLosses(unittest.TestCase):
def test_glvqloss_forward_2ppc(self):
criterion = losses.GLVQLoss(margin=0,
squashing='sigmoid_beta',
squashing="sigmoid_beta",
beta=100)
d = torch.stack([
torch.ones(100),

15
tox.ini
View File

@@ -1,15 +0,0 @@
# tox (https://tox.readthedocs.io/) is a tool for running tests
# in multiple virtualenvs. This configuration file will run the
# test suite on all supported python versions. To use it, "pip install tox"
# and then run "tox" from this directory.
[tox]
envlist = py36,py37,py38
[testenv]
deps =
pytest
coverage
commands =
pip install -e .
coverage run -m pytest