26616170697

Committed 29 May 2026 03:28AM UTC coverage: 75.347%. First build

Build # 26616170697

Build Type

Pull #360

github

Committed by

web-flow

Commit Message

Merge 101146123 into 936e3dd5a

Pull Request Pull Request #360: feat(drift): add ``from_cli`` constructor and add docs

Coverage Stats

137 of 154 new or added lines in 27 files covered. (88.96%)

7228 of 9593 relevant lines covered (75.35%)

0.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.82

/src/capymoa/_cli.py

"""Internal functions for generating CLI creation strings for MOA objects."""

from capymoa.base import MOAClassifier, MOARegressor, MOAPredictionIntervalLearner
from capymoa.drift.base_detector import MOADriftDetector
from capymoa.stream import MOAStream

from moa.streams import InstanceStream as _InstanceStream
from moa.options import AbstractOptionHandler as _AbstractOptionHandler
from moa.classifiers import AbstractClassifier as _AbstractClassifier
from moa.classifiers import Regressor as _AbstractRegressor
from moa.classifiers.core.driftdetection import (
    AbstractChangeDetector as _AbstractChangeDetector,
)
from moa.classifiers.predictioninterval import (
    PredictionIntervalLearner as _PredictionIntervalLearner,
)
from typing import Type


def object_cli_str(
    object: _AbstractOptionHandler, type_: Type[_AbstractOptionHandler]
) -> str:
    """Return a CLI string for creating MOA objects.

    >>> from moa.classifiers.trees import HoeffdingTree
    >>> from moa.classifiers import AbstractClassifier
    >>> object_cli_str(HoeffdingTree(), AbstractClassifier)
    '(trees.HoeffdingTree)'

    :param object: MOA object to generate CLI string for.
    :param type_: Type of the MOA object, used to strip the fully qualified name to be
        more concise.
    :return: CLI string for creating the MOA object.
    """
    return f"({str(object.getCLICreationString(type_)).strip()})"


def classifier_cli_str(classifier: _AbstractClassifier | MOAClassifier) -> str:
    """Return a CLI string for creating a MOA classifier.

    >>> from moa.classifiers.trees import HoeffdingTree
    >>> classifier_cli_str(HoeffdingTree())
    '(trees.HoeffdingTree)'

    """
    if isinstance(classifier, MOAClassifier):
        return object_cli_str(classifier.moa_learner, _AbstractClassifier)
    elif isinstance(classifier, _AbstractClassifier):
        return object_cli_str(classifier, _AbstractClassifier)
    else:
        raise ValueError("Unknown Type")


def regressor_cli_str(regressor: _AbstractRegressor | MOARegressor) -> str:
    """Return a CLI string for creating a MOA regressor.

    >>> from moa.classifiers.trees import HoeffdingTree
    >>> regressor_cli_str(HoeffdingTree())
    '(trees.HoeffdingTree)'

    """
    if isinstance(regressor, MOARegressor):
        # TODO: type ignore should be removed if regressor.moa_learner is properly typed
        return object_cli_str(regressor.moa_learner, _AbstractRegressor)  # type: ignore
    elif isinstance(regressor, _AbstractRegressor):
        return object_cli_str(regressor, _AbstractRegressor)  # type: ignore
    elif isinstance(regressor, (_AbstractClassifier, MOAClassifier)):
        # Some regressor are just classifiers.
        return classifier_cli_str(regressor)
    else:
        raise ValueError("Unknown Type")


def drift_detector_cli_str(
    detector: _AbstractChangeDetector | MOADriftDetector,
) -> str:
    """Return a CLI string for creating a MOA drift detector.

    >>> from moa.classifiers.core.driftdetection import ADWINChangeDetector
    >>> drift_detector_cli_str(ADWINChangeDetector())
    '(ADWINChangeDetector)'

    """
    if isinstance(detector, MOADriftDetector):
        return object_cli_str(detector.moa_detector, _AbstractChangeDetector)
    elif isinstance(detector, _AbstractChangeDetector):
        return object_cli_str(detector, _AbstractChangeDetector)
    else:
        raise ValueError("Unknown Type")


def prediction_interval_cli_str(
    predictor: _PredictionIntervalLearner | MOAPredictionIntervalLearner,
) -> str:
    """Return a CLI string for creating a MOA prediction interval learner.

    >>> from moa.classifiers.predictioninterval import MVEPredictionInterval
    >>> prediction_interval_cli_str(MVEPredictionInterval())
    '(MVEPredictionInterval...

    """
    if isinstance(predictor, MOAPredictionIntervalLearner):
        return object_cli_str(predictor.moa_learner, _PredictionIntervalLearner)  # type: ignore
    elif isinstance(predictor, _PredictionIntervalLearner):
        return object_cli_str(predictor, _PredictionIntervalLearner)
    else:
        raise ValueError("Unknown Type")


def stream_cli_str(stream: MOAStream | _InstanceStream) -> str:
    """Return a CLI string for creating a MOA stream.

    >>> from moa.streams.generators import RandomRBFGenerator
    >>> stream_cli_str(RandomRBFGenerator())
    '(generators.RandomRBFGenerator)'

    """
    if isinstance(stream, MOAStream):
        return object_cli_str(stream.moa_stream, _InstanceStream)
    elif isinstance(stream, _InstanceStream):
        return object_cli_str(stream, _InstanceStream)
    else:
        raise ValueError("Unknown Type")

