9d3541fCache a grid of estimates on an `estimates` branch via workflow dispatchJeremy Magland 1"""The parameter grid whose estimates are cached on the `estimates` branch.
3Each cell is a plain dict of CLI-equivalent parameters; `cell_id` names it and
4`kernel_from_params` rebuilds its kernel. Params are stored verbatim in every
5record on the data branch, so aggregation never needs this file — the grid can
6grow or change without invalidating what is already cached.
7"""
9import sys
10from pathlib import Path
12sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'src'))
14from timeseries_entropy import kernels # noqa: E402
16RATE = 30000.0
17TAPS = 101
19SIGMAS = [1.0, 2.0, 4.0, 8.0, 16.0, 32.0]
21FILTERS = [
22 {'filter': 'none'},
23 {'filter': 'moving-average', 'width': 8},
24 {'filter': 'lowpass', 'high': 3000.0, 'rate': RATE, 'taps': TAPS},
25 {'filter': 'bandpass', 'low': 300.0, 'high': 6000.0, 'rate': RATE,
26 'taps': TAPS},
27]
30def grid():
31 """The full list of cells: one representative per filter x every sigma."""
32 return [dict(f, sigma=s) for f in FILTERS for s in SIGMAS]
35def kernel_from_params(p):
36 """Kernel array for a params dict — mirrors the CLI's --filter handling."""
37 f = p['filter']
38 if f == 'none':
39 return kernels.identity()
40 if f == 'first-difference':
41 return kernels.first_difference()
42 if f == 'moving-average':
43 return kernels.moving_average(int(p['width']))
44 if f == 'lowpass':
45 return kernels.windowed_sinc_lowpass(p['high'] / p['rate'],
46 int(p['taps']))
47 if f == 'bandpass':
48 return kernels.windowed_sinc_bandpass(p['low'] / p['rate'],
49 p['high'] / p['rate'],
50 int(p['taps']))
51 raise ValueError(f'unknown filter {f!r}')
54def cell_id(p):
55 """Stable, filename-safe name for a cell, e.g. 'bp300-6000_s8'."""
56 f = p['filter']
57 if f == 'none':
58 tag = 'none'
59 elif f == 'first-difference':
60 tag = 'diff'
61 elif f == 'moving-average':
62 tag = f"ma{_num(p['width'])}"
63 elif f == 'lowpass':
64 tag = f"lp{_num(p['high'])}"
65 elif f == 'bandpass':
66 tag = f"bp{_num(p['low'])}-{_num(p['high'])}"
67 else:
68 raise ValueError(f'unknown filter {f!r}')
69 return f"{tag}_s{_num(p['sigma'])}"
72def _num(v):
73 """4.0 -> '4', 0.5 -> '0.5'."""
74 return str(int(v)) if float(v) == int(v) else str(v)