"""CFKD teacher that gathers human feedback through the ViReLay web GUI.
ViReLay (with CoRelAy for the spectral analysis) is a tool for browsing
attribution maps grouped by clusters. This teacher reuses it for counterfactual
feedback: for every counterfactual that flipped the student it computes a
per-channel relevance vector from the difference of an intermediate student
activation between factual and counterfactual, writes ViReLay's HDF5 dataset,
attribution and analysis databases plus a project yaml into ``base_dir``, serves
the project and waits for the user to save the selected (i.e. "false") data
point indices as ``feedback.json``. The helper functions and CoRelAy processors
in this module are adapted from the ViReLay example scripts.
"""
import shutil
import torch
import json
import h5py
import yaml
import os
import time
import multiprocessing
import numpy as np
from tqdm import tqdm
from os.path import dirname, relpath
from typing import List
from corelay.base import Param
from corelay.processor.base import Processor
from corelay.processor.affinity import SparseKNN
from corelay.processor.distance import SciPyPDist
from corelay.processor.flow import Sequential, Parallel
from corelay.pipeline.spectral import SpectralClustering
from corelay.processor.embedding import TSNEEmbedding, UMAPEmbedding, EigenDecomposition
from corelay.processor.clustering import KMeans, DBSCAN, AgglomerativeClustering
from virelay.server import Server
from virelay.model import Workspace
from PIL import Image
from torchvision.transforms import ToTensor
from itertools import product
from pathlib import Path
from peal.global_utils import load_yaml_config, is_port_in_use, get_intermediate_output
from peal.teachers.interfaces import TeacherInterface
from peal.log import get_logger
_log = get_logger(__name__)
[docs]
class ImageDataset(torch.utils.data.Dataset):
"""All images of one directory as RGB tensors with a dummy label of 0.
Used to turn the copied collage images into the HDF5 dataset ViReLay
displays. Files are listed with ``os.listdir`` (unsorted).
Parameters
----------
root_dir : str
Directory whose every file is opened as an image.
transform : callable
Applied to the PIL image; only the first three channels of the result
are kept.
"""
[docs]
def __init__(self, root_dir, transform=ToTensor()):
"""
Args:
root_dir (string): Directory with all the images.
transform (callable, optional): Optional transform to be applied
on a sample.
"""
self.root_dir = root_dir
self.transform = transform
self.urls = os.listdir(self.root_dir)
def __len__(self):
return len(self.urls)
def __getitem__(self, idx):
"""Return ``(image[:3], tensor(0.0))`` for the ``idx``-th file."""
file = self.urls[idx]
img = Image.open(os.path.join(self.root_dir, file))
img = self.transform(img)[:3]
return img, torch.tensor(0.0)
[docs]
def create_dataset(
dataset_file_path: str,
samples_shape: tuple,
number_of_samples: int,
dataloader: torch.utils.data.DataLoader,
) -> h5py.File:
"""Creates a new dataset HDF5 file.
The file gets a float32 ``data`` dataset of shape
``(number_of_samples, *samples_shape)`` and a uint16 ``label`` dataset,
both filled by iterating once over ``dataloader``.
Parameters
----------
dataset_file_path : str
The path to the dataset HDF5 file that is to be created.
samples_shape : tuple
The shape of one sample (without the batch axis).
number_of_samples : int
The number of samples in the dataset.
dataloader : torch.utils.data.DataLoader
Yields ``(X, y)`` batches that are written in order.
Returns
-------
h5py.File
The still open file handle of the dataset database.
"""
dataset_file = h5py.File(dataset_file_path, "w")
dataset_file.create_dataset(
"data", shape=(number_of_samples,) + tuple(samples_shape), dtype="float32"
)
dataset_file.create_dataset("label", shape=(number_of_samples,), dtype="uint16")
index = 0
for X, y in dataloader:
dataset_file["data"][index : X.shape[0] + index] = X.detach().numpy()
dataset_file["label"][index : y.shape[0] + index] = y.detach().numpy()
index += X.shape[0]
return dataset_file
[docs]
def create_attribution_database(
attribution_database_file_path, attribution_shape, num_classes, number_of_samples
):
"""
Creates an attribution database file
Parameters
----------
attribution_database_file_path : str
Path of the HDF5 file to create (overwritten).
attribution_shape : tuple
Shape of one attribution (without the batch axis).
num_classes : int
Width of the ``prediction`` dataset.
number_of_samples : int
Length of the ``attribution``, ``prediction`` and ``label`` datasets.
Returns
-------
h5py.File
The open file with empty float32 ``attribution`` and ``prediction``
datasets and a uint16 ``label`` dataset.
"""
attribution_database_file = h5py.File(attribution_database_file_path, "w")
attribution_database_file.create_dataset(
"attribution",
shape=(number_of_samples,) + tuple(attribution_shape),
dtype="float32",
)
attribution_database_file.create_dataset(
"prediction", shape=(number_of_samples, num_classes), dtype="float32"
)
attribution_database_file.create_dataset(
"label", shape=(number_of_samples,), dtype="uint16"
)
return attribution_database_file
[docs]
def append_attributions(
attribution_database_file, index, attributions, predictions, labels
):
"""
Appends attributions to the attribution database file
Parameters
----------
attribution_database_file : h5py.File
File created by :func:`create_attribution_database`.
index : int
Row at which the batch is written.
attributions : torch.Tensor
Batch of attributions ``[B, *attribution_shape]``.
predictions : torch.Tensor
Batch of prediction scores ``[B, num_classes]``.
labels : torch.Tensor
Batch of labels ``[B]``.
"""
attribution_database_file["attribution"][
index : attributions.shape[0] + index
] = attributions.detach().numpy()
attribution_database_file["prediction"][
index : predictions.shape[0] + index
] = predictions.detach().numpy()
attribution_database_file["label"][
index : labels.shape[0] + index
] = labels.detach().numpy()
[docs]
def create_attribution_dataset(dataloader, attribution_database_file_path, num_classes):
"""Write a sequence of single attributions into a new attribution database.
Parameters
----------
dataloader : sequence of tuple
Indexable sequence of ``(attribution, prediction, label)`` triples;
``attribution`` is a tensor, the other two scalars. Its length is the
number of samples and ``dataloader[0][0].shape`` the attribution shape.
attribution_database_file_path : str
Path of the HDF5 file to create.
num_classes : int
Width of the ``prediction`` dataset.
"""
number_of_samples = len(dataloader)
with create_attribution_database(
attribution_database_file_path=attribution_database_file_path,
attribution_shape=dataloader[0][0].shape,
num_classes=num_classes,
number_of_samples=number_of_samples,
) as attribution_database_file:
number_of_samples_processed = 0
for attributions, predictions, labels in dataloader:
append_attributions(
attribution_database_file=attribution_database_file,
index=number_of_samples_processed,
attributions=attributions.unsqueeze(0),
predictions=torch.tensor([predictions]),
labels=torch.tensor([labels]),
)
number_of_samples_processed += 1
[docs]
class Flatten(Processor):
"""Represents a CoRelAy processor, which flattens its input data."""
[docs]
def function(self, data: np.ndarray) -> np.ndarray:
"""Applies the flattening to the input data.
Parameters
----------
data: np.ndarray
The input data that is to be flattened.
Returns
-------
np.ndarray
Returns the flattened data.
"""
return data.reshape(data.shape[0], np.prod(data.shape[1:]))
[docs]
class SumChannel(Processor):
"""Represents a CoRelAy processor, which sums its input data across channels, i.e., its second axis."""
[docs]
def function(self, data: np.ndarray) -> np.ndarray:
"""Applies the summation over the channels to the input data.
Parameters
----------
data: np.ndarray
The input data that is to be summed over its channels.
Returns
-------
np.ndarray
Returns the data that was summed up over its channels.
"""
return data.sum(axis=1)
[docs]
class Absolute(Processor):
"""Represents a CoRelAy processor, which computes the absolute value of its input data."""
[docs]
def function(self, data: np.ndarray) -> np.ndarray:
"""Computes the absolute value of the specified input data.
Parameters
----------
data: np.ndarray
The input data for which the absolute value is to be computed.
Returns
-------
np.ndarray
Returns the absolute value of the input data.
"""
return np.absolute(data)
[docs]
class Normalize(Processor):
"""Represents a CoRelAy processor, which normalizes its input data.
Attributes
----------
axes: Param
A parameter of the processor, which determines the axis over which the data is to be normalized. Defaults to
the second and third axes.
"""
axes = Param(tuple, (1, 2))
[docs]
def function(self, data: np.ndarray) -> np.ndarray:
"""Normalizes the specified input data.
Parameters
----------
data: np.ndarray
The input data that is to be normalized.
Returns
-------
np.ndarray
Returns the normalized input data.
"""
return data / data.sum(self.axes, keepdims=True)
[docs]
class Histogram(Processor):
"""Represents a CoRelAy processor, which computes a histogram over its input data.
Attributes
----------
bins: Param
A parameter of the processor, which determines the number of bins that are used to compute the histogram.
"""
bins = Param(int, 256)
[docs]
def function(self, data: np.ndarray) -> np.ndarray:
"""Computes histograms over the specified input data. One histogram is computed for each channel and each sample
in a batch of input data.
Parameters
----------
data: np.ndarray
The input data over which the histograms are to be computed.
Returns
-------
np.ndarray
Returns the histograms that were computed over the input data.
"""
return np.stack(
[
np.stack(
[
np.histogram(
sample.reshape(sample.shape[0], np.prod(sample.shape[1:3])),
bins=self.bins,
density=True,
)
for sample in channel
]
)
for channel in data.transpose(3, 0, 1, 2)
]
)
# Contains the various pre-processing method and distance metric variants that can be used to compute the analysis
VARIANTS = {
"absspectral": {
"preprocessing": Sequential([Absolute(), SumChannel(), Normalize(), Flatten()]),
"distance": SciPyPDist(metric="euclidean"),
},
"spectral": {
"preprocessing": Sequential([SumChannel(), Normalize(), Flatten()]),
"distance": SciPyPDist(metric="euclidean"),
},
"fullspectral": {
"preprocessing": Sequential([Normalize(axes=(1, 2, 3)), Flatten()]),
"distance": SciPyPDist(metric="euclidean"),
},
"histogram": {
"preprocessing": Sequential(
[Normalize(axes=(1, 2, 3)), Histogram(), Flatten()]
),
"distance": SciPyPDist(metric="euclidean"),
},
}
[docs]
def make_project(
dataset_file_path: str,
attribution_database_file_path: str,
analysis_file_path: str,
label_map_file_path: str,
project_name: str,
dataset_name: str,
dataset_down_sampling_method: str,
dataset_up_sampling_method: str,
model_name: str,
attribution_name: str,
analysis_name: str,
output_file_path: str,
) -> None:
"""Generates a ViRelAy project file.
Parameters
----------
dataset_file_path: str
The path to the dataset HDF5 file.
attribution_database_file_path: str
The path to the attribution HDF5 file.
analysis_file_path: str
The path to the analysis HDF5 file.
label_map_file_path: str
The path to the label map YAML file.
project_name: str
The name of the project.
dataset_name: str
The name of the dataset that the classifier was trained on.
dataset_down_sampling_method: str
The method that is to be used to down-sample images from the dataset that are larger than the input to the
model. Must be one of "none", "center_crop", or "resize".
dataset_up_sampling_method: str
The method that is to be used to up-sample images from the dataset that are smaller than the input to the
model. Must be one of "none", "fill_zeros", "fill_ones", "edge_repeat", "mirror_edge", "wrap_around", or
"resize".
model_name: str
The name of the classifier model on which the project is based.
attribution_name: str
The name of the method that was used to compute the attributions.
analysis_name: str
The name of the analysis that was performed on the attributions.
output_file_path: str
The path to the YAML file into which the project will be saved.
"""
# Determines the root path of the project, which is needed to make all paths stored in the project file relative to
# the project file
if output_file_path is not None:
project_root_path = dirname(output_file_path)
else:
project_root_path = "."
if label_map_file_path is None:
label_map_file_path = os.path.join(project_root_path, "label_map.json")
# Determines the shape of the dataset samples
with h5py.File(dataset_file_path, "r") as dataset_file:
input_shape = dataset_file["data"].shape # pylint: disable=no-member
# Generates the information, which will be stored in the project file
project = {
"project": {
"name": project_name,
"model": model_name,
"label_map": relpath(label_map_file_path, start=project_root_path),
"dataset": {
"name": dataset_name,
"type": "hdf5",
"path": relpath(dataset_file_path, start=project_root_path),
"input_width": input_shape[2],
"input_height": input_shape[3],
"down_sampling_method": dataset_down_sampling_method,
"up_sampling_method": dataset_up_sampling_method,
},
"attributions": {
"attribution_method": attribution_name,
"attribution_strategy": "true_label",
"sources": [
relpath(attribution_database_file_path, start=project_root_path)
],
},
"analyses": [
{
"analysis_method": analysis_name,
"sources": [relpath(analysis_file_path, start=project_root_path)],
}
],
}
}
# If an output file path was specified, then the project is saved into the specified file, otherwise, the project
# information is written to the standard output
if output_file_path is None:
_log.info("%s", yaml.dump(project, default_flow_style=False))
else:
with open(output_file_path, "w", encoding="utf-8") as project_file:
yaml.dump(project, project_file, default_flow_style=False)
[docs]
def run_virelay(project_path: str, output_path: str, port: int) -> None:
"""Serve a ViReLay project until the user has saved their feedback file.
The ViReLay server runs in a separate process bound to ``0.0.0.0``; the
function polls once per second for ``output_path`` and terminates the
server once the file exists.
Parameters
----------
project_path : str
The project yaml written by :func:`make_project`.
output_path : str
JSON file the user exports from the GUI; must contain
``selectedDataPointIndices``.
port : int
Port to serve on.
Returns
-------
dict
The parsed contents of ``output_path``.
"""
host_name = "0.0.0.0"
workspace = Workspace()
workspace.add_project(project_path)
app = Server(workspace)
# thread = threading.Thread(target=lambda: app.run(host=host_name, port=port))
# thread.start()
proc = multiprocessing.Process(
target=lambda: app.run(host=host_name, port=port), args=()
)
proc.start()
_log.info("%s", "ViReLay GUI is active on localhost:" + str(port))
with tqdm(range(100000)) as pbar:
for it in pbar:
if os.path.exists(output_path):
time.sleep(1.0)
feedback = json.load(open(output_path, "r"))
break
else:
pbar.set_description(
"Give feedback at localhost:"
+ str(port)
+ " and save under "
+ output_path
+ " when done."
)
time.sleep(1.0)
proc.terminate()
return feedback
[docs]
class VirelayTeacher(TeacherInterface):
"""Human teacher whose verdicts come from a ViReLay cluster selection.
Parameters
----------
port : int
First port for the ViReLay server; incremented while occupied.
dataset : peal.data.datasets.ImageDataset, optional
Unused by this teacher beyond being stored.
tracking_level : int
Verbosity level of the surrounding run (stored only).
num_classes : int
Stored only; the analysis currently runs with a single class.
teacher_config : str
Yaml with ``meta_analysis_variant``, ``number_of_eigenvalues``,
``number_of_clusters_list``, ``number_of_neighbors``,
``dataset_down_sampling_method`` and ``dataset_up_sampling_method``.
distance_from_last_layer : int
Which intermediate student output (counted from the end, see
``get_intermediate_output``) the relevance vectors are taken from.
"""
[docs]
def __init__(
self,
port=2000,
dataset=None,
tracking_level=5,
num_classes=2,
teacher_config="configs/cfkd_experiments/teachers/virelay_teacher.yaml",
distance_from_last_layer=27, # 15, #3, #39, #2, #50,
):
"""Store the settings and load ``teacher_config``; see the class docstring."""
self.port = port
self.dataset = dataset
self.tracking_level = tracking_level
self.num_classes = num_classes
self.teacher_config = load_yaml_config(teacher_config).__dict__
self.distance_from_last_layer = distance_from_last_layer
[docs]
def get_feedback(
self,
collage_path_list,
x_counterfactual_list,
x_list,
y_list,
base_dir,
student,
y_source_list,
y_target_end_confidence_list,
**kwargs,
):
"""Build a ViReLay project for the flipped counterfactuals and collect verdicts.
Only counterfactuals with ``y_target_end_confidence > 0.5`` are shown.
For each of them the difference of the student's intermediate output
(layer ``distance_from_last_layer``) between counterfactual and factual
is summed spatially and L1-normalised over channels to form a
``[1, C, 1, 1]`` relevance vector. ``base_dir`` is wiped and filled with
``collage_greater_than_05/``, ``database.hdf5``,
``attribution_database.hdf5``, ``analysis.hdf5`` and ``analysis.yaml``
(each step is skipped if its file already exists), then ViReLay is
served until ``feedback.json`` appears.
Parameters
----------
collage_path_list : list of str
Collage image per counterfactual.
x_counterfactual_list : list of torch.Tensor
Counterfactual inputs.
x_list : list of torch.Tensor
Factual inputs.
y_list, y_source_list : list
Labels and source classes; currently not used for the analysis.
base_dir : str
Working directory for the ViReLay project (recreated).
student : torch.nn.Module
The classifier whose intermediate activations are attributed.
y_target_end_confidence_list : list of float
Target-class confidence of every counterfactual.
**kwargs
Further tracked values, ignored.
Returns
-------
list of str
One entry per input counterfactual: ``"false"`` if its index was in
``selectedDataPointIndices`` of the feedback file, ``"true"`` for
the other shown ones and ``"Student not flipped"`` for those with
confidence at most 0.5.
"""
# filter all lists to only contain only values where y_target_end_confidence_list is greater than 0.5
# first create a list of indices where y_target_end_confidence_list is greater than 0.5
y_target_confidence_greater_than_05_indices = [
i
for i in range(len(y_target_end_confidence_list))
if y_target_end_confidence_list[i] > 0.5
]
collage_path_greater_than_05_list = [
collage_path_list[i]
for i in range(len(y_target_end_confidence_list))
if y_target_end_confidence_list[i] > 0.5
]
x_counterfactual_greater_than_05_list = [
x_counterfactual_list[i]
for i in range(len(y_target_end_confidence_list))
if y_target_end_confidence_list[i] > 0.5
]
x_greater_than_05_list = [
x_list[i]
for i in range(len(y_target_end_confidence_list))
if y_target_end_confidence_list[i] > 0.5
]
y_greater_than_05_list = [
0
for i in range(len(y_target_end_confidence_list))
if y_target_end_confidence_list[i] > 0.5
]
y_source_greater_than_05_list = [
0
for i in range(len(y_target_end_confidence_list))
if y_target_end_confidence_list[i] > 0.5
]
attributions = []
for i in range(1, 1000):
try:
x_latent = get_intermediate_output(
student,
x_greater_than_05_list[0].unsqueeze(0).to("cuda"),
i,
)
_log.info("%s", str(i) + ": " + str(x_latent.shape))
except Exception:
break
for i in range(len(x_greater_than_05_list)):
x_latent = (
get_intermediate_output(
student,
x_greater_than_05_list[i].unsqueeze(0).to("cuda"),
self.distance_from_last_layer,
)
.detach()
.cpu()
)
cf_latent = (
get_intermediate_output(
student,
x_counterfactual_greater_than_05_list[i].unsqueeze(0).to("cuda"),
self.distance_from_last_layer,
)
.detach()
.cpu()
)
attribution_difference = cf_latent - x_latent
# attribution_difference = attribution_difference.squeeze(0).flatten().unsqueeze(-1).unsqueeze(-1)
rel = torch.sum(
attribution_difference.view(*attribution_difference.shape[:2], -1),
dim=-1,
)
rel = rel / (torch.abs(rel).sum(-1).view(-1, 1) + 1e-10)
rel = rel.unsqueeze(-1).unsqueeze(-1)
attributions.append(rel)
_log.info("%s", "attributions[0].shape")
_log.info("%s", "attributions[0].shape")
_log.info("%s", "attributions[0].shape")
_log.info("%s", attributions[0].shape)
shutil.rmtree(base_dir, ignore_errors=True)
Path(base_dir).mkdir(parents=True, exist_ok=True)
if not os.path.exists(os.path.join(base_dir, "database.hdf5")):
# write collage_path_greater_than_05_list in the os.path.join(base_dir, "collage_greater_than_05") folder
collage_path_greater_than_05_dir = os.path.join(
base_dir, "collage_greater_than_05"
)
os.makedirs(collage_path_greater_than_05_dir, exist_ok=True)
for i, collage_path in enumerate(collage_path_greater_than_05_list):
# copy the collage_path to the collage_path_greater_than_05_dir
os.system(
f"cp {collage_path} {collage_path_greater_than_05_dir}/{i}.jpg"
)
collage_path_list_dataloader = torch.utils.data.DataLoader(
ImageDataset(collage_path_greater_than_05_dir),
)
create_dataset(
dataset_file_path=os.path.join(base_dir, "database.hdf5"),
samples_shape=collage_path_list_dataloader.dataset[0][0].shape,
number_of_samples=len(collage_path_list_dataloader.dataset),
dataloader=collage_path_list_dataloader,
)
# TODO get the actual prediction scores of the network
if not os.path.exists(os.path.join(base_dir, "attribution_database.hdf5")):
attributions_dataloader = list(
zip(attributions, y_source_greater_than_05_list, y_greater_than_05_list)
)
create_attribution_dataset(
attribution_database_file_path=os.path.join(
base_dir, "attribution_database.hdf5"
),
dataloader=attributions_dataloader,
num_classes=1, # self.num_classes,
)
if not os.path.exists(os.path.join(base_dir, "analysis.hdf5")):
meta_analysis(
base_dir=base_dir,
attribution_database_file_path=os.path.join(
base_dir, "attribution_database.hdf5"
),
analysis_file_path=os.path.join(base_dir, "analysis.hdf5"),
variant=self.teacher_config["meta_analysis_variant"],
class_indices=range(1), # range(self.num_classes),
label_map_file_path=None,
number_of_eigenvalues=self.teacher_config["number_of_eigenvalues"],
number_of_clusters_list=self.teacher_config["number_of_clusters_list"],
number_of_neighbors=self.teacher_config["number_of_neighbors"],
)
if not os.path.exists(os.path.join(base_dir, "analysis.yaml")):
make_project(
dataset_file_path=os.path.join(base_dir, "database.hdf5"),
attribution_database_file_path=os.path.join(
base_dir, "attribution_database.hdf5"
),
analysis_file_path=os.path.join(base_dir, "analysis.hdf5"),
label_map_file_path=None,
project_name="analysis",
dataset_name="dataset",
dataset_down_sampling_method=self.teacher_config[
"dataset_down_sampling_method"
],
dataset_up_sampling_method=self.teacher_config[
"dataset_up_sampling_method"
],
model_name="model",
attribution_name="attribution",
analysis_name="analysis",
output_file_path=os.path.join(base_dir, "analysis.yaml"),
)
# TODO check if port is free
if not os.path.exists(os.path.join(base_dir, "feedback.json")):
while is_port_in_use(self.port):
_log.info("%s", "port " + str(self.port) + " is occupied!")
self.port += 1
run_virelay(
project_path=os.path.join(base_dir, "analysis.yaml"),
port=self.port,
output_path=os.path.join(base_dir, "feedback.json"),
)
feedback = json.load(open(os.path.join(base_dir, "feedback.json"), "r"))
# set the default feedback out to "Student not flipped" for the full length collage list
feedback_out = ["Student not flipped" for _ in range(len(collage_path_list))]
for i in range(len(collage_path_greater_than_05_list)):
if i in feedback["selectedDataPointIndices"]:
feedback_out[y_target_confidence_greater_than_05_indices[i]] = "false"
else:
feedback_out[y_target_confidence_greater_than_05_indices[i]] = "true"
return feedback_out