| |
| from __future__ import annotations |
| from dataclasses import asdict, dataclass |
| from typing import Any, Callable, Literal |
| from wisp.cpu.algorithms import HarSculptForestHMM, HybridCPUEnsemble, InvariantIntervalShapeForest, PosteriorCPUEnsemble, RocketSketchRidgeHMM, StateRocketIntervalHMM, SymbolicIntervalForest |
| MethodGroup = Literal['base', 'composite'] |
|
|
| @dataclass(frozen=True) |
| class MethodSpec: |
| method_id: str |
| paper_name: str |
| code_class: str |
| group: MethodGroup |
| family_index: int |
| params: dict[str, Any] |
| representation_scope: str |
|
|
| def to_dict(self) -> dict[str, Any]: |
| return asdict(self) |
| _CANONICAL_SPECS: tuple[MethodSpec, ...] = (MethodSpec(method_id='wisp_ss', paper_name='WISP-SS', code_class='HarSculptForestHMM', group='base', family_index=1, params={'classifier': 'extratrees', 'use_hmm': True, 'segment_sizes': (4, 8, 16), 'symbolic_bins': 6}, representation_scope='global/multiscale statistics, relative spectrum, autocorrelation, symbolic transitions, cross-axis correlation'), MethodSpec(method_id='wisp_rc', paper_name='WISP-RC', code_class='RocketSketchRidgeHMM', group='base', family_index=2, params={'n_kernels': 384, 'classifier': 'ridge', 'use_hmm': True}, representation_scope='random dilated multichannel convolutions with PPV/max/mean/std pooling'), MethodSpec(method_id='wisp_so', paper_name='WISP-SO', code_class='SymbolicIntervalForest', group='base', family_index=3, params={'use_hmm': True}, representation_scope='per-window quantile states, transition matrices, fixed multiscale segment summaries'), MethodSpec(method_id='wisp_cse', paper_name='WISP-CSE', code_class='HybridCPUEnsemble', group='composite', family_index=4, params={'use_hmm': True}, representation_scope='score-level equal fusion of F1, F2 and F3 followed by one optional HMM'), MethodSpec(method_id='wisp_gis', paper_name='WISP-GIS', code_class='InvariantIntervalShapeForest', group='base', family_index=5, params={'classifier': 'extratrees', 'use_hmm': True, 'segment_sizes': (4, 8, 16), 'symbolic_bins': 6, 'n_shapelets': 32}, representation_scope='axis-dependent statistics plus axis-reduced geometry, interval distributions and random local-shape prototypes'), MethodSpec(method_id='wisp_cis', paper_name='WISP-CIS', code_class='StateRocketIntervalHMM', group='base', family_index=6, params={'n_kernels': 384, 'n_intervals': 24, 'classifier': 'ridge', 'use_hmm': True}, representation_scope='random convolution, interval distribution, within-window state summaries and optional across-window HMM'), MethodSpec(method_id='wisp_ese', paper_name='WISP-ESE', code_class='PosteriorCPUEnsemble', group='composite', family_index=7, params={'use_hmm': True, 'weights': (0.34, 0.33, 0.33)}, representation_scope='score-level weighted fusion of fixed F6, F5 and F3 members followed by one optional HMM')) |
| SPEC_BY_ID = {spec.method_id: spec for spec in _CANONICAL_SPECS} |
| CANONICAL_METHOD_IDS = tuple((spec.method_id for spec in _CANONICAL_SPECS)) |
| BASE_METHOD_IDS = tuple((spec.method_id for spec in _CANONICAL_SPECS if spec.group == 'base')) |
| COMPOSITE_METHOD_IDS = tuple((spec.method_id for spec in _CANONICAL_SPECS if spec.group == 'composite')) |
| _BUILDERS: dict[str, Callable[..., Any]] = {'HarSculptForestHMM': HarSculptForestHMM, 'RocketSketchRidgeHMM': RocketSketchRidgeHMM, 'SymbolicIntervalForest': SymbolicIntervalForest, 'HybridCPUEnsemble': HybridCPUEnsemble, 'InvariantIntervalShapeForest': InvariantIntervalShapeForest, 'StateRocketIntervalHMM': StateRocketIntervalHMM, 'PosteriorCPUEnsemble': PosteriorCPUEnsemble} |
|
|
| def list_specs() -> tuple[MethodSpec, ...]: |
| return _CANONICAL_SPECS |
|
|
| def get_spec(method_id: str) -> MethodSpec: |
| try: |
| return SPEC_BY_ID[method_id] |
| except KeyError as exc: |
| raise KeyError(f'Unknown method_id={method_id!r}; expected one of {sorted(SPEC_BY_ID)}') from exc |
|
|
| def _tupleize_params(params: dict[str, Any]) -> dict[str, Any]: |
| out = dict(params) |
| for key in ('segment_sizes', 'weights'): |
| if key in out and isinstance(out[key], list): |
| out[key] = tuple(out[key]) |
| return out |
|
|
| def build_method(method_id: str, *, seed: int, n_jobs: int=1, overrides: dict[str, Any] | None=None) -> Any: |
| spec = get_spec(method_id) |
| params = _tupleize_params(spec.params) |
| if overrides: |
| params.update(_tupleize_params(overrides)) |
| params['seed'] = int(seed) |
| params['n_jobs'] = int(n_jobs) |
| return _BUILDERS[spec.code_class](**params) |
|
|
| def resolved_spec(method_id: str, *, seed: int, n_jobs: int, overrides: dict[str, Any] | None=None) -> dict[str, Any]: |
| spec = get_spec(method_id) |
| params = _tupleize_params(spec.params) |
| if overrides: |
| params.update(_tupleize_params(overrides)) |
| params.update({'seed': int(seed), 'n_jobs': int(n_jobs)}) |
| payload = spec.to_dict() |
| payload['params'] = params |
| return payload |
|
|
