AbductiveLearning
/
ABLkit
Not watched
1
0
Fork 0
Code
Releases 0 Wiki evaluate Activity
Issues 0 Pull Requests 0 Datasets Model Cloudbrain HPC
Browse Source

Merge branch 'Dev' of https://github.com/AbductiveLearning/ABL-Package into Dev

pull/3/head
troyyyyy 3 years ago
parent
f090eb26db c569b45801
commit
c285dbff6c
34 changed files with 1291 additions and 1076 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +1
    -1
      .github/workflows/build-and-test.yaml
  2. +5
    -1
      .github/workflows/lint.yaml
  3. +6
    -3
      .gitignore
  4. +1
    -1
      abl/abducer/readme.md
  5. +31
    -44
      abl/framework.py
  6. +326
    -39
      abl/models/basic_model.py
  7. +0
    -83
      abl/models/lenet5.py
  8. +136
    -0
      abl/models/readme.md
  9. +75
    -108
      abl/models/wabl_models.py
  10. +1
    -49
      abl/utils/utils.py
  11. +0
    -0
      examples/hed/datasets/BK.pl
  12. +0
    -0
      examples/hed/datasets/README.md
  13. +3
    -3
      examples/hed/datasets/get_hed.py
  14. +0
    -0
      examples/hed/datasets/learn_add.pl
  15. +6
    -82
      examples/hed/framework_hed.py
  16. +199
    -0
      examples/hed/hed_example.ipynb
  17. +69
    -0
      examples/hed/hed_knn_example.py
  18. +47
    -0
      examples/hed/utils.py
  19. +0
    -0
      examples/hed/weights/all_weights_here.txt
  20. +0
    -0
      examples/hwf/datasets/README.md
  21. +3
    -3
      examples/hwf/datasets/get_hwf.py
  22. +184
    -0
      examples/hwf/hwf_example.ipynb
  23. +0
    -0
      examples/hwf/weights/all_weights_here.txt
  24. +0
    -0
      examples/mnist_add/datasets/add.pl
  25. +3
    -5
      examples/mnist_add/datasets/get_mnist_add.py
  26. +0
    -0
      examples/mnist_add/datasets/test_data.txt
  27. +0
    -0
      examples/mnist_add/datasets/train_data.txt
  28. +190
    -0
      examples/mnist_add/mnist_add_example.ipynb
  29. +0
    -0
      examples/mnist_add/weights/all_weights_here.txt
  30. +1
    -50
      examples/models/nn.py
  31. +0
    -97
      examples/nonshare_example.py
  32. +0
    -96
      examples/share_example.py
  33. +0
    -407
      framework_hed_knn.py
  34. +4
    -4
      tests/test_models.py

+ 1
- 1
.github/workflows/build-and-test.yaml View File

@@ -2,7 +2,7 @@ name: ABL-Package-CI

on:
push:
branches: [ main, Dev ]
branches: [ main ]
pull_request:
branches: [ main ]



.github/workflows/lint.yml → .github/workflows/lint.yaml View File

@@ -1,6 +1,10 @@
name: flake8 Lint
on: [push, pull_request]
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
flake8-lint:

+ 6
- 3
.gitignore View File

@@ -1,6 +1,9 @@
*.pyc
/results
raw/
*.jpg
*.png
*.pk
*.pk
*.pth
*.json
*.ckpt
results
raw/

+ 1
- 1
abl/abducer/readme.md View File

@@ -34,7 +34,7 @@

## GKB

建立 KB 时, 用户可以在`__init__`中指定`GKB_flag`, 说明是否需要建立GKB (Ground Knowledge Base, 领域知识库). GKB 是一个 Python 字典, key 为`pseudo_label`组成的 list 代入`logic_forward`得到的所有可能的结果, 每个 key 对应的 value 为前述的`pseudo_label`组成的 list. 建立好 GKB 之后可以加快反绎所需的时间.
建立 KB 时, 用户可以在`__init__`中指定`GKB_flag`, 说明是否需要建立GKB (Ground Knowledge Base, 领域知识库). GKB 是一个 Python 字典, key 为`pseudo_label`组成的 list 代入`logic_forward`得到的所有可能的结果, 每个 key 对应的 value 为前述的`pseudo_label`组成的 list. 建立好 GKB 之后可以加快反绎的速度.

### GKB 的建立



+ 31
- 44
abl/framework.py View File

@@ -1,32 +1,18 @@
# coding: utf-8
#================================================================#
# ================================================================#
# Copyright (C) 2021 Freecss All rights reserved.
#
#
# File Name ：framework.py
# Author ：freecss
# Email ：karlfreecss@gmail.com
# Created Date ：2021/06/07
# Description ：
#
#================================================================#

import pickle as pk

import numpy as np

from .utils.plog import INFO, DEBUG, clocker
# ================================================================#