1	"""Internal functions for generating CLI creation strings for MOA objects."""
2
3	from capymoa.base import MOAClassifier, MOARegressor, MOAPredictionIntervalLearner	1✔
4	from capymoa.drift.base_detector import MOADriftDetector	1✔
5	from capymoa.stream import MOAStream	1✔
6
7	from moa.streams import InstanceStream as _InstanceStream	1✔
8	from moa.options import AbstractOptionHandler as _AbstractOptionHandler	1✔
9	from moa.classifiers import AbstractClassifier as _AbstractClassifier	1✔
10	from moa.classifiers import Regressor as _AbstractRegressor	1✔
11	from moa.classifiers.core.driftdetection import (	1✔
12	AbstractChangeDetector as _AbstractChangeDetector,
13	)
14	from moa.classifiers.predictioninterval import (	1✔
15	PredictionIntervalLearner as _PredictionIntervalLearner,
16	)
17	from typing import Type	1✔
18
19
20	def object_cli_str(	1✔
21	object: _AbstractOptionHandler, type_: Type[_AbstractOptionHandler]
22	) -> str:
23	"""Return a CLI string for creating MOA objects.
24
25	>>> from moa.classifiers.trees import HoeffdingTree
26	>>> from moa.classifiers import AbstractClassifier
27	>>> object_cli_str(HoeffdingTree(), AbstractClassifier)
28	'(trees.HoeffdingTree)'
29
30	:param object: MOA object to generate CLI string for.
31	:param type_: Type of the MOA object, used to strip the fully qualified name to be
32	more concise.
33	:return: CLI string for creating the MOA object.
34	"""
35	return f"({str(object.getCLICreationString(type_)).strip()})"	1✔
36
37
38	def classifier_cli_str(classifier: _AbstractClassifier \| MOAClassifier) -> str:	1✔
39	"""Return a CLI string for creating a MOA classifier.
40
41	>>> from moa.classifiers.trees import HoeffdingTree
42	>>> classifier_cli_str(HoeffdingTree())
43	'(trees.HoeffdingTree)'
44
45	"""
46	if isinstance(classifier, MOAClassifier):	1✔
47	return object_cli_str(classifier.moa_learner, _AbstractClassifier)	1✔
48	elif isinstance(classifier, _AbstractClassifier):	1✔
49	return object_cli_str(classifier, _AbstractClassifier)	1✔
50	else:
NEW 51	raise ValueError("Unknown Type")	×
52
53
54	def regressor_cli_str(regressor: _AbstractRegressor \| MOARegressor) -> str:	1✔
55	"""Return a CLI string for creating a MOA regressor.
56
57	>>> from moa.classifiers.trees import HoeffdingTree
58	>>> regressor_cli_str(HoeffdingTree())
59	'(trees.HoeffdingTree)'
60
61	"""
62	if isinstance(regressor, MOARegressor):	1✔
63	# TODO: type ignore should be removed if regressor.moa_learner is properly typed
64	return object_cli_str(regressor.moa_learner, _AbstractRegressor) # type: ignore	1✔
65	elif isinstance(regressor, _AbstractRegressor):	1✔
NEW 66	return object_cli_str(regressor, _AbstractRegressor) # type: ignore	×
67	elif isinstance(regressor, (_AbstractClassifier, MOAClassifier)):	1✔
68	# Some regressor are just classifiers.
69	return classifier_cli_str(regressor)	1✔
70	else:
NEW 71	raise ValueError("Unknown Type")	×
72
73
74	def drift_detector_cli_str(	1✔
75	detector: _AbstractChangeDetector \| MOADriftDetector,
76	) -> str:
77	"""Return a CLI string for creating a MOA drift detector.
78
79	>>> from moa.classifiers.core.driftdetection import ADWINChangeDetector
80	>>> drift_detector_cli_str(ADWINChangeDetector())
81	'(ADWINChangeDetector)'
82
83	"""
84	if isinstance(detector, MOADriftDetector):	1✔
85	return object_cli_str(detector.moa_detector, _AbstractChangeDetector)	1✔
86	elif isinstance(detector, _AbstractChangeDetector):	1✔
87	return object_cli_str(detector, _AbstractChangeDetector)	1✔
88	else:
NEW 89	raise ValueError("Unknown Type")	×
90
91
92	def prediction_interval_cli_str(	1✔
93	predictor: _PredictionIntervalLearner \| MOAPredictionIntervalLearner,
94	) -> str:
95	"""Return a CLI string for creating a MOA prediction interval learner.
96
97	>>> from moa.classifiers.predictioninterval import MVEPredictionInterval
98	>>> prediction_interval_cli_str(MVEPredictionInterval())
99	'(MVEPredictionInterval...
100
101	"""
102	if isinstance(predictor, MOAPredictionIntervalLearner):	1✔
NEW 103	return object_cli_str(predictor.moa_learner, _PredictionIntervalLearner) # type: ignore	×
104	elif isinstance(predictor, _PredictionIntervalLearner):	1✔
105	return object_cli_str(predictor, _PredictionIntervalLearner)	1✔
106	else:
NEW 107	raise ValueError("Unknown Type")	×
108
109
110	def stream_cli_str(stream: MOAStream \| _InstanceStream) -> str:	1✔
111	"""Return a CLI string for creating a MOA stream.
112
113	>>> from moa.streams.generators import RandomRBFGenerator
114	>>> stream_cli_str(RandomRBFGenerator())
115	'(generators.RandomRBFGenerator)'
116
117	"""
118	if isinstance(stream, MOAStream):	1✔
NEW 119	return object_cli_str(stream.moa_stream, _InstanceStream)	×
120	elif isinstance(stream, _InstanceStream):	1✔
121	return object_cli_str(stream, _InstanceStream)	1✔
122	else:
NEW 123	raise ValueError("Unknown Type")	×

adaptive-machine-learning / CapyMOA / 26616170697

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous