cbrkit.revise

 33@dataclass(slots=True, frozen=True)
 34class build[K, V, S: Float](ReviserFunc[K, V, S]):
 35    """Builds a reviser that assesses solution quality and optionally repairs solutions.
 36
 37    The reviser first optionally repairs solutions using the repair function,
 38    then assesses their quality using the assess function (a similarity function
 39    comparing solutions to queries).
 40
 41    Args:
 42        assess_func: Similarity function to evaluate solution quality.
 43        repair_func: Optional adaptation function to repair solutions before assessment.
 44        multiprocessing: Multiprocessing configuration for repair.
 45        chunksize: Number of batches to process at a time using the repair function.
 46            If 0, it will be set to the number of batches divided by the number of processes.
 47
 48    Returns:
 49        The revised casebases with quality scores.
 50
 51    Examples:
 52        >>> import cbrkit
 53        >>> reviser = build(
 54        ...     assess_func=cbrkit.sim.generic.equality(),
 55        ... )
 56    """
 57
 58    assess_func: MaybeFactory[AnySimFunc[V, S]]
 59    repair_func: MaybeFactory[SimpleAdaptationFunc[V]] | None = None
 60    multiprocessing: Pool | int | bool = False
 61    chunksize: int = 0
 62
 63    @override
 64    def __call__(
 65        self,
 66        batches: Sequence[tuple[Casebase[K, V], V]],
 67    ) -> Sequence[tuple[Casebase[K, V], SimMap[K, S]]]:
 68        current_batches = batches
 69
 70        # Step 1: Optionally repair solutions
 71        if self.repair_func is not None:
 72            repair_func = produce_factory(self.repair_func)
 73            repaired_casebases = self._repair(current_batches, repair_func)
 74            current_batches = [
 75                (repaired_casebase, query)
 76                for repaired_casebase, (_, query) in zip(
 77                    repaired_casebases, current_batches, strict=True
 78                )
 79            ]
 80
 81        # Step 2: Assess quality of all solutions
 82        assess_func = batchify_sim(produce_factory(self.assess_func))
 83
 84        flat_batches: list[tuple[V, V]] = []
 85        flat_index: list[tuple[int, K]] = []
 86
 87        for idx, (casebase, query) in enumerate(current_batches):
 88            for key, solution in casebase.items():
 89                flat_index.append((idx, key))
 90                flat_batches.append((solution, query))
 91
 92        quality_scores = assess_func(flat_batches)
 93
 94        quality_maps: list[dict[K, S]] = [{} for _ in current_batches]
 95        for (idx, key), quality in zip(flat_index, quality_scores, strict=True):
 96            quality_maps[idx][key] = quality
 97
 98        return [
 99            (casebase, quality_map)
100            for (casebase, _), quality_map in zip(
101                current_batches, quality_maps, strict=True
102            )
103        ]
104
105    def _repair(
106        self,
107        batches: Sequence[tuple[Casebase[K, V], V]],
108        repair_func: SimpleAdaptationFunc[V],
109    ) -> Sequence[Casebase[K, V]]:
110        batch_repair_func = batchify_adaptation(repair_func)
111        batches_index: list[tuple[int, K]] = []
112        flat_batches: list[tuple[V, V]] = []
113
114        for idx, (casebase, query) in enumerate(batches):
115            for key, case in casebase.items():
116                batches_index.append((idx, key))
117                flat_batches.append((case, query))
118
119        repaired_cases: Sequence[V]
120
121        if use_mp(self.multiprocessing) or self.chunksize > 0:
122            chunksize = (
123                self.chunksize
124                if self.chunksize > 0
125                else len(flat_batches) // mp_count(self.multiprocessing)
126            )
127            batch_chunks = list(chunkify(flat_batches, chunksize))
128            repaired_chunks = mp_map(
129                batch_repair_func, batch_chunks, self.multiprocessing, logger
130            )
131            repaired_cases = list(itertools.chain.from_iterable(repaired_chunks))
132        else:
133            repaired_cases = list(batch_repair_func(flat_batches))
134
135        repaired_casebases: list[dict[K, V]] = [{} for _ in range(len(batches))]
136
137        for (idx, key), repaired_case in zip(
138            batches_index, repaired_cases, strict=True
139        ):
140            repaired_casebases[idx][key] = repaired_case
141
142        return repaired_casebases