def block_sample(X, Z, Y, sample_num, epoch_idx):
part_num = (len(X) // sample_num)
if part_num == 0:
part_num = 1
seg_idx = epoch_idx % part_num
INFO("seg_idx:", seg_idx, ", part num:", part_num, ", data num:", len(X))
X = X[sample_num * seg_idx: sample_num * (seg_idx + 1)]
Z = Z[sample_num * seg_idx: sample_num * (seg_idx + 1)]
Y = Y[sample_num * seg_idx: sample_num * (seg_idx + 1)]
from .utils.plog import INFO, clocker
from .utils.utils import block_sample

return X, Z, Y

def result_statistics(pred_Z, Z, Y, logic_forward, char_acc_flag):
result = {}
@@ -36,72 +22,73 @@ def result_statistics(pred_Z, Z, Y, logic_forward, char_acc_flag):
for pred_z, z in zip(pred_Z, Z):
char_num += len(z)
for zidx in range(len(z)):
if(pred_z[zidx] == z[zidx]):
if pred_z[zidx] == z[zidx]:
char_acc_num += 1
char_acc = char_acc_num / char_num
result["Character level accuracy"] = char_acc
abl_acc_num = 0
for pred_z, y in zip(pred_Z, Y):
if(logic_forward(pred_z) == y):
abl_acc_num += 1
if logic_forward(pred_z) == y:
abl_acc_num += 1
abl_acc = abl_acc_num / len(Y)
result["ABL accuracy"] = abl_acc

return result


def filter_data(X, abduced_Z):
finetune_Z = []
finetune_X = []
for abduced_x, abduced_z in zip(X, abduced_Z):
if abduced_z is not []:
finetune_X.append(abduced_x)
for x, abduced_z in zip(X, abduced_Z):
if len(abduced_z) > 0:
finetune_X.append(x)
finetune_Z.append(abduced_z)
return finetune_X, finetune_Z

def pretrain(model, X, Z):
pass

def train(model, abducer, train_data, test_data, epochs = 50, sample_num = -1, verbose = -1):
def train(
model, abducer, train_data, test_data, loop_num=50, sample_num=-1, verbose=-1
):
train_X, train_Z, train_Y = train_data
test_X, test_Z, test_Y = test_data
# Set default parameters
if sample_num == -1:
sample_num = len(train_X)

if verbose < 1:
verbose = epochs
verbose = loop_num
char_acc_flag = 1
if train_Z == None:
char_acc_flag = 0
train_Z = [None] * len(X)
train_Z = [None] * len(train_X)

predict_func = clocker(model.predict)
train_func = clocker(model.train)
abduce_func = clocker(abducer.batch_abduce)
# Abductive learning train process
for epoch_idx in range(epochs):
X, Z, Y = block_sample(train_X, train_Z, train_Y, sample_num, epoch_idx)

for loop_idx in range(loop_num):
X, Z, Y = block_sample(train_X, train_Z, train_Y, sample_num, loop_idx)
preds_res = predict_func(X)
abduced_Z = abduce_func(preds_res, Y)

if ((epoch_idx + 1) % verbose == 0) or (epoch_idx == epochs - 1):
res = result_statistics(preds_res['cls'], Z, Y, abducer.kb.logic_forward, char_acc_flag)
INFO('epoch: ', epoch_idx + 1, ' ', res)
if ((loop_idx + 1) % verbose == 0) or (loop_idx == loop_num - 1):
res = result_statistics(
preds_res["cls"], Z, Y, abducer.kb.logic_forward, char_acc_flag
)
INFO("loop: ", loop_idx + 1, " ", res)

finetune_X, finetune_Z = filter_data(X, abduced_Z)
if len(finetune_X) > 0:
# model.valid(finetune_X, finetune_Z)
train_func(finetune_X, finetune_Z)
else:
INFO("lack of data, all abduced failed", len(finetune_X))
return res


if __name__ == "__main__":
pass



+ 326
- 39
abl/models/basic_model.py View File

@@ -15,22 +15,53 @@ import sys
sys.path.append("..")

import torch
from torch.utils.data import Dataset
import numpy
from torch.utils.data import Dataset, DataLoader

import os
from multiprocessing import Pool
from typing import List, Any, T, Tuple, Optional, Callable


class BasicDataset(Dataset):
def __init__(self, X, Y):
def __init__(self, X: List[Any], Y: List[Any]):
"""Initialize a basic dataset.

Parameters
----------
X : List[Any]
A list of objects representing the input data.
Y : List[Any]
A list of objects representing the output data.
"""
self.X = X
self.Y = Y

def __len__(self):
"""Return the length of the dataset.

Returns
-------
int
The length of the dataset.
"""
return len(self.X)

def __getitem__(self, index):
assert index < len(self), "index range error"
def __getitem__(self, index: int) -> Tuple[Any, Any]:
"""Get an item from the dataset.

Parameters
----------
index : int
The index of the item to retrieve.

Returns
-------
Tuple[Any, Any]
A tuple containing the input and output data at the specified index.
"""
if index >= len(self):
raise ValueError("index range error")

img = self.X[index]
label = self.Y[index]
@@ -39,18 +70,52 @@ class BasicDataset(Dataset):


class XYDataset(Dataset):
def __init__(self, X, Y, transform=None):
def __init__(self, X: List[Any], Y: List[int], transform: Callable[..., Any] = None):
"""
Initialize the dataset used for classification task.

Parameters
----------
X : List[Any]
The input data.
Y : List[int]
The target data.
transform : Callable[..., Any], optional
A function/transform that takes in an object and returns a transformed version. Defaults to None.
"""
self.X = X
self.Y = torch.LongTensor(Y)

self.n_sample = len(X)
self.transform = transform

def __len__(self):
def __len__(self) -> int:
"""
Return the length of the dataset.

Returns
-------
int
The length of the dataset.
"""
return len(self.X)

def __getitem__(self, index):
assert index < len(self), "index range error"
def __getitem__(self, index: int) -> Tuple[Any, torch.Tensor]:
"""
Get the item at the given index.

Parameters
----------
index : int
The index of the item to get.

Returns
-------
Tuple[Any, torch.Tensor]
A tuple containing the object and its label.
"""
if index >= len(self):
raise ValueError("index range error")

img = self.X[index]
if self.transform is not None:
@@ -70,20 +135,103 @@ class FakeRecorder:


class BasicModel:
"""
Wrap NN models into the form of an sklearn estimator

Parameters
----------
model : torch.nn.Module
The PyTorch model to be trained or used for prediction.
criterion : torch.nn.Module
The loss function used for training.
optimizer : torch.nn.Module
The optimizer used for training.
device : torch.device, optional
The device on which the model will be trained or used for prediction, by default torch.decive("cpu").
batch_size : int, optional
The batch size used for training, by default 1.
num_epochs : int, optional
The number of epochs used for training, by default 1.
stop_loss : Optional[float], optional
The loss value at which to stop training, by default 0.01.
num_workers : int, optional
The number of workers used for loading data, by default 0.
save_interval : Optional[int], optional
The interval at which to save the model during training, by default None.
save_dir : Optional[str], optional
The directory in which to save the model during training, by default None.
transform : Callable[..., Any], optional
A function/transform that takes in an object and returns a transformed version. Defaults to None.
collate_fn : Callable[[List[T]], Any], optional
The function used to collate data, by default None.
recorder : Any, optional
The recorder used to record training progress, by default None.

Attributes
----------
model : torch.nn.Module
The PyTorch model to be trained or used for prediction.
batch_size : int
The batch size used for training.
num_epochs : int
The number of epochs used for training.
stop_loss : Optional[float]
The loss value at which to stop training.
num_workers : int
The number of workers used for loading data.
criterion : torch.nn.Module
The loss function used for training.
optimizer : torch.nn.Module
The optimizer used for training.
transform : Callable[..., Any]
The transformation function used for data augmentation.
device : torch.device
The device on which the model will be trained or used for prediction.
recorder : Any
The recorder used to record training progress.
save_interval : Optional[int]
The interval at which to save the model during training.
save_dir : Optional[str]
The directory in which to save the model during training.
collate_fn : Callable[[List[T]], Any]
The function used to collate data.

Methods
-------
fit(data_loader=None, X=None, y=None)
Train the model.
train_epoch(data_loader)
Train the model for one epoch.
predict(data_loader=None, X=None, print_prefix="")
Predict the class of the input data.
predict_proba(data_loader=None, X=None, print_prefix="")
Predict the probability of each class for the input data.
val(data_loader=None, X=None, y=None, print_prefix="")
Validate the model.
score(data_loader=None, X=None, y=None, print_prefix="")
Score the model.
_data_loader(X, y=None)
Generate the data_loader.
save(epoch_id, save_dir="")
Save the model.
load(epoch_id, load_dir="")
Load the model.
"""

def __init__(
self,
model,
criterion,
optimizer,
device,
batch_size=1,
num_epochs=1,
stop_loss=0.01,
num_workers=0,
save_interval=None,
save_dir=None,
transform=None,
collate_fn=None,
model: torch.nn.Module,
criterion: torch.nn.Module,
optimizer: torch.nn.Module,
device: torch.device = torch.device("cpu"),
batch_size: int = 1,
num_epochs: int = 1,
stop_loss: Optional[float] = 0.01,
num_workers: int = 0,
save_interval: Optional[int] = None,
save_dir: Optional[str] = None,
transform: Callable[..., Any] = None,
collate_fn: Callable[[List[T]], Any] = None,
recorder=None,
):

@@ -106,7 +254,6 @@ class BasicModel:
self.save_interval = save_interval
self.save_dir = save_dir
self.collate_fn = collate_fn
pass

def _fit(self, data_loader, n_epoch, stop_loss):
recorder = self.recorder
@@ -119,19 +266,54 @@ class BasicModel:
if min_loss < 0 or loss_value < min_loss:
min_loss = loss_value
if self.save_interval is not None and (epoch + 1) % self.save_interval == 0:
assert self.save_dir is not None
if self.save_dir is None:
raise ValueError(
"save_dir should not be None if save_interval is not None"
)
self.save(epoch + 1, self.save_dir)
if stop_loss is not None and loss_value < stop_loss:
break
recorder.print("Model fitted, minimal loss is ", min_loss)
return loss_value

def fit(self, data_loader=None, X=None, y=None):
def fit(
self, data_loader: DataLoader = None, X: List[Any] = None, y: List[int] = None
) -> float:
"""
Train the model.

Parameters
----------
data_loader : DataLoader, optional
The data loader used for training, by default None
X : List[Any], optional
The input data, by default None
y : List[int], optional
The target data, by default None

Returns
-------
float
The loss value of the trained model.
"""
if data_loader is None:
data_loader = self._data_loader(X, y)
return self._fit(data_loader, self.num_epochs, self.stop_loss)

def train_epoch(self, data_loader):
def train_epoch(self, data_loader: DataLoader):
"""
Train the model for one epoch.

Parameters
----------
data_loader : DataLoader
The data loader used for training.

Returns
-------
float
The loss value of the trained model.
"""
model = self.model
criterion = self.criterion
optimizer = self.optimizer
@@ -169,7 +351,29 @@ class BasicModel:

return torch.cat(results, axis=0)

def predict(self, data_loader=None, X=None, print_prefix=""):
def predict(
self,
data_loader: DataLoader = None,
X: List[Any] = None,
print_prefix: str = "",
) -> numpy.ndarray:
"""
Predict the class of the input data.

Parameters
----------
data_loader : DataLoader, optional
The data loader used for prediction, by default None
X : List[Any], optional
The input data, by default None
print_prefix : str, optional
The prefix used for printing, by default ""

Returns
-------
numpy.ndarray
The predicted class of the input data.
"""
recorder = self.recorder
recorder.print("Start Predict Class ", print_prefix)

@@ -177,15 +381,37 @@ class BasicModel:
data_loader = self._data_loader(X)
return self._predict(data_loader).argmax(axis=1).cpu().numpy()

def predict_proba(self, data_loader=None, X=None, print_prefix=""):
def predict_proba(
self,
data_loader: DataLoader = None,
X: List[Any] = None,
print_prefix: str = "",
) -> numpy.ndarray:
"""
Predict the probability of each class for the input data.

Parameters
----------
data_loader : DataLoader, optional
The data loader used for prediction, by default None
X : List[Any], optional
The input data, by default None
print_prefix : str, optional
The prefix used for printing, by default ""

Returns
-------
numpy.ndarray
The predicted probability of each class for the input data.
"""
recorder = self.recorder
# recorder.print('Start Predict Probability ', print_prefix)
recorder.print("Start Predict Probability ", print_prefix)

if data_loader is None:
data_loader = self._data_loader(X)
return self._predict(data_loader).softmax(axis=1).cpu().numpy()

def _val(self, data_loader):
def _score(self, data_loader):
model = self.model
criterion = self.criterion
device = self.device
@@ -215,22 +441,63 @@ class BasicModel:

return mean_loss, accuracy

def val(self, data_loader=None, X=None, y=None, print_prefix=""):
def score(
self,
data_loader: DataLoader = None,
X: List[Any] = None,
y: List[int] = None,
print_prefix: str = "",
) -> float:
"""
Validate the model.

Parameters
----------
data_loader : DataLoader, optional
The data loader used for scoring, by default None
X : List[Any], optional
The input data, by default None
y : List[int], optional
The target data, by default None
print_prefix : str, optional
The prefix used for printing, by default ""

Returns
-------
float
The accuracy of the model.
"""
recorder = self.recorder
recorder.print("Start val ", print_prefix)
recorder.print("Start validation ", print_prefix)

if data_loader is None:
data_loader = self._data_loader(X, y)
mean_loss, accuracy = self._val(data_loader)
mean_loss, accuracy = self._score(data_loader)
recorder.print(
"[%s] Val loss: %f, accuray: %f" % (print_prefix, mean_loss, accuracy)
"[%s] mean loss: %f, accuray: %f" % (print_prefix, mean_loss, accuracy)
)
return accuracy

def score(self, data_loader=None, X=None, y=None, print_prefix=""):
return self.val(data_loader, X, y, print_prefix)

def _data_loader(self, X, y=None):
def _data_loader(
self,
X: List[Any],
y: List[int] = None,
) -> DataLoader:
"""
Generate data_loader for user provided data.

Parameters
----------
X : List[Any]
The input data.
y : List[int], optional
The target data, by default None

Returns
-------
DataLoader
The data loader.
"""
collate_fn = self.collate_fn
transform = self.transform

@@ -238,7 +505,7 @@ class BasicModel:
y = [0] * len(X)
dataset = XYDataset(X, y, transform=transform)
sampler = None
data_loader = torch.utils.data.DataLoader(
data_loader = DataLoader(
dataset,
batch_size=self.batch_size,
shuffle=False,
@@ -248,7 +515,17 @@ class BasicModel:
)
return data_loader

def save(self, epoch_id, save_dir):
def save(self, epoch_id: int, save_dir: str = ""):
"""
Save the model and the optimizer.

Parameters
----------
epoch_id : int
The epoch id.
save_dir : str, optional
The directory to save the model, by default ""
"""
recorder = self.recorder
if not os.path.exists(save_dir):
os.makedirs(save_dir)
@@ -259,7 +536,17 @@ class BasicModel:
save_path = os.path.join(save_dir, str(epoch_id) + "_opt.pth")
torch.save(self.optimizer.state_dict(), save_path)

def load(self, epoch_id, load_dir):
def load(self, epoch_id: int, load_dir: str = ""):
"""
Load the model and the optimizer.

Parameters
----------
epoch_id : int
The epoch id.
load_dir : str, optional
The directory to load the model, by default ""
"""
recorder = self.recorder
recorder.print("Loading model and opter")
load_path = os.path.join(load_dir, str(epoch_id) + "_net.pth")


+ 0
- 83
abl/models/lenet5.py View File

@@ -1,83 +0,0 @@
# coding: utf-8
#================================================================#
# Copyright (C) 2021 Freecss All rights reserved.
#
# File Name ：lenet5.py
# Author ：freecss
# Email ：karlfreecss@gmail.com
# Created Date ：2021/03/03
# Description ：
#
#================================================================#

import sys
sys.path.append("..")

import torchvision

import torch
from torch import nn
from torch.nn import functional as F
from torch.autograd import Variable
import torchvision.transforms as transforms
import numpy as np

from models.basic_model import BasicModel
import utils.plog as plog

class LeNet5(nn.Module):
def __init__(self, num_classes=10, image_size=(28, 28)):
super().__init__()
self.conv1 = nn.Conv2d(1, 6, 3, padding=1)
self.conv2 = nn.Conv2d(6, 16, 3)
self.conv3 = nn.Conv2d(16, 16, 3)

feature_map_size = ((np.array(image_size) // 2 - 2) // 2 - 2)
num_features = 16 * feature_map_size[0] * feature_map_size[1]

self.fc1 = nn.Linear(num_features, 120)
self.fc2 = nn.Linear(120, 84)
self.fc3 = nn.Linear(84, num_classes)

def forward(self, x):
'''前向传播函数'''
x = F.max_pool2d(F.relu(self.conv1(x)), (2, 2))
x = F.max_pool2d(F.relu(self.conv2(x)), (2, 2))
x = F.relu(self.conv3(x))
x = x.view(-1, self.num_flat_features(x))
#print(x.size())
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
x = self.fc3(x)
return x

def num_flat_features(self, x):
size = x.size()[1:]
num_features = 1
for s in size:
num_features *= s
return num_features


class SymbolNet(nn.Module):
def __init__(self, num_classes=14):
super(SymbolNet, self).__init__()
self.conv1 = nn.Conv2d(1, 32, 3, stride = 1, padding = 1)
self.conv2 = nn.Conv2d(32, 64, 3, stride = 1, padding = 1)
self.dropout1 = nn.Dropout2d(0.25)
self.dropout2 = nn.Dropout2d(0.5)
self.fc1 = nn.Linear(30976, 128)
self.fc2 = nn.Linear(128, num_classes)

def forward(self, x):
x = self.conv1(x)
x = F.relu(x)
x = self.conv2(x)
x = F.max_pool2d(x, 2)
x = self.dropout1(x)
x = torch.flatten(x, 1)
x = self.fc1(x)
x = F.relu(x)
x = self.dropout2(x)
x = self.fc2(x)
return x

+ 136
- 0
abl/models/readme.md View File

@@ -0,0 +1,136 @@
# `basic_model.py`

可以使用`basic_model.py`中实现的`BasicModel`类将`pytorch`神经网络模型包装成`sklearn`模型的形式.

## BasicModel 类提供的接口

| 方法 | 功能 |
| ---- | ---- |
| fit(X, y) | 训练神经网络 |
| predict(X) | 预测 X 的类别 |
| predict_proba(X) | 预测 X 的类别概率 |
| score(X, y) | 计算模型在测试数据上的准确率 |
| save() | 保存模型 |
| load() | 加载模型 |


## BasicModel 类的参数

**model : torch.nn.Module**
+ The PyTorch model to be trained or used for prediction.

**batch_size : int**
+ The batch size used for training.

**num_epochs : int**
+ The number of epochs used for training.

**stop_loss : Optional[float]**
+ The loss value at which to stop training.

**num_workers : int**
+ The number of workers used for loading data.

**criterion : torch.nn.Module**
+ The loss function used for training.

**optimizer : torch.nn.Module**
+ The optimizer used for training.

**transform : Callable[..., Any]**
+ The transformation function used for data augmentation.

**device : torch.device**
+ The device on which the model will be trained or used for prediction.

**recorder : Any**
+ The recorder used to record training progress.

**save_interval : Optional[int]**
+ The interval at which to save the model during training.

**save_dir : Optional[str]**
+ The directory in which to save the model during training.

**collate_fn : Callable[[List[T]], Any]**
+ The function used to collate data.

## 例子
>
> ```python
> # Three necessary component
> cls = LeNet5()
> criterion = nn.CrossEntropyLoss()
> optimizer = torch.optim.Adam(cls.parameters())
>
> # Initialize base_model
> base_model = BasicModel(
> cls,
> criterion,
> optimizer,
> torch.device("cuda:0"),
> batch_size=32,
> num_epochs=10,
> )
>
> # Prepare data
> train_X, train_y = get_train_data()
> test_X, test_y = get_test_data()
>
> # Train model
> base_model.fit(train_X, train_y)
>
> # Predict
> base_model.predict(test_X)
>
> # Validation
> base_model.score(test_X, test_y)
> ```

# `wabl_models.py`

`wabl_models.py`中实现的`WABLBasicModel`能够序列化数据并为不同的机器学习模型提供统一的接口.

## WABLBasicModel 类提供的接口

| 方法 | 功能 |
| ---- | ---- |
| train(X, Y) | 利用训练数据训练机器学习模型（不涉及反绎） |
| predict(X) | 预测 X 的类别和概率 |
| valid(X, Y) | 计算模型在测试数据上的准确率 |

## WABLBasicModel 类的参数
**base_model : Machine Learning Model**
+ The base model to use for training and prediction.

**pseudo_label_list : List[Any]**
+ A list of pseudo labels to use for training.

## 序列化数据
考虑到训练数据可能多种组织形式，比如：\
`X: List[List[img]], Y: List[List[label]]`\
`X: List[List[img]], Y: List[label]`\
`X: List[img], Y: List[label]`
... \
不便于训练. 因此先将形式统一为：`X: List[img], Y: List[label]`，也就是所谓的序列化数据.

## 例子
>
> ```python
> # Three necessary component
> # 'ml_model' is no longer limited to NN models
> model = WABLBasicModel(ml_model, kb.pseudo_label_list)
>
> # Prepare data
> train_X, train_y = get_train_data()
> test_X, test_y = get_test_data()
>
> # Train model
> model.train(train_X, train_y)
>
> # Predict
> model.predict(test_X)
>
> # Validation
> model.valid(test_X, test_y)
> ```

+ 75
- 108
abl/models/wabl_models.py View File

@@ -10,23 +10,7 @@
#
# ================================================================#
from itertools import chain

from sklearn.tree import DecisionTreeClassifier
from sklearn.model_selection import cross_val_score

from sklearn.svm import LinearSVC

from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.svm import SVC
from sklearn.gaussian_process import GaussianProcessClassifier
from sklearn.gaussian_process.kernels import RBF

import pickle as pk
import random

from sklearn.neighbors import KNeighborsClassifier
import numpy as np
from typing import List, Any


def get_part_data(X, i):
@@ -50,7 +34,37 @@ def reshape_data(Y, marks):


class WABLBasicModel:
def __init__(self, base_model, pseudo_label_list):
"""
Serialize data and provide a unified interface for different machine learning models.

Parameters
----------
base_model : Machine Learning Model
The base model to use for training and prediction.
pseudo_label_list : List[Any]
A list of pseudo labels to use for training.

Attributes
----------
cls_list : List[Any]
A list of classifiers.
pseudo_label_list : List[Any]
A list of pseudo labels to use for training.
mapping : dict
A dictionary mapping pseudo labels to integers.
remapping : dict
A dictionary mapping integers to pseudo labels.

Methods
-------
predict(X: List[List[Any]]) -> dict
Predict the class labels and probabilities for the given data.
valid(X: List[List[Any]], Y: List[Any]) -> float
Calculate the accuracy score for the given data.
train(X: List[List[Any]], Y: List[Any])
Train the model on the given data.
"""
def __init__(self, base_model, pseudo_label_list: List[Any]):
self.cls_list = []
self.cls_list.append(base_model)

@@ -60,7 +74,20 @@ class WABLBasicModel:
zip(list(range(len(pseudo_label_list))), pseudo_label_list)
)

def predict(self, X):
def predict(self, X: List[List[Any]]) -> dict:
"""
Predict the class labels and probabilities for the given data.

Parameters
----------
X : List[List[Any]]
The data to predict on.

Returns
-------
dict
A dictionary containing the predicted class labels and probabilities.
"""
data_X, marks = merge_data(X)
prob = self.cls_list[0].predict_proba(X=data_X)
_cls = prob.argmax(axis=1)
@@ -71,100 +98,40 @@ class WABLBasicModel:

return {"cls": cls, "prob": prob}

def valid(self, X, Y):
def valid(self, X: List[List[Any]], Y: List[Any]) -> float:
"""
Calculate the accuracy for the given data.

Parameters
----------
X : List[List[Any]]
The data to calculate the accuracy on.
Y : List[Any]
The true class labels for the given data.

Returns
-------
float
The accuracy score for the given data.
"""
data_X, _ = merge_data(X)
_data_Y, _ = merge_data(Y)
data_Y = list(map(lambda y: self.mapping[y], _data_Y))
score = self.cls_list[0].score(X=data_X, y=data_Y)
return score, [score]

def train(self, X, Y):
# self.label_lists = []
return score

def train(self, X: List[List[Any]], Y: List[Any]):
"""
Train the model on the given data.

Parameters
----------
X : List[List[Any]]
The data to train on.
Y : List[Any]
The true class labels for the given data.
"""
data_X, _ = merge_data(X)
_data_Y, _ = merge_data(Y)
data_Y = list(map(lambda y: self.mapping[y], _data_Y))
self.cls_list[0].fit(X=data_X, y=data_Y)


class DecisionTree(WABLBasicModel):
def __init__(self, code_len, label_lists, share=False):
self.code_len = code_len
self._set_label_lists(label_lists)

self.cls_list = []
self.share = share
if share:
# 本质上是同一个分类器
self.cls_list.append(
DecisionTreeClassifier(random_state=0, min_samples_leaf=3)
)
self.cls_list = self.cls_list * self.code_len
else:
for _ in range(code_len):
self.cls_list.append(
DecisionTreeClassifier(random_state=0, min_samples_leaf=3)
)


class KNN(WABLBasicModel):
def __init__(self, code_len, label_lists, share=False, k=3):
self.code_len = code_len
self._set_label_lists(label_lists)

self.cls_list = []
self.share = share
if share:
# 本质上是同一个分类器
self.cls_list.append(KNeighborsClassifier(n_neighbors=k))
self.cls_list = self.cls_list * self.code_len
else:
for _ in range(code_len):
self.cls_list.append(KNeighborsClassifier(n_neighbors=k))


class CNN(WABLBasicModel):
def __init__(self, base_model, code_len, label_lists, share=True):
assert share == True, "Not implemented"

label_lists = [sorted(list(set(label_list))) for label_list in label_lists]
self.label_lists = label_lists

self.code_len = code_len

self.cls_list = []
self.share = share
if share:
self.cls_list.append(base_model)

def train(self, X, Y, n_epoch=100):
# self.label_lists = []
if self.share:
# 因为是同一个分类器，所以只需要把数据放在一起，然后训练其中任意一个即可
data_X, _ = merge_data(X)
data_Y, _ = merge_data(Y)
self.cls_list[0].fit(X=data_X, y=data_Y, n_epoch=n_epoch)
# self.label_lists = [sorted(list(set(data_Y)))] * self.code_len
else:
for i in range(self.code_len):
data_X = get_part_data(X, i)
data_Y = get_part_data(Y, i)
self.cls_list[i].fit(data_X, data_Y)
# self.label_lists.append(sorted(list(set(data_Y))))


if __name__ == "__main__":
# data_path = "utils/hamming_data/generated_data/hamming_7_3_0.20.pk"
data_path = "datasets/generated_data/0_code_7_2_0.00.pk"
codes, data, labels = pk.load(open(data_path, "rb"))

cls = KNN(7, False, k=3)
cls.train(data, labels)
print(cls.valid(data, labels))
for res in cls.predict_proba(data):
print(res)
break

for res in cls.predict(data):
print(res)
break
print("Trained")

+ 1
- 49
abl/utils/utils.py View File

@@ -1,8 +1,5 @@
import torch
import torch.nn as nn
import numpy as np
from .plog import INFO
from collections import OrderedDict
from itertools import chain

def flatten(l):
@@ -51,31 +48,6 @@ def block_sample(X, Z, Y, sample_num, epoch_idx):

return X, Z, Y

def gen_mappings(chars, symbs):
n_char = len(chars)
n_symbs = len(symbs)
if n_char != n_symbs:
print("Characters and symbols size dosen't match.")
return
from itertools import permutations

mappings = []
# returned mappings
perms = permutations(symbs)
for p in perms:
mappings.append(dict(zip(chars, list(p))))
return mappings


def mapping_res(original_pred_res, m):
return [[m[symbol] for symbol in formula] for formula in original_pred_res]


def remapping_res(pred_res, m):
remapping = {}
for key, value in m.items():
remapping[value] = key
return [[remapping[symbol] for symbol in formula] for formula in pred_res]

def check_equal(a, b, max_err=0):
if isinstance(a, (int, float)) and isinstance(b, (int, float)):
@@ -90,27 +62,7 @@ def check_equal(a, b, max_err=0):
return True

else:
return a == b


def extract_feature(img):
extractor = nn.AvgPool2d(2, stride=2)
feature_map = np.array(extractor(torch.Tensor(img)))
return feature_map.reshape((-1,))
return np.concatenate(
(np.squeeze(np.sum(img, axis=1)), np.squeeze(np.sum(img, axis=2))), axis=0
)


def reduce_dimension(data):
for truth_value in [0, 1]:
for equation_len in range(5, 27):
equations = data[truth_value][equation_len]
reduced_equations = [
[extract_feature(symbol_img) for symbol_img in equation]
for equation in equations
]
data[truth_value][equation_len] = reduced_equations
return a == b

def to_hashable(l):


examples/datasets/hed/BK.pl → examples/hed/datasets/BK.pl View File


examples/datasets/hed/README.md → examples/hed/datasets/README.md View File


examples/datasets/hed/get_hed.py → examples/hed/datasets/get_hed.py View File

@@ -41,7 +41,7 @@ def get_pretrain_data(labels, image_size=(28, 28, 1)):
X = []
for label in labels:
label_path = os.path.join(
"./datasets/hed/mnist_images", label
"./datasets/mnist_images", label
)
img_path_list = os.listdir(label_path)
for img_path in img_path_list:
@@ -107,13 +107,13 @@ def get_hed(dataset="mnist", train=True):

if dataset == "mnist":
with open(
"./datasets/hed/mnist_equation_data_train_len_26_test_len_26_sys_2_.pk",
"./datasets/mnist_equation_data_train_len_26_test_len_26_sys_2_.pk",
"rb",
) as f:
img_dataset = pickle.load(f)
elif dataset == "random":
with open(
"./datasets/hed/random_equation_data_train_len_26_test_len_26_sys_2_.pk",
"./datasets/random_equation_data_train_len_26_test_len_26_sys_2_.pk",
"rb",
) as f:
img_dataset = pickle.load(f)

examples/datasets/hed/learn_add.pl → examples/hed/datasets/learn_add.pl View File


abl/framework_hed.py → examples/hed/framework_hed.py View File

@@ -10,94 +10,18 @@
#
# ================================================================#

import pickle as pk
import torch
import torch.nn as nn
import numpy as np
import os

from .utils.plog import INFO, DEBUG, clocker
from .utils.utils import flatten, reform_idx, block_sample, gen_mappings, mapping_res, remapping_res
from abl.utils.plog import INFO
from abl.utils.utils import flatten, reform_idx
from abl.models.basic_model import BasicModel, BasicDataset

from .models.nn import MLP, SymbolNetAutoencoder
from .models.basic_model import BasicModel, BasicDataset

import sys
sys.path.append("..")
from examples.datasets.hed.get_hed import get_pretrain_data

def result_statistics(pred_Z, Z, Y, logic_forward, char_acc_flag):
result = {}
if char_acc_flag:
char_acc_num = 0
char_num = 0
for pred_z, z in zip(pred_Z, Z):
char_num += len(z)
for zidx in range(len(z)):
if pred_z[zidx] == z[zidx]:
char_acc_num += 1
char_acc = char_acc_num / char_num
result["Character level accuracy"] = char_acc

abl_acc_num = 0
for pred_z, y in zip(pred_Z, Y):
if logic_forward(pred_z) == y:
abl_acc_num += 1
abl_acc = abl_acc_num / len(Y)
result["ABL accuracy"] = abl_acc

return result


def filter_data(X, abduced_Z):
finetune_Z = []
finetune_X = []
for x, abduced_z in zip(X, abduced_Z):
if len(abduced_z) > 0:
finetune_X.append(x)
finetune_Z.append(abduced_z)
return finetune_X, finetune_Z


def train(model, abducer, train_data, test_data, epochs=50, sample_num=-1, verbose=-1):
train_X, train_Z, train_Y = train_data
test_X, test_Z, test_Y = test_data

# Set default parameters
if sample_num == -1:
sample_num = len(train_X)

if verbose < 1:
verbose = epochs

char_acc_flag = 1
if train_Z == None:
char_acc_flag = 0
train_Z = [None] * len(train_X)

predict_func = clocker(model.predict)
train_func = clocker(model.train)
abduce_func = clocker(abducer.batch_abduce)

for epoch_idx in range(epochs):
X, Z, Y = block_sample(train_X, train_Z, train_Y, sample_num, epoch_idx)
preds_res = predict_func(X)
abduced_Z = abduce_func(preds_res, Y)

if ((epoch_idx + 1) % verbose == 0) or (epoch_idx == epochs - 1):
res = result_statistics(preds_res['cls'], Z, Y, abducer.kb.logic_forward, char_acc_flag)
INFO('epoch: ', epoch_idx + 1, ' ', res)

finetune_X, finetune_Z = filter_data(X, abduced_Z)
if len(finetune_X) > 0:
# model.valid(finetune_X, finetune_Z)
train_func(finetune_X, finetune_Z)
else:
INFO("lack of data, all abduced failed", len(finetune_X))

return res
from utils import gen_mappings, mapping_res, remapping_res
from models.nn import SymbolNetAutoencoder
from datasets.get_hed import get_pretrain_data


def hed_pretrain(kb, cls, recorder):

+ 199
- 0
examples/hed/hed_example.ipynb View File

@@ -0,0 +1,199 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 10,
"metadata": {},
"outputs": [],
"source": [
"import sys\n",
"\n",
"sys.path.append(\"../../\")\n",
"\n",
"import torch.nn as nn\n",
"import torch\n",
"\n",
"from abl.abducer.abducer_base import HED_Abducer\n",
"from abl.abducer.kb import HED_prolog_KB\n",
"\n",
"from abl.utils.plog import logger\n",
"from abl.models.basic_model import BasicModel\n",
"from abl.models.wabl_models import WABLBasicModel\n",
"\n",
"from models.nn import SymbolNet\n",
"from datasets.get_hed import get_hed, split_equation\n",
"import framework_hed"
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {},
"outputs": [],
"source": [
"# Initialize logger\n",
"recorder = logger()"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Logic Part"
]
},
{
"cell_type": "code",
"execution_count": 12,
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"ERROR: /home/gaoeh/ABL-Package/examples/hed/datasets/learn_add.pl:67:9: Syntax error: Operator expected\n"
]
}
],
"source": [
"# Initialize knowledge base and abducer\n",
"kb = HED_prolog_KB(pseudo_label_list=[1, 0, '+', '='], pl_file='./datasets/learn_add.pl')\n",
"abducer = HED_Abducer(kb)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Machine Learning Part"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize necessary component for machine learning part\n",
"cls = SymbolNet(\n",
" num_classes=len(kb.pseudo_label_list),\n",
" image_size=(28, 28, 1),\n",
")\n",
"device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
"criterion = nn.CrossEntropyLoss()\n",
"optimizer = torch.optim.RMSprop(cls.parameters(), lr=0.001, weight_decay=1e-6)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Pretrain NN classifier\n",
"framework_hed.hed_pretrain(kb, cls, recorder)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize BasicModel\n",
"# The function of BasicModel is to wrap NN models into the form of an sklearn estimator\n",
"base_model = BasicModel(\n",
" cls,\n",
" criterion,\n",
" optimizer,\n",
" device,\n",
" save_interval=1,\n",
" save_dir=recorder.save_dir,\n",
" batch_size=32,\n",
" num_epochs=1,\n",
" recorder=recorder,\n",
")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Use WABL model to join two parts"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"model = WABLBasicModel(base_model, kb.pseudo_label_list)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Dataset"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"total_train_data = get_hed(train=True)\n",
"train_data, val_data = split_equation(total_train_data, 3, 1)\n",
"test_data = get_hed(train=False)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Train and save"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"model, mapping = framework_hed.train_with_rule(model, abducer, train_data, val_data, select_num=10, min_len=5, max_len=8)\n",
"framework_hed.hed_test(model, abducer, mapping, train_data, test_data, min_len=5, max_len=8)\n",
"\n",
"recorder.dump()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "ABL",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.8.16"
},
"orig_nbformat": 4
},
"nbformat": 4,
"nbformat_minor": 2
}

