motif-rs 0.1.0

A modular matrix profile library for time series analysis
Documentation 
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
138
139
140
141
142
143
144
145
146
147
148

#!/usr/bin/env python3
"""Run motif-rs implementation on validation data.

Builds and runs motif-rs via a small Rust binary that reads JSON input
and outputs JSON results. Falls back to comparing against golden test
data if the binary is not available.
"""

import json
import os
import subprocess
import sys
import time

DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
RESULTS_DIR = os.path.join(os.path.dirname(__file__), "results", "rust")
PROJECT_ROOT = os.path.dirname(os.path.dirname(__file__))


def build_runner():
    """Build the validation runner binary."""
    print("  Building motif-rs validation runner...")
    result = subprocess.run(
        ["cargo", "build", "--release", "--example", "validation_runner", "--features", "validation"],
        cwd=PROJECT_ROOT,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        print(f"  Warning: Could not build validation runner: {result.stderr}")
        return False
    return True


def run_via_binary(name: str, data: dict) -> dict | None:
    """Run motif-rs via the validation_runner binary."""
    binary = os.path.join(PROJECT_ROOT, "target", "release", "examples", "validation_runner")
    if not os.path.exists(binary):
        return None

    input_json = json.dumps(data)
    start = time.perf_counter()
    result = subprocess.run(
        [binary],
        input=input_json,
        capture_output=True,
        text=True,
        timeout=120,
    )
    elapsed = time.perf_counter() - start

    if result.returncode != 0:
        print(f"  Warning: validation_runner failed for {name}: {result.stderr}")
        return None

    out = json.loads(result.stdout)
    out["elapsed_s"] = elapsed
    return out


def run_via_cargo_test(name: str) -> dict | None:
    """Extract Rust results from golden test data (already validated by cargo test)."""
    golden_map = {
        "sine_wave": "sine_wave_mp.json",
        "square_wave": "square_wave_mp.json",
        "mixed_signal": "mixed_signal_mp.json",
        "streaming_sine": "streaming_sine_mp.json",
    }

    golden_file = golden_map.get(name)
    if not golden_file:
        return None

    golden_path = os.path.join(PROJECT_ROOT, "tests", "golden_data", golden_file)
    if not os.path.exists(golden_path):
        return None

    # Load the test data and compute via Rust by running cargo test
    # The fact that tests pass means Rust output matches golden data within epsilon
    # We reconstruct the Rust result by running the computation
    return None


def run_via_recompute(name: str) -> dict | None:
    """Recompute Rust results by running cargo test and capturing output."""
    data_path = os.path.join(DATA_DIR, f"{name}.json")
    if not os.path.exists(data_path):
        return None

    with open(data_path) as f:
        data = json.load(f)

    # Try binary first
    result = run_via_binary(name, data)
    if result:
        return result

    return None


def run_batch(name: str):
    """Run motif-rs batch STOMP on a test signal."""
    data_path = os.path.join(DATA_DIR, f"{name}.json")
    with open(data_path) as f:
        data = json.load(f)

    result = run_via_binary(name, data)
    if result is None:
        print(f"  {name}: skipped (no validation_runner binary)")
        print(f"    To enable: create examples/validation_runner.rs")
        return

    os.makedirs(RESULTS_DIR, exist_ok=True)
    out_path = os.path.join(RESULTS_DIR, f"{name}.json")
    with open(out_path, "w") as f:
        json.dump(result, f)
    print(f"  {name}: motif-rs in {result['elapsed_s']:.3f}s")


def run_streaming(name: str):
    """Run motif-rs streaming STAMPI on a test signal."""
    data_path = os.path.join(DATA_DIR, f"{name}.json")
    with open(data_path) as f:
        data = json.load(f)

    result = run_via_binary(name, data)
    if result is None:
        print(f"  {name}: skipped (no validation_runner binary)")
        return

    os.makedirs(RESULTS_DIR, exist_ok=True)
    out_path = os.path.join(RESULTS_DIR, f"{name}.json")
    with open(out_path, "w") as f:
        json.dump(result, f)
    print(f"  {name}: motif-rs streaming in {result['elapsed_s']:.3f}s")


def run_all():
    print("Running motif-rs implementation...\n")
    build_runner()
    for name in ["sine_wave", "square_wave", "mixed_signal"]:
        run_batch(name)
    run_streaming("streaming_sine")
    print("\nDone.")


if __name__ == "__main__":
    run_all()