17def apply_result[Q, C, V, S: Float](
18    result: Result[Q, C, V, S] | ResultStep[Q, C, V, S],
19    revisers: MaybeFactories[ReviserFunc[C, V, S]],
20) -> Result[Q, C, V, S]:
21    """Applies reviser functions to a previous result.
22
23    Args:
24        result: The result whose solutions will be revised.
25        revisers: The reviser functions that will be applied.
26
27    Returns:
28        Returns an object of type Result.
29    """
30    if isinstance(result, ResultStep):
31        result = Result(steps=[result], duration=0.0)
32
33    if not produce_sequence(revisers):
34        return result
35
36    return apply_batches(
37        {
38            query_key: (entry.casebase, entry.query)
39            for query_key, entry in result.queries.items()
40        },
41        revisers,
42    )

45def apply_batches[Q, C, V, S: Float](
46    batch: Mapping[Q, tuple[Mapping[C, V], V]],
47    revisers: MaybeFactories[ReviserFunc[C, V, S]],
48) -> Result[Q, C, V, S]:
49    """Applies batches containing a casebase and a query using reviser functions.
50
51    Args:
52        batch: A mapping of queries to casebases and queries.
53        revisers: Reviser functions that will assess and optionally repair solutions.
54
55    Returns:
56        Returns an object of type Result.
57    """
58    reviser_factories = produce_sequence(revisers)
59    steps: list[ResultStep[Q, C, V, S]] = []
60    current_batches: Mapping[Q, tuple[Mapping[C, V], V]] = batch
61
62    loop_start_time = default_timer()
63
64    for i, reviser_factory in enumerate(reviser_factories, start=1):
65        reviser_func = produce_factory(reviser_factory)
66        logger.info(f"Processing reviser {i}/{len(reviser_factories)}")
67        start_time = default_timer()
68        queries_results = reviser_func(list(current_batches.values()))
69        end_time = default_timer()
70
71        step_queries = {
72            query_key: QueryResultStep(
73                similarities=revised_sims,
74                casebase=revised_casebase,
75                query=current_batches[query_key][1],
76                duration=0.0,
77            )
78            for query_key, (revised_casebase, revised_sims) in zip(
79                current_batches.keys(), queries_results, strict=True
80            )
81        }
82
83        steps.append(
84            ResultStep(
85                queries=step_queries,
86                metadata=get_metadata(reviser_func),
87                duration=end_time - start_time,
88            )
89        )
90        current_batches = {
91            query_key: (step_queries[query_key].casebase, step_queries[query_key].query)
92            for query_key in current_batches.keys()
93        }
94
95    return Result(steps=steps, duration=default_timer() - loop_start_time)

 98def apply_queries[Q, C, V, S: Float](
 99    casebase: Mapping[C, V],
100    queries: Mapping[Q, V],
101    revisers: MaybeFactories[ReviserFunc[C, V, S]],
102) -> Result[Q, C, V, S]:
103    """Applies queries to a casebase using reviser functions.
104
105    Args:
106        casebase: The casebase containing solutions to revise.
107        queries: The queries used for revision.
108        revisers: The reviser functions that will be applied.
109
110    Returns:
111        Returns an object of type Result.
112    """
113    return apply_batches(
114        {query_key: (casebase, query) for query_key, query in queries.items()},
115        revisers,
116    )

141def apply_query[K, V, S: Float](
142    casebase: Casebase[K, V],
143    query: V,
144    revisers: MaybeFactories[ReviserFunc[K, V, S]],
145) -> Result[str, K, V, S]:
146    """Applies a single query to a casebase using reviser functions.
147
148    Args:
149        casebase: The casebase containing solutions to revise.
150        query: The query used for revision.
151        revisers: The reviser functions that will be applied.
152
153    Returns:
154        Returns an object of type Result.
155    """
156    return apply_queries(casebase, {"default": query}, revisers)

119def apply_pair[V, S: Float](
120    case: V,
121    query: V,
122    revisers: MaybeFactories[ReviserFunc[str, V, S]],
123) -> Result[str, str, V, S]:
124    """Applies a single query to a single case using reviser functions.
125
126    Args:
127        case: The case that will be revised.
128        query: The query used for revision.
129        revisers: The reviser functions that will be applied.
130
131    Returns:
132        Returns an object of type Result.
133    """
134    return apply_queries(
135        {"default": case},
136        {"default": query},
137        revisers,
138    )