+ 69
- 0
examples/hed/hed_knn_example.py View File

@@ -0,0 +1,69 @@
# coding: utf-8
# ================================================================#
# Copyright (C) 2021 Freecss All rights reserved.
#
# File Name ：share_example.py
# Author ：freecss
# Email ：karlfreecss@gmail.com
# Created Date ：2021/06/07
# Description ：
#
# ================================================================#

import sys
sys.path.append("../")

from abl.utils.plog import logger, INFO
from abl.utils.utils import reduce_dimension
import torch.nn as nn
import torch

from abl.models.nn import LeNet5, SymbolNet
from abl.models.basic_model import BasicModel, BasicDataset
from abl.models.wabl_models import DecisionTree, WABLBasicModel
from sklearn.neighbors import KNeighborsClassifier

from abl.abducer.abducer_base import AbducerBase
from abl.abducer.kb import add_KB, HWF_KB, prolog_KB
from datasets.mnist_add.get_mnist_add import get_mnist_add
from datasets.hwf.get_hwf import get_hwf
from datasets.hed.get_hed import get_hed, split_equation
from abl import framework_hed_knn


def run_test():

# kb = add_KB(True)
# kb = HWF_KB(True)
# abducer = AbducerBase(kb)

kb = prolog_KB(pseudo_label_list=[1, 0, '+', '='], pl_file='../examples/datasets/hed/learn_add.pl')
abducer = AbducerBase(kb, zoopt=True, multiple_predictions=True)

