# Paper-scoped implementation; see SOURCE_PROVENANCE.json. 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