Source code for alts.core.query.query_selector

#Version 1.1.1 conform as of 13.12.2024
"""
| *alts.core.query.query_selector*
"""
from __future__ import annotations
from typing import TYPE_CHECKING

from dataclasses import dataclass

from alts.core.subscriber import ProcessDataSubscriber, ResultDataSubscriber, StreamDataSubscriber
from alts.core.query.query_optimizer import QueryOptimizer
from alts.core.query.query_decider import QueryDecider
from alts.core.configuration import Required, init
from alts.core.experiment_module import ExperimentModule
from alts.core.data.constrains import QueryConstrained



if TYPE_CHECKING:
    from typing_extensions import Self #type: ignore
    from alts.core.configuration import Required
    from alts.core.subscribable import Subscribable


[docs]
@dataclass
class QuerySelector(ExperimentModule, QueryConstrained):
    """
    QuerySelector(query_optimizer, query_decider)
    | **Description**
    |   The QuerySelector chooses which queries are going to be asked results for next.
    |   It does so with the help of the QueryOptimizer and the QueryDecider.

    :param query_optimizer: Chooses the most informative queries to be asked next
    :type query_optimizer: QueryOptimizer
    :param query_decider: Decided if the chosen queries are informative enough to be asked
    :type query_decider: QueryDecider
    """
    query_optimizer: QueryOptimizer = init()
    query_decider: QueryDecider = init()


[docs]
    def post_init(self):
        """
        post_init(self) -> None
        | **Description**
        |   Initializes ``query_optimizer`` and ``query_Decider`` with its experiment modules.
        """
        super().post_init()
        self.query_optimizer = self.query_optimizer(exp_modules = self.exp_modules)
        self.query_decider = self.query_decider(exp_modules = self.exp_modules)




[docs]
    def decide(self):
        """
        decide(self) -> None
        | **Description**
        |   1. Chooses the best queries with the optimizer
        |   2. Decides if chosen queries are worth the effort
        |   3. If they are worth it, forwards the chosen queries to the oracle
        """
        query_candidates, scores = self.query_optimizer.select()
        query_flag, queries = self.query_decider.decide(query_candidates, scores)
        if query_flag:
            self.oracles.add(queries)




[docs]
class ResultQuerySelector(QuerySelector, ResultDataSubscriber):
    """
    ResultQuerySelector(query_optimizer, query_decider)
    | **Description**
    |   Has additional access to a result data pool foor decision making.

    :param query_optimizer: Chooses the most informative queries to be asked next
    :type query_optimizer: QueryOptimizer
    :param query_decider: Decided if the chosen queries are informative enough to be asked
    :type query_decider: QueryDecider
    """

[docs]
    def result_update(self, subscription: Subscribable):
        super().result_update(subscription)
        self.decide()




[docs]
class StreamQuerySelector(QuerySelector, StreamDataSubscriber):
    """
    StreamQuerySelector(query_optimizer, query_decider)
    | **Description**
    |   Has additional access to a stream for decision making.

    :param query_optimizer: Chooses the most informative queries to be asked next
    :type query_optimizer: QueryOptimizer
    :param query_decider: Decided if the chosen queries are informative enough to be asked
    :type query_decider: QueryDecider
    """

[docs]
    def stream_update(self, subscription: Subscribable):
        super().stream_update(subscription)
        self.decide()




[docs]
class ProcessQuerySelector(QuerySelector, ProcessDataSubscriber):
    """
    ProcessQuerySelector(query_optimizer, query_decider)
    | **Description**
    |   Has additional access to a process for decision making.

    :param query_optimizer: Chooses the most informative queries to be asked next
    :type query_optimizer: QueryOptimizer
    :param query_decider: Decided if the chosen queries are informative enough to be asked
    :type query_decider: QueryDecider
    """

[docs]
    def process_update(self, subscription: Subscribable):
        super().process_update(subscription)
        self.decide()