recorder = logger()

total_train_data = get_hed(train=True)
train_data, val_data = split_equation(total_train_data, 3, 1)
test_data = get_hed(train=False)

# ========================= KNN model ============================ #
reduce_dimension(train_data)
reduce_dimension(val_data)
reduce_dimension(test_data)
base_model = KNeighborsClassifier(n_neighbors=3)
pretrain_data_X, pretrain_data_Y = framework_hed_knn.hed_pretrain(base_model)
model = WABLBasicModel(base_model, kb.pseudo_label_list)
model, mapping = framework_hed_knn.train_with_rule(
model, abducer, train_data, val_data, (pretrain_data_X, pretrain_data_Y), select_num=10, min_len=5, max_len=8
)
framework_hed_knn.hed_test(
model, abducer, mapping, train_data, test_data, min_len=5, max_len=8
)
# ============================ End =============================== #

recorder.dump()
return True


if __name__ == "__main__":
run_test()

+ 47
- 0
examples/hed/utils.py View File

@@ -0,0 +1,47 @@
import torch
import torch.nn as nn
import numpy as np


def gen_mappings(chars, symbs):
n_char = len(chars)
n_symbs = len(symbs)
if n_char != n_symbs:
print("Characters and symbols size dosen't match.")
return
from itertools import permutations