113class Result[Q, C, V, S: Float](BaseModel):
114    """Complete result containing all pipeline steps."""
115
116    model_config = ConfigDict(frozen=True)
117    steps: list[ResultStep[Q, C, V, S]]
118    duration: float
119
120    @property
121    def first_step(self) -> ResultStep[Q, C, V, S]:
122        """Return the first step of the pipeline."""
123        return self.steps[0]
124
125    @property
126    def final_step(self) -> ResultStep[Q, C, V, S]:
127        """Return the last step of the pipeline."""
128        return self.steps[-1]
129
130    @property
131    def metadata(self) -> JsonEntry:
132        """Return the metadata from the final step."""
133        return self.final_step.metadata
134
135    @property
136    def queries(self) -> Mapping[Q, QueryResultStep[C, V, S]]:
137        """Return the query results from the final step."""
138        return self.final_step.queries
139
140    @property
141    def default_query(self) -> QueryResultStep[C, V, S]:
142        """Return the single query result from the final step."""
143        return singleton(self.queries.values())
144
145    @property
146    def similarities(self) -> SimMap[C, S]:
147        """Return the similarity map from the final step."""
148        return self.final_step.similarities
149
150    @property
151    def ranking(self) -> Sequence[C]:
152        """Return the case ranking from the final step."""
153        return self.final_step.ranking
154
155    @property
156    def casebase(self) -> Casebase[C, V]:
157        """Return the casebase from the final step."""
158        return self.final_step.casebase
159
160    @property
161    def similarity(self) -> S:
162        """Return the similarity value from the final step's single case."""
163        return self.final_step.similarity
164
165    @property
166    def case(self) -> V:
167        """Return the case value from the final step's single case."""
168        return self.final_step.case
169
170    def remove_cases(self) -> "Result[Q, C, None, S]":
171        """Return a copy of this result with all case data removed."""
172        return Result[Q, C, None, S](
173            steps=[step.remove_cases() for step in self.steps],
174            duration=self.duration,
175        )

 66class ResultStep[Q, C, V, S: Float](BaseModel):
 67    """Aggregated result step across multiple queries."""
 68
 69    model_config = ConfigDict(frozen=True)
 70    queries: Mapping[Q, QueryResultStep[C, V, S]]
 71    metadata: JsonEntry
 72    duration: float
 73
 74    @property
 75    def default_query(self) -> QueryResultStep[C, V, S]:
 76        """Return the single query result when only one query exists."""
 77        return singleton(self.queries.values())
 78
 79    @property
 80    def similarities(self) -> SimMap[C, S]:
 81        """Return the similarity map of the default query."""
 82        return self.default_query.similarities
 83
 84    @property
 85    def ranking(self) -> Sequence[C]:
 86        """Return the case ranking of the default query."""
 87        return self.default_query.ranking
 88
 89    @property
 90    def casebase(self) -> Casebase[C, V]:
 91        """Return the casebase of the default query."""
 92        return self.default_query.casebase
 93
 94    @property
 95    def similarity(self) -> S:
 96        """Return the similarity value of the default query's single case."""
 97        return self.default_query.similarity
 98
 99    @property
100    def case(self) -> V:
101        """Return the case value of the default query's single case."""
102        return self.default_query.case
103
104    def remove_cases(self) -> "ResultStep[Q, C, None, S]":
105        """Return a copy of this result step with all case data removed."""
106        return ResultStep[Q, C, None, S](
107            queries={key: value.remove_cases() for key, value in self.queries.items()},
108            metadata=self.metadata,
109            duration=self.duration,
110        )

16class QueryResultStep[K, V, S: Float](BaseModel):
17    """Result of a single query with similarities, ranking, and matched cases."""
18
19    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
20    similarities: SimMap[K, S]
21    ranking: Sequence[K] = ()
22    casebase: Casebase[K, V]
23    query: V
24    duration: float
25
26    @model_validator(mode="before")
27    @classmethod
28    def _custom_validator(cls, data: dict[str, Any]) -> dict[str, Any]:
29        assert len(data["similarities"]) == len(data["casebase"]), (
30            "similarities and casebase must have equal length"
31        )
32
33        if not data.get("ranking"):
34            data["ranking"] = tuple(sim_map2ranking(data["similarities"]))
35
36        return data
37
38    @property
39    def casebase_similarities(self) -> Mapping[K, tuple[V, S]]:
40        """Return a mapping of case keys to their case-similarity pairs."""
41        return {
42            key: (self.casebase[key], self.similarities[key]) for key in self.ranking
43        }
44
45    @property
46    def similarity(self) -> S:
47        """Return the similarity value when only a single case exists."""
48        return singleton(self.similarities.values())
49
50    @property
51    def case(self) -> V:
52        """Return the case value when only a single case exists."""
53        return singleton(self.casebase.values())
54
55    def remove_cases(self) -> "QueryResultStep[K, None, S]":
56        """Return a copy of this result step with all case data removed."""
57        return QueryResultStep[K, None, S](
58            similarities=self.similarities,
59            ranking=self.ranking,
60            casebase={},
61            query=None,
62            duration=self.duration,
63        )