abd-clam 0.11.1

Clustered Learning of Approximate Manifolds
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

import pathlib
import typing

import numpy
import pydantic
from matplotlib import pyplot


class TreeReport(pydantic.BaseModel):
    data_name: str
    metric_name: str
    cardinality: int
    dimensionality: int
    root_name: str
    max_depth: int
    build_time: float


class ClusterReport(pydantic.BaseModel):
    cardinality: int
    indices: typing.Union[list[int], None]
    name: str
    arg_center: int
    arg_radius: int
    radius: float
    lfd: float
    left_child: typing.Union[str, None]
    right_child: typing.Union[str, None]
    # ratios: list[float]


def load_tree(base_path: pathlib.Path) -> list[ClusterReport]:
    return [
        ClusterReport.parse_file(path)
        for path in base_path.iterdir()
        if path.is_file() and "tree" not in path.name
    ]


class RnnReport(pydantic.BaseModel):
    data_name: str
    metric_name: str
    num_queries: int
    num_runs: int
    cardinality: int
    dimensionality: int
    tree_depth: int
    build_time: float
    root_radius: float
    search_radii: numpy.ndarray
    search_times: numpy.ndarray
    outputs: list[list[int]]
    recalls: numpy.ndarray

    # noinspection PyMethodParameters
    @pydantic.validator("search_radii", pre=True)
    def parse_search_radii(search_radii):
        return numpy.array(search_radii, dtype=numpy.float64)

    # noinspection PyMethodParameters
    @pydantic.validator("search_times", pre=True)
    def parse_search_times(search_times):
        return numpy.array(search_times, dtype=numpy.float64)

    # noinspection PyMethodParameters
    @pydantic.validator("recalls", pre=True)
    def parse_recalls(recalls):
        return numpy.array(recalls, dtype=numpy.float64)

    class Config:
        arbitrary_types_allowed = True

    def is_valid(self):
        reasons: list[str] = []

        if self.num_queries != len(self.search_radii):
            reasons.append(
                f"self.num_queries != len(self.search_radii): {self.num_queries} != {len(self.search_radii)}",
            )

        if self.num_queries != len(self.search_times):
            reasons.append(
                f"self.num_queries != len(self.search_times): {self.num_queries} != {len(self.search_times)}",
            )

        if self.num_queries != len(self.outputs):
            reasons.append(
                f"self.num_queries != len(self.outputs): {self.num_queries} != {len(self.outputs)}",
            )

        if self.num_queries != len(self.recalls):
            reasons.append(
                f"self.num_queries != len(self.recalls): {self.num_queries} != {len(self.recalls)}",
            )

        for i, samples in enumerate(self.search_times, start=1):
            if self.num_runs != len(samples):
                reasons.append(
                    f"{i}/{self.num_queries}: self.num_runs != samples.len(): {self.num_runs} != {len(samples)}",
                )

        return reasons

    @property
    def output_sizes(self) -> numpy.ndarray:
        return numpy.asarray(list(map(len, self.outputs)))

    def _plot_xy(self, ax: pyplot.Axes, x: numpy.ndarray, y: numpy.ndarray):
        ax.scatter(x, y, s=0.2)

    def _plot_search_radii_vs_search_times(self, ax: pyplot.Axes) -> pyplot.Axes:
        self._plot_xy(ax, self.search_radii, numpy.mean(self.search_times, axis=1))
        ax.set_yscale("log")
        ax.set_xlabel("Search Radius")
        ax.set_ylabel("Search Time (sec)")
        return ax

    def _plot_search_radii_vs_output_sizes(self, ax: pyplot.Axes) -> pyplot.Axes:
        self._plot_xy(ax, self.search_radii, self.output_sizes)
        ax.set_xlabel("Search Radius")
        ax.set_ylabel("Output Size")
        return ax

    def _plot_search_radii_vs_recalls(self, ax: pyplot.Axes) -> pyplot.Axes:
        self._plot_xy(ax, self.search_radii, self.recalls)
        ax.set_xlabel("Search Radius")
        ax.set_ylabel("Recall")
        ax.set_ylim(0, 1.1)
        return ax

    def _plot_output_sizes_vs_search_times(self, ax: pyplot.Axes) -> pyplot.Axes:
        self._plot_xy(ax, self.output_sizes, numpy.mean(self.search_times, axis=1))
        ax.set_xlabel("Output Size")
        ax.set_ylabel("Search Time (sec)")
        return ax

    def _plot_output_sizes_vs_recalls(self, ax: pyplot.Axes) -> pyplot.Axes:
        self._plot_xy(ax, self.output_sizes, self.recalls)
        ax.set_xlabel("Output Size")
        ax.set_ylabel("Recall")
        return ax

    def plot(
        self,
        show: bool,
        output_dir: pathlib.Path,
    ):
        figure, ((ax00, ax01), (ax10, ax11)) = pyplot.subplots(2, 2, figsize=(8.5, 11))
        # figure.suptitle(f'Ranged Nearest Neighbors Search')
        figure.suptitle(
            f"{self.data_name}, {self.metric_name}, ({self.cardinality}, {self.dimensionality})",
        )

        self._plot_search_radii_vs_search_times(ax00)
        self._plot_output_sizes_vs_recalls(ax01)
        self._plot_search_radii_vs_recalls(ax10)
        self._plot_output_sizes_vs_search_times(ax11)

        if show:
            pyplot.show()
        else:
            figure.savefig(
                output_dir.joinpath(
                    f"rnn-search-{self.data_name}-{self.metric_name}.png",
                ),
                dpi=300,
            )

        pyplot.close(figure)