mappings = []
# returned mappings
perms = permutations(symbs)
for p in perms:
mappings.append(dict(zip(chars, list(p))))
return mappings


def mapping_res(original_pred_res, m):
return [[m[symbol] for symbol in formula] for formula in original_pred_res]


def remapping_res(pred_res, m):
remapping = {}
for key, value in m.items():
remapping[value] = key
return [[remapping[symbol] for symbol in formula] for formula in pred_res]


def extract_feature(img):
extractor = nn.AvgPool2d(2, stride=2)
feature_map = np.array(extractor(torch.Tensor(img)))
return feature_map.reshape((-1,))


def reduce_dimension(data):
for truth_value in [0, 1]:
for equation_len in range(5, 27):
equations = data[truth_value][equation_len]
reduced_equations = [
[extract_feature(symbol_img) for symbol_img in equation]
for equation in equations
]
data[truth_value][equation_len] = reduced_equations

examples/weights/all_weights_here.txt → examples/hed/weights/all_weights_here.txt View File


examples/datasets/hwf/README.md → examples/hwf/datasets/README.md View File


examples/datasets/hwf/get_hwf.py → examples/hwf/datasets/get_hwf.py View File

@@ -12,7 +12,7 @@ def get_data(file, get_pseudo_label):
if get_pseudo_label:
Z = []
Y = []
img_dir = './datasets/hwf/data/Handwritten_Math_Symbols/'
img_dir = './datasets/data/Handwritten_Math_Symbols/'
with open(file) as f:
data = json.load(f)
for idx in range(len(data)):
@@ -36,9 +36,9 @@ def get_data(file, get_pseudo_label):

def get_hwf(train = True, get_pseudo_label = False):
if(train):
file = './datasets/hwf/data/expr_train.json'
file = './datasets/data/expr_train.json'
else:
file = './datasets/hwf/data/expr_test.json'
file = './datasets/data/expr_test.json'
return get_data(file, get_pseudo_label)


+ 184
- 0
examples/hwf/hwf_example.ipynb View File

