File size: 4,818 Bytes
4e21335
 
 
 
 
d3d1d01
 
4e21335
 
 
 
d3d1d01
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
f110a8c
d3d1d01
 
 
4e21335
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""Helper utilities for the app."""

from __future__ import annotations

import ast
import concurrent.futures
from typing import Any, Callable, Sequence

import numpy as np
from scipy.optimize import linear_sum_assignment

# ---------------------------------------------------------------------------
# Execution timeout
# ---------------------------------------------------------------------------

DEFAULT_TIMEOUT = 300  # seconds


class ExecutionTimeoutError(Exception):
    """Raised when a detector exceeds the allowed execution time."""


def run_with_timeout(fn: Callable, *args, timeout: int = DEFAULT_TIMEOUT, **kwargs) -> Any:
    """Run *fn* in a thread with a timeout.

    Raises ``ExecutionTimeoutError`` if the function does not return in
    *timeout* seconds.
    """
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        future = pool.submit(fn, *args, **kwargs)
        try:
            return future.result(timeout=timeout)
        except concurrent.futures.TimeoutError:
            raise ExecutionTimeoutError(
                f"Execution timed out after {timeout}s. Try a smaller dataset or simpler algorithm parameters."
            ) from None


def states_to_change_points(states: Sequence[int]) -> list[int]:
    """Derive change points from state labels."""
    diffs = np.diff(states)
    return list(np.nonzero(diffs)[0] + 1)


def change_points_to_states(change_points: Sequence[int], length: int) -> np.ndarray:
    """Convert sorted change-point list to per-sample state labels."""
    labels = np.zeros(length, dtype=int)
    boundaries = [0, *sorted(cp for cp in change_points if 0 < cp < length), length]
    for idx, (start, stop) in enumerate(zip(boundaries[:-1], boundaries[1:])):
        labels[start:stop] = idx
    return labels


def one_hot_from_change_points(change_points: Sequence[int], length: int) -> np.ndarray:
    vec = np.zeros(length, dtype=int)
    for cp in change_points:
        if 0 <= cp < length:
            vec[cp] = 1
    return vec


def align_states_via_hungarian(true_states: Sequence[int], pred_states: Sequence[int]) -> np.ndarray:
    """Relabel predictions to maximise agreement with ground truth."""
    true_states = np.asarray(true_states, dtype=int)
    pred_states = np.asarray(pred_states, dtype=int)
    n_true = true_states.max() + 1
    n_pred = pred_states.max() + 1
    size = max(n_true, n_pred)
    cost = np.zeros((size, size), dtype=int)
    for t, p in zip(true_states, pred_states):
        cost[t, p] -= 1
    row_ind, col_ind = linear_sum_assignment(cost)
    mapping = {pred: true for true, pred in zip(row_ind, col_ind)}
    return np.array([mapping.get(label, label) for label in pred_states], dtype=int)


def parse_user_value(raw: str, default: Any) -> Any:
    """Parse a string input using ``ast.literal_eval`` with fallbacks."""
    if raw.strip() == "":
        return default
    try:
        return ast.literal_eval(raw)
    except (ValueError, SyntaxError):
        return raw


def normalize_zscore(signal: np.ndarray) -> np.ndarray:
    """Apply z-normalisation along time for each dimension."""
    mean = signal.mean(axis=0, keepdims=True)
    std = signal.std(axis=0, keepdims=True)
    std = np.where(std == 0, 1.0, std)
    return (signal - mean) / std


def interpret_predictions(output: Any, length: int, expect_states: bool):
    """Normalise detector output into (pred_states, pred_cps).

    Returns
    -------
    pred_states : np.ndarray or None
    pred_cps : list[int] or None
    """
    pred_states = None
    pred_cps = None
    raw_state_series = None

    if output is None:
        return pred_states, pred_cps

    if isinstance(output, dict) and "change_points" in output:
        pred_cps = [int(x) for x in output["change_points"]]
    elif isinstance(output, (list, tuple)):
        primary = output[0] if output and isinstance(output[0], (list, tuple, np.ndarray)) else output
        arr = np.asarray(primary)
        if arr.ndim == 1 and arr.size == length:
            raw_state_series = arr.astype(int)
        else:
            pred_cps = [int(x) for x in arr]
    else:
        arr = np.asarray(output)
        if arr.ndim == 1 and arr.size == length:
            raw_state_series = arr.astype(int)
        else:
            pred_cps = [int(x) for x in np.ravel(arr)]

    if expect_states:
        pred_states = raw_state_series
        if pred_states is None and pred_cps is not None:
            pred_states = change_points_to_states(pred_cps, length)
        if pred_states is not None and pred_cps is None:
            pred_cps = states_to_change_points(pred_states)
        return pred_states, pred_cps

    if pred_cps is None and raw_state_series is not None:
        pred_cps = states_to_change_points(raw_state_series)
    return None, pred_cps