30801214514

Committed 03 Aug 2026 09:23AM UTC coverage: 76.112% (+0.7%) from 75.386%

Build # 30801214514

Build Type

push

github

Committed by

web-flow

Commit Message

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

Coverage Stats

160 of 192 new or added lines in 27 files covered. (83.33%)

1 existing line in 1 file now uncovered.

7669 of 10076 relevant lines covered (76.11%)

0.76 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 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
    >>> 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 cli_str_classifier(classifier: _AbstractClassifier | MOAClassifier) -> str:
    """Return a CLI string for creating a MOA classifier.

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

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


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

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

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


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

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

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


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

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

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


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

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

    """
    if isinstance(stream, MOAStream):
        return cli_str(stream.moa_stream, _InstanceStream)
    elif isinstance(stream, _InstanceStream):
        return 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 cli_str(object: _AbstractOptionHandler, type_: Type[_AbstractOptionHandler]) -> str:	1✔
21	"""Return a CLI string for creating MOA objects.
22
23	>>> from moa.classifiers.trees import HoeffdingTree
24	>>> from moa.classifiers import AbstractClassifier
25	>>> cli_str(HoeffdingTree(), AbstractClassifier)
26	'(trees.HoeffdingTree)'
27
28	:param object: MOA object to generate CLI string for.
29	:param type_: Type of the MOA object, used to strip the fully qualified name to be
30	more concise.
31	:return: CLI string for creating the MOA object.
32	"""
33	return f"({str(object.getCLICreationString(type_)).strip()})"	1✔
34
35
36	def cli_str_classifier(classifier: _AbstractClassifier \| MOAClassifier) -> str:	1✔
37	"""Return a CLI string for creating a MOA classifier.
38
39	>>> from moa.classifiers.trees import HoeffdingTree
40	>>> cli_str_classifier(HoeffdingTree())
41	'(trees.HoeffdingTree)'
42
43	"""
44	if isinstance(classifier, MOAClassifier):	1✔
45	return cli_str(classifier.moa_learner, _AbstractClassifier)	1✔
46	elif isinstance(classifier, _AbstractClassifier):	1✔
47	return cli_str(classifier, _AbstractClassifier)	1✔
48	else:
NEW 49	raise ValueError("Unknown Type")	×
50
51
52	def cli_str_regressor(regressor: _AbstractRegressor \| MOARegressor) -> str:	1✔
53	"""Return a CLI string for creating a MOA regressor.
54
55	>>> from moa.classifiers.trees import HoeffdingTree
56	>>> cli_str_regressor(HoeffdingTree())
57	'(trees.HoeffdingTree)'
58
59	"""
60	if isinstance(regressor, MOARegressor):	1✔
61	# TODO: type ignore should be removed if regressor.moa_learner is properly typed
62	return cli_str(regressor.moa_learner, _AbstractRegressor) # type: ignore	1✔
63	elif isinstance(regressor, _AbstractRegressor):	1✔
NEW 64	return cli_str(regressor, _AbstractRegressor) # type: ignore	×
65	elif isinstance(regressor, (_AbstractClassifier, MOAClassifier)):	1✔
66	# Some regressor are just classifiers.
67	return cli_str_classifier(regressor)	1✔
68	else:
NEW 69	raise ValueError("Unknown Type")	×
70
71
72	def cli_str_drift_detector(	1✔
73	detector: _AbstractChangeDetector \| MOADriftDetector,
74	) -> str:
75	"""Return a CLI string for creating a MOA drift detector.
76
77	>>> from moa.classifiers.core.driftdetection import ADWINChangeDetector
78	>>> cli_str_drift_detector(ADWINChangeDetector())
79	'(ADWINChangeDetector)'
80
81	"""
82	if isinstance(detector, MOADriftDetector):	1✔
83	return cli_str(detector.moa_detector, _AbstractChangeDetector)	1✔
84	elif isinstance(detector, _AbstractChangeDetector):	1✔
85	return cli_str(detector, _AbstractChangeDetector)	1✔
86	else:
NEW 87	raise ValueError("Unknown Type")	×
88
89
90	def cli_str_prediction_interval(	1✔
91	predictor: _PredictionIntervalLearner \| MOAPredictionIntervalLearner,
92	) -> str:
93	"""Return a CLI string for creating a MOA prediction interval learner.
94
95	>>> from moa.classifiers.predictioninterval import MVEPredictionInterval
96	>>> cli_str_prediction_interval(MVEPredictionInterval())
97	'(MVEPredictionInterval...
98
99	"""
100	if isinstance(predictor, MOAPredictionIntervalLearner):	1✔
NEW 101	return cli_str(predictor.moa_learner, _PredictionIntervalLearner) # type: ignore	×
102	elif isinstance(predictor, _PredictionIntervalLearner):	1✔
103	return cli_str(predictor, _PredictionIntervalLearner)	1✔
104	else:
NEW 105	raise ValueError("Unknown Type")	×
106
107
108	def cli_str_stream(stream: MOAStream \| _InstanceStream) -> str:	1✔
109	"""Return a CLI string for creating a MOA stream.
110
111	>>> from moa.streams.generators import RandomRBFGenerator
112	>>> cli_str_stream(RandomRBFGenerator())
113	'(generators.RandomRBFGenerator)'
114
115	"""
116	if isinstance(stream, MOAStream):	1✔
NEW 117	return cli_str(stream.moa_stream, _InstanceStream)	×
118	elif isinstance(stream, _InstanceStream):	1✔
119	return cli_str(stream, _InstanceStream)	1✔
120	else:
NEW 121	raise ValueError("Unknown Type")	×

adaptive-machine-learning / CapyMOA / 30801214514

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