@@ -0,0 +1,184 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import sys\n",
"\n",
"sys.path.append(\"../../\")\n",
"\n",
"import torch.nn as nn\n",
"import torch\n",
"\n",
"from abl.abducer.abducer_base import AbducerBase\n",
"from abl.abducer.kb import HWF_KB\n",
"\n",
"from abl.utils.plog import logger\n",
"from abl.models.basic_model import BasicModel\n",
"from abl.models.wabl_models import WABLBasicModel\n",
"\n",
"from models.nn import SymbolNet\n",
"from datasets.get_hwf import get_hwf\n",
"from abl import framework"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize logger\n",
"recorder = logger()"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Logic Part"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize knowledge base and abducer\n",
"kb = HWF_KB(GKB_flag=True)\n",
"abducer = AbducerBase(kb)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Machine Learning Part"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize necessary component for machine learning part\n",
"cls = SymbolNet(num_classes=len(kb.pseudo_label_list), image_size=(45, 45, 1))\n",
"device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
"criterion = nn.CrossEntropyLoss()\n",
"optimizer = torch.optim.Adam(cls.parameters(), lr=0.001, betas=(0.9, 0.99))"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize BasicModel\n",
"# The function of BasicModel is to wrap NN models into the form of an sklearn estimator\n",
"base_model = BasicModel(\n",
" cls,\n",
" criterion,\n",
" optimizer,\n",
" device,\n",
" save_interval=1,\n",
" save_dir=recorder.save_dir,\n",
" batch_size=32,\n",
" num_epochs=1,\n",
" recorder=recorder,\n",
")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Use WABL model to join two parts"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize WABL model\n",
"# The main function of the WABL model is to serialize data and \n",
"# provide a unified interface for different machine learning models\n",
"model = WABLBasicModel(base_model, kb.pseudo_label_list)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Dataset"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Get training and testing data\n",
"train_data = get_hwf(train=True, get_pseudo_label=True)\n",
"test_data = get_hwf(train=False, get_pseudo_label=True)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Train and save"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Train model\n",
"framework.train(\n",
" model, abducer, train_data, test_data, loop_num=15, sample_num=5000, verbose=1\n",
")\n",
"\n",
"# Save results\n",
"recorder.dump()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "ABL",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.8.16"
},
"orig_nbformat": 4
},
"nbformat": 4,
"nbformat_minor": 2
}

+ 0
- 0
examples/hwf/weights/all_weights_here.txt View File


examples/datasets/mnist_add/add.pl → examples/mnist_add/datasets/add.pl View File


examples/datasets/mnist_add/get_mnist_add.py → examples/mnist_add/datasets/get_mnist_add.py View File

@@ -1,6 +1,4 @@
import torch
import torchvision
from torch.utils.data import Dataset
from torchvision.transforms import transforms

def get_data(file, img_dataset, get_pseudo_label):
@@ -23,12 +21,12 @@ def get_data(file, img_dataset, get_pseudo_label):

def get_mnist_add(train = True, get_pseudo_label = False):
transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081, ))])
img_dataset = torchvision.datasets.MNIST(root='./datasets/mnist_add/', train=train, download=True, transform=transform)
img_dataset = torchvision.datasets.MNIST(root='./datasets/', train=train, download=True, transform=transform)
if train:
file = './datasets/mnist_add/train_data.txt'
file = './datasets/train_data.txt'
else:
file = './datasets/mnist_add/test_data.txt'
file = './datasets/test_data.txt'
return get_data(file, img_dataset, get_pseudo_label)

examples/datasets/mnist_add/test_data.txt → examples/mnist_add/datasets/test_data.txt View File


examples/datasets/mnist_add/train_data.txt → examples/mnist_add/datasets/train_data.txt View File


+ 190
- 0
examples/mnist_add/mnist_add_example.ipynb View File

@@ -0,0 +1,190 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import sys\n",
"\n",
"sys.path.append(\"../../\")\n",
"\n",
"import torch.nn as nn\n",
"import torch\n",
"\n",
"from abl.abducer.abducer_base import AbducerBase\n",
"from abl.abducer.kb import add_KB\n",
"\n",
"from abl.utils.plog import logger\n",
"from abl.models.basic_model import BasicModel\n",
"from abl.models.wabl_models import WABLBasicModel\n",
"\n",
"from models.nn import LeNet5\n",
"from datasets.get_mnist_add import get_mnist_add\n",
"from abl import framework"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize logger\n",
"recorder = logger()"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Logic Part"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize knowledge base and abducer\n",
"kb = add_KB(GKB_flag=True)\n",
"abducer = AbducerBase(kb, dist_func=\"confidence\")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Machine Learning Part"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize necessary component for machine learning part\n",
"cls = LeNet5(num_classes=len(kb.pseudo_label_list))\n",
"device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
"criterion = nn.CrossEntropyLoss()\n",
"optimizer = torch.optim.Adam(cls.parameters(), lr=0.001, betas=(0.9, 0.99))"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize BasicModel\n",
"# The function of BasicModel is to wrap NN models into the form of an sklearn estimator\n",
"base_model = BasicModel(\n",
" cls,\n",
" criterion,\n",
" optimizer,\n",
" device,\n",
" save_interval=1,\n",
" save_dir=recorder.save_dir,\n",
" batch_size=32,\n",
" num_epochs=1,\n",
" recorder=recorder,\n",
")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Use WABL model to join two parts"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Initialize WABL model\n",
"# The main function of the WABL model is to serialize data and \n",
"# provide a unified interface for different machine learning models\n",
"model = WABLBasicModel(base_model, kb.pseudo_label_list)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Dataset"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Get training and testing data\n",
"train_X, train_Z, train_Y = get_mnist_add(train=True, get_pseudo_label=True)\n",
"test_X, test_Z, test_Y = get_mnist_add(train=False, get_pseudo_label=True)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Train and save"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Train model\n",
"framework.train(\n",
" model,\n",
" abducer,\n",
" (train_X, train_Z, train_Y),\n",
" (test_X, test_Z, test_Y),\n",
" loop_num=15,\n",
" sample_num=5000,\n",
" verbose=1,\n",
")\n",
"\n",
"# Save results\n",
"recorder.dump()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "ABL",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.8.16"
},
"orig_nbformat": 4
},
"nbformat": 4,
"nbformat_minor": 2
}

+ 0
- 0
examples/mnist_add/weights/all_weights_here.txt View File


abl/models/nn.py → examples/models/nn.py View File

@@ -11,15 +11,10 @@
# ================================================================#


import torchvision

import torch
import numpy as np
from torch import nn
from torch.nn import functional as F
from torch.autograd import Variable
import torchvision.transforms as transforms
import numpy as np



class LeNet5(nn.Module):
@@ -56,36 +51,6 @@ class LeNet5(nn.Module):
return num_features


# class SymbolNet(nn.Module):
# def __init__(self, num_classes=4, image_size=(28, 28, 1)):
# super(SymbolNet, self).__init__()
# self.conv1 = nn.Sequential(
# nn.Conv2d(1, 32, 3, stride=1, padding=1),
# nn.ReLU(inplace=True),
# nn.BatchNorm2d(32),
# )
# self.conv2 = nn.Sequential(
# nn.Conv2d(32, 64, 3, stride=1, padding=1),
# nn.ReLU(inplace=True),
# nn.MaxPool2d(kernel_size=2, stride=2),
# nn.BatchNorm2d(64),
# nn.Dropout(0.25),
# )
# num_features = 64 * (image_size[0] // 2) * (image_size[1] // 2)
# self.fc1 = nn.Sequential(
# nn.Linear(num_features, 128), nn.ReLU(inplace=True), nn.Dropout(0.5)
# )
# self.fc2 = nn.Sequential(nn.Linear(128, num_classes), nn.Softmax(dim=1))

# def forward(self, x):
# x = self.conv1(x)
# x = self.conv2(x)
# x = torch.flatten(x, 1)
# x = self.fc1(x)
# x = self.fc2(x)
# return x


class SymbolNet(nn.Module):
def __init__(self, num_classes=4, image_size=(28, 28, 1)):
super(SymbolNet, self).__init__()
@@ -131,17 +96,3 @@ class SymbolNetAutoencoder(nn.Module):
x = self.fc1(x)
x = self.fc2(x)
return x


class MLP(nn.Module):
def __init__(self, input_dim=50, num_classes=2):
super(MLP, self).__init__()
assert input_dim > 0
hidden_dim = int(np.sqrt(input_dim))
self.fc1 = nn.Sequential(nn.Linear(input_dim, hidden_dim), nn.ReLU())
self.fc2 = nn.Sequential(nn.Linear(hidden_dim, num_classes), nn.Softmax(dim=1))

def forward(self, x):
x = self.fc1(x)
x = self.fc2(x)
return x

+ 0
- 97
examples/nonshare_example.py View File

@@ -1,97 +0,0 @@
# coding: utf-8
#================================================================#
# Copyright (C) 2021 Freecss All rights reserved.
#
# File Name ：nonshare_example.py
# Author ：freecss
# Email ：karlfreecss@gmail.com
# Created Date ：2021/06/07
# Description ：
#
#================================================================#

from utils.plog import logger
from models.wabl_models import DecisionTree, KNN
import pickle as pk
import numpy as np
import time
import framework

from multiprocessing import Pool
import os
from datasets.data_generator import generate_data_via_codes, code_generator
from collections import defaultdict
from abducer.abducer_base import AbducerBase
from abducer.kb import ClsKB, RegKB

def run_test(params):
code_len, times, code_num, share, model_type, need_prob, letter_num = params

if share:
result_dir = "share_result"
else:
result_dir = "non_share_result"

recoder_file_path = f"{result_dir}/random_{times}_{code_len}_{code_num}_{model_type}_{need_prob}.pk"

words = code_generator(code_len, code_num, letter_num)
kb = ClsKB(words)
abducer = AbducerBase(kb, dist_func = "confidence", pred_res_parse = lambda x : x["prob"])

label_lists = [[] for _ in range(code_len)]
for widx, word in enumerate(words):
for cidx, c in enumerate(word):
label_lists[cidx].append(c)

if share:
label_lists = [sum(label_lists, [])]

recoder = logger()
recoder.set_savefile("test.log")
for idx, err in enumerate(range(15, 41)):
start = time.process_time()
err = err / 40.
if 1 - err < (1. / letter_num):
break

print("Start expriment", idx)
if model_type == "KNN":
model = KNN(code_len, label_lists = label_lists, share=share)
elif model_type == "DT":
model = DecisionTree(code_len, label_lists = label_lists, share=share)

pre_X, pre_Y = generate_data_via_codes(words, err, letter_num)
X, Y = generate_data_via_codes(words, 0, letter_num)

str_words = ["".join(str(c) for c in word) for word in words]

recoder.print(str_words)

model.train(pre_X, pre_Y)
abl_epoch = 30
res = framework.train(model, abducer, X, Y, sample_num = 10000, verbose = 1)
print("Initial data accuracy:", 1 - err)
print("Abd word accuracy: ", res["accuracy_word"] * 1.0 / res["total_word"])
print("Abd char accuracy: ", res["accuracy_abd_char"] * 1.0 / res["total_abd_char"])
print("Ori char accuracy: ", res["accuracy_ori_char"] * 1.0 / res["total_ori_char"])
print("End expriment", idx)
print()

recoder.dump(open(recoder_file_path, "wb"))
return True

if __name__ == "__main__":
os.system("mkdir share_result")
os.system("mkdir non_share_result")
for times in range(5):
for code_num in [32, 64, 128]:
params = [11, times, code_num, False, "KNN", True, 2]
run_test(params)

params = [11, times, code_num, False, "KNN", False, 2]
run_test(params)

#params = [11, 0, 32, False, "DT", False, 2]
#run_test(params)


+ 0
- 96
examples/share_example.py View File

@@ -1,96 +0,0 @@
# coding: utf-8
#================================================================#
# Copyright (C) 2021 Freecss All rights reserved.
#
# File Name ：share_example.py
# Author ：freecss
# Email ：karlfreecss@gmail.com
# Created Date ：2021/06/07
# Description ：
#
#================================================================#

from utils.plog import logger
from models.wabl_models import DecisionTree, KNN
import pickle as pk
import numpy as np
import time
import framework

from multiprocessing import Pool
import os
from datasets.data_generator import generate_data_via_codes, code_generator
from collections import defaultdict
from abducer.abducer_base import AbducerBase
from abducer.kb import ClsKB, RegKB

def run_test(params):
code_len, times, code_num, share, model_type, need_prob, letter_num = params

if share:
result_dir = "share_result"
else:
result_dir = "non_share_result"

recoder_file_path = f"{result_dir}/random_{times}_{code_len}_{code_num}_{model_type}_{need_prob}.pk"#

words = code_generator(code_len, code_num, letter_num)
kb = ClsKB(words)
abducer = AbducerBase(kb)

label_lists = [[] for _ in range(code_len)]
for widx, word in enumerate(words):
for cidx, c in enumerate(word):
label_lists[cidx].append(c)

if share:
label_lists = [sum(label_lists, [])]

recoder = logger()
recoder.set_savefile("test.log")
for idx, err in enumerate(range(0, 41)):
print("Start expriment", idx)
start = time.process_time()
err = err / 40.
if 1 - err < (1. / letter_num):
break
if model_type == "KNN":
model = KNN(code_len, label_lists = label_lists, share=share)
elif model_type == "DT":
model = DecisionTree(code_len, label_lists = label_lists, share=share)

pre_X, pre_Y = generate_data_via_codes(words, err, letter_num)
X, Y = generate_data_via_codes(words, 0, letter_num)

str_words = ["".join(str(c) for c in word) for word in words]

recoder.print(str_words)

model.train(pre_X, pre_Y)
abl_epoch = 30
res = framework.train(model, abducer, X, Y, sample_num = 10000, verbose = 1)
print("Initial data accuracy:", 1 - err)
print("Abd word accuracy: ", res["accuracy_word"] * 1.0 / res["total_word"])
print("Abd char accuracy: ", res["accuracy_abd_char"] * 1.0 / res["total_abd_char"])
print("Ori char accuracy: ", res["accuracy_ori_char"] * 1.0 / res["total_ori_char"])
print("End expriment", idx)
print()

recoder.dump(open(recoder_file_path, "wb"))
return True

if __name__ == "__main__":
os.system("mkdir share_result")
os.system("mkdir non_share_result")
for times in range(5):
for code_num in [32, 64, 128]:
params = [11, times, code_num, True, "KNN", True, 2]
run_test(params)

params = [11, times, code_num, True, "KNN", False, 2]
run_test(params)

#params = [11, 0, 32, True, "DT", True, 2]
#run_test(params)


+ 0
- 407
framework_hed_knn.py View File

@@ -1,407 +0,0 @@
# coding: utf-8
# ================================================================#
# Copyright (C) 2021 Freecss All rights reserved.
#
# File Name ：framework.py
# Author ：freecss
# Email ：karlfreecss@gmail.com
# Created Date ：2021/06/07
# Description ：
#
# ================================================================#

import pickle as pk
import torch
import torch.nn as nn
import numpy as np
import os

from utils.plog import INFO, DEBUG, clocker
from utils.utils import (
flatten,
reform_idx,
block_sample,
gen_mappings,
mapping_res,
remapping_res,
extract_feature,
)

from models.nn import MLP, SymbolNetAutoencoder
from models.basic_model import BasicModel, BasicDataset
from datasets.hed.get_hed import get_pretrain_data


def result_statistics(pred_Z, Z, Y, logic_forward, char_acc_flag):
result = {}
if char_acc_flag:
char_acc_num = 0
char_num = 0
for pred_z, z in zip(pred_Z, Z):
char_num += len(z)
for zidx in range(len(z)):
if pred_z[zidx] == z[zidx]:
char_acc_num += 1
char_acc = char_acc_num / char_num
result["Character level accuracy"] = char_acc

abl_acc_num = 0
for pred_z, y in zip(pred_Z, Y):
if logic_forward(pred_z) == y:
abl_acc_num += 1
abl_acc = abl_acc_num / len(Y)
result["ABL accuracy"] = abl_acc

return result


def filter_data(X, abduced_Z):
finetune_Z = []
finetune_X = []
for abduced_x, abduced_z in zip(X, abduced_Z):
if abduced_z is not []:
finetune_X.append(abduced_x)
finetune_Z.append(abduced_z)
return finetune_X, finetune_Z


def hed_pretrain(cls, image_size=(28, 28, 1)):
import cv2

INFO("Pretrain Start")
pretrain_data_X, pretrain_data_Y = [], []
for i, label in enumerate(["0", "1", "10", "11"]):
label_path = os.path.join("./datasets/hed/dataset/mnist_images", label)
img_path_list = os.listdir(label_path)
for j in range(10):
img = cv2.imread(
os.path.join(label_path, img_path_list[j]), cv2.IMREAD_GRAYSCALE
)
img = np.array(cv2.resize(img, (image_size[1], image_size[0])), np.float32)
img = (img - 127) / 128.0
pretrain_data_X.append(
extract_feature(img.reshape((1, image_size[0], image_size[1])))
)
pretrain_data_Y.append(i)
cls.fit(pretrain_data_X, pretrain_data_Y)
import random

for i, label in enumerate(["0", "1", "10", "11"]):
label_path = os.path.join("./datasets/hed/dataset/mnist_images", label)
img_path_list = os.listdir(label_path)
cnt = 0
for j in range(50):
img = cv2.imread(
os.path.join(label_path, random.choice(img_path_list)),
cv2.IMREAD_GRAYSCALE,
)
img = np.array(cv2.resize(img, (image_size[1], image_size[0])), np.float32)
img = (img - 127) / 128.0
predict_label = cls.predict(
[extract_feature(img.reshape((1, image_size[0], image_size[1])))]
)
# predict_label = cls.predict_proba(
# [
# extract_feature(
# np.array(img, dtype=np.float32).reshape(
# (1, image_size[0], image_size[1])
# )
# )
# ]
# ).argmax(axis=1)

if predict_label == i:
cnt += 1
INFO(
"%d predict accuracy is " % i,
cnt / 50,
)

return pretrain_data_X, pretrain_data_Y


def _get_char_acc(model, X, consistent_pred_res, mapping):
original_pred_res = model.predict(X)["cls"]
pred_res = flatten(mapping_res(original_pred_res, mapping))
INFO("Current model's output: ", pred_res)
INFO("Abduced labels: ", flatten(consistent_pred_res))
assert len(pred_res) == len(flatten(consistent_pred_res))
return sum(
[
pred_res[idx] == flatten(consistent_pred_res)[idx]
for idx in range(len(pred_res))
]
) / len(pred_res)


def abduce_and_train(model, abducer, mapping, train_X_true, pretrain_data, select_num):
select_idx = np.random.randint(len(train_X_true), size=select_num)
X = []
for idx in select_idx:
X.append(train_X_true[idx])

original_pred_res = model.predict(X)["cls"]

if mapping == None:
mappings = gen_mappings(["+", "=", 0, 1], ["+", "=", 0, 1])
else:
mappings = [mapping]

consistent_idx = []
consistent_pred_res = []

for m in mappings:
pred_res = mapping_res(original_pred_res, m)
max_abduce_num = 20
solution = abducer.zoopt_get_solution(
pred_res, [1] * len(pred_res), max_abduce_num
)
all_address_flag = reform_idx(solution, pred_res)

consistent_idx_tmp = []
consistent_pred_res_tmp = []

for idx in range(len(pred_res)):
address_idx = [
i for i, flag in enumerate(all_address_flag[idx]) if flag != 0
]
candidate = abducer.kb.address_by_idx([pred_res[idx]], 1, address_idx, True)
if len(candidate) > 0:
consistent_idx_tmp.append(idx)
consistent_pred_res_tmp.append(candidate[0][0])

if len(consistent_idx_tmp) > len(consistent_idx):
consistent_idx = consistent_idx_tmp
consistent_pred_res = consistent_pred_res_tmp
if len(mappings) > 1:
mapping = m

if len(consistent_idx) == 0:
return 0, 0, None

if len(mappings) > 1:
INFO("Final mapping is: ", mapping)

INFO("Train pool size is:", len(flatten(consistent_pred_res)))
INFO("Start to use abduced pseudo label to train model...")
pretrain_data_X, pretrain_data_Y = pretrain_data
pretrain_mappping = {0: 0, 1: 1, 2: "+", 3: "="}
pretrain_data_X = [[X] for X in pretrain_data_X]
pretrain_data_Y = [[pretrain_mappping[Y]] for Y in pretrain_data_Y]
model.train(
[X[idx] for idx in consistent_idx] + pretrain_data_X,
remapping_res(consistent_pred_res + pretrain_data_Y, mapping),
)

consistent_acc = len(consistent_idx) / select_num
char_acc = _get_char_acc(
model, [X[idx] for idx in consistent_idx], consistent_pred_res, mapping
)
INFO("consistent_acc is %s, char_acc is %s" % (consistent_acc, char_acc))
return consistent_acc, char_acc, mapping


def _remove_duplicate_rule(rule_dict):
add_nums_dict = {}
for r in list(rule_dict):
add_nums = str(r.split("]")[0].split("[")[1]) + str(
r.split("]")[1].split("[")[1]
) # r = 'my_op([1], [0], [1, 0])' then add_nums = '10'
if add_nums in add_nums_dict:
old_r = add_nums_dict[add_nums]
if rule_dict[r] >= rule_dict[old_r]:
rule_dict.pop(old_r)
add_nums_dict[add_nums] = r
else:
rule_dict.pop(r)
else:
add_nums_dict[add_nums] = r
return list(rule_dict)


def get_rules_from_data(
model, abducer, mapping, train_X_true, samples_per_rule, samples_num
):
rules = []
for _ in range(samples_num):
while True:
select_idx = np.random.randint(len(train_X_true), size=samples_per_rule)
X = []
for idx in select_idx:
X.append(train_X_true[idx])
original_pred_res = model.predict(X)["cls"]
pred_res = mapping_res(original_pred_res, mapping)

consistent_idx = []
consistent_pred_res = []
for idx in range(len(pred_res)):
if abducer.kb.logic_forward([pred_res[idx]]):
consistent_idx.append(idx)
consistent_pred_res.append(pred_res[idx])

if len(consistent_pred_res) != 0:
rule = abducer.abduce_rules(consistent_pred_res)
if rule != None:
break
rules.append(rule)

all_rule_dict = {}
for rule in rules:
for r in rule:
all_rule_dict[r] = 1 if r not in all_rule_dict else all_rule_dict[r] + 1
rule_dict = {rule: cnt for rule, cnt in all_rule_dict.items() if cnt >= 5}
rules = _remove_duplicate_rule(rule_dict)

return rules


def _get_consist_rule_acc(model, abducer, mapping, rules, X):
cnt = 0
for x in X:
original_pred_res = model.predict([x])["cls"]
pred_res = flatten(mapping_res(original_pred_res, mapping))
if abducer.kb.consist_rule(pred_res, rules):
cnt += 1
return cnt / len(X)


def train_with_rule(
model,
abducer,
train_data,
val_data,
pretrain_data,
select_num=10,
min_len=5,
max_len=8,
):
train_X = train_data
val_X = val_data

samples_num = 50
samples_per_rule = 3

# Start training / for each length of equations
for equation_len in range(min_len, max_len):
INFO(
"============== equation_len: %d-%d ================"
% (equation_len, equation_len + 1)
)
train_X_true = train_X[1][equation_len]
train_X_false = train_X[0][equation_len]
val_X_true = val_X[1][equation_len]
val_X_false = val_X[0][equation_len]

train_X_true.extend(train_X[1][equation_len + 1])
train_X_false.extend(train_X[0][equation_len + 1])
val_X_true.extend(val_X[1][equation_len + 1])
val_X_false.extend(val_X[0][equation_len + 1])

condition_cnt = 0
while True:
if equation_len == min_len:
mapping = None

# Abduce and train NN
consistent_acc, char_acc, mapping = abduce_and_train(
model, abducer, mapping, train_X_true, pretrain_data, select_num
)
if consistent_acc == 0:
continue

# Test if we can use mlp to evaluate
if consistent_acc >= 0.9 and char_acc >= 0.9:
condition_cnt += 1
else:
condition_cnt = 0

# The condition has been satisfied continuously five times
if condition_cnt >= 5:
INFO("Now checking if we can go to next course")
rules = get_rules_from_data(
model, abducer, mapping, train_X_true, samples_per_rule, samples_num
)
INFO("Learned rules from data:", rules)

true_consist_rule_acc = _get_consist_rule_acc(
model, abducer, mapping, rules, val_X_true
)
false_consist_rule_acc = _get_consist_rule_acc(
model, abducer, mapping, rules, val_X_false
)

INFO(
"consist_rule_acc is %f, %f\n"
% (true_consist_rule_acc, false_consist_rule_acc)
)
# decide next course or restart
if true_consist_rule_acc > 0.9 and false_consist_rule_acc < 0.1:
break
else:
if equation_len == min_len:
# model.cls_list[0].model.load_state_dict(
# torch.load("./weights/pretrain_weights.pth")
# )
pretrain_data_X, pretrain_data_Y = pretrain_data
model.cls_list[0].fit(pretrain_data_X, pretrain_data_Y)
else:
pretrain_data_X, pretrain_data_Y = pretrain_data
model.cls_list[0].fit(pretrain_data_X, pretrain_data_Y)
# model.cls_list[0].model.load_state_dict(
# torch.load("./weights/weights_%d.pth" % (equation_len - 1))
# )
condition_cnt = 0
INFO("Reload Model and retrain")

return model, mapping


def hed_test(model, abducer, mapping, train_data, test_data, min_len=5, max_len=8):
train_X = train_data
test_X = test_data

# Calcualte how many equations should be selected in each length
# for each length, there are equation_samples_num[equation_len] rules
print("Now begin to train final mlp model")
equation_samples_num = []
len_cnt = max_len - min_len + 1
samples_num = 50
equation_samples_num += [0] * min_len
if samples_num % len_cnt == 0:
equation_samples_num += [samples_num // len_cnt] * len_cnt
else:
equation_samples_num += [samples_num // len_cnt] * len_cnt
equation_samples_num[-1] += samples_num % len_cnt
assert sum(equation_samples_num) == samples_num

# Abduce rules
rules = []
samples_per_rule = 3
for equation_len in range(min_len, max_len + 1):
equation_rules = get_rules_from_data(
model,
abducer,
mapping,
train_X[1][equation_len],
samples_per_rule,
equation_samples_num[equation_len],
)
rules.extend(equation_rules)
rules = list(set(rules))
INFO("Learned rules from data:", rules)

for equation_len in range(5, 27):
true_consist_rule_acc = _get_consist_rule_acc(
model, abducer, mapping, rules, test_X[1][equation_len]
)
false_consist_rule_acc = _get_consist_rule_acc(
model, abducer, mapping, rules, test_X[0][equation_len]
)
INFO(
"consist_rule_acc of testing length %d equations are %f, %f"
% (equation_len, true_consist_rule_acc, false_consist_rule_acc)
)


if __name__ == "__main__":
pass

+ 4
- 4
tests/test_models.py View File

@@ -8,7 +8,7 @@ import torch
import torch.nn as nn
import numpy as np

from abl.models.nn import LeNet5, SymbolNet
from examples.models.nn import LeNet5, SymbolNet
from abl.models.basic_model import BasicModel


@@ -39,7 +39,7 @@ class TestBasicModel(object):
self._test_fit()
self._test_predict()
self._test_predict_proba()
self._test_val()
self._test_score()
self._test_save()
self._test_load()

@@ -58,8 +58,8 @@ class TestBasicModel(object):
assert predict_result.shape == (5, self.num_classes)
assert (0 <= predict_result).all() and (predict_result <= 1).all()

def _test_val(self):
accuracy = self.model.val(X=self.data_X, y=self.data_y)
def _test_score(self):
accuracy = self.model.score(X=self.data_X, y=self.data_y)
assert type(accuracy) == float
assert 0 <= accuracy <= 1



Write Preview
Loading…
Cancel
Save