異常検知のテストに最適。複雑な合成波形をサクッと生成するSignalComposer

composer = SignalComposer(start_time=datetime.now())

# 任意のパターン (10秒周期 = 10000ms, 1時間 = 3600000ms)
pattern = composer.generate_pattern(
    cycle_time=10000, 
    pattern="001234567890ABCDEEEEEEEEEAAAAEEEEEEEEEEDCBA098765432100",
    gen_time=3600000,
    bias=30
)

# モーターの回転（サイン波） (0.1秒周期 = 100ms)
vibration = composer.generate_curve(
    cycle_time=100, 
    func="sin", 
    freq=0.01, 
    amplitude=5.0,
    gen_time=3600000, 
    bias=50
)

# 徐々に上がる温度（トレンド）
temp_rise = composer.generate_trend(
    cycle_time=100, 
    start=20.0, 
    end=80.0, 
    gen_time=3600000
)

# 3秒おきに発生するガタつき（スパイク） 
# intervalはms指定なので、3秒なら3000。指定されていた(600000, 120000)はそのままmsとして維持。
impact = composer.generate_spike(
    cycle_time=100,
    level=(3, 15), 
    interval=(120000, 600000), # 最小, 最大でランダム
    gen_time=3600000
)

# 電気ノイズ (10分後 = 600000ms から発生)
noise = composer.generate_noise(
    cycle_time=10,
    level=(0, 3), 
    gen_time=3600000,
    delay=600000
)

import matplotlib.pyplot as plt

# 周波数から周期（ms）を計算
freq = 0.002  # Hz
period = 1000 / freq  # = 500000ms

composer = SignalComposer(start_time=datetime.now(), cycle_time=100, gen_time=3600000)

# --- モーターAの回転（サイン波） ---
vibration_a = composer.generate_curve(
    cycle_time=100, 
    func="sin", 
    freq=0.001, 
    amplitude=3.0,
    gen_time=3600000
)
# センサーの限界値（2.5）で値をクリッピング
vibration_a_limited = composer.limit(vibration_a, upper=2.5, lower=-2.5)
composer.add("motor_a_limited", vibration_a_limited)

# --- モーターBの回転（コサイン波） ---
vibration_b = composer.generate_curve(
    cycle_time=100, 
    func="cos", 
    freq=freq,
    amplitude=1.0, 
    gen_time=3600000
)
# ピークに合わせて増幅（指定周期ごとに30秒間だけ1.5倍に）
vibration_b_distorted = composer.multiply(
    vibration_b,
    factor=1.5,
    interval=period,    # 計算した500000ms
    duration=30000,     # ピーク前後30秒(30000ms)
    delay=0
)
composer.add("motor_b_distorted", vibration_b_distorted)

# --- グラフ描画 ---
df = composer.create_dataframe()

def plot_signals(df):
    columns = [col for col in df.columns if col != "timestamp"]
    fig, axes = plt.subplots(len(columns), 1, figsize=(12, 4 * len(columns)), sharex=True)
    
    if len(columns) == 1: axes = [axes]
        
    for ax, col in zip(axes, columns):
        ax.plot(df["timestamp"], df[col], label=col)
        ax.set_ylabel("Amplitude")
        ax.grid(True, alpha=0.3)
        ax.legend()
    
    plt.tight_layout()
    plt.show()

plot_signals(df)

# 初期化 (1時間 = 3600000ms)
composer = SignalComposer(start_time=datetime.now(), cycle_time=100, gen_time=3600000)

# --- 1. トレンドで上昇し続ける波形 ---
# ベースのノイズを含んだ上昇トレンド
pattern_base = composer.generate_pattern(cycle_time=10000,pattern="001234567890ABCDEEEEEEEEEAAAAEEEEEEEEEEDCBA098765432100")
trend_base = composer.generate_trend(start=20.0, end=80.0)
noise_base = composer.generate_noise(level=(0, 2), interval=500)
# 合成して「上昇トレンド波形」として登録
rising_wave = composer.merge([trend_base, noise_base, pattern_base])
composer.add("上昇トレンド", rising_wave)


# --- 2. 一定期間ごとに収束するサイン波 ---
# まずはベースとなる連続したサイン波を作成
vibration_base = composer.generate_curve(func="sin", freq=0.05, amplitude=10.0)
# 5分(300000ms)ごとに発生し、最初の1分(60000ms)で急激に減衰する処理を繰り返す
# ※multiplyやdecayのロジックを応用
converging_vibration = composer.decay(
    vibration_base, 
    rate=0.5,       # 減衰率
    interval=10000, # 10秒ごとに半減
    delay=0,
    max_repeats=4, # 4回目でリピートを終了
    fill_after=False # リピート終了後は値を生成しない
)

composer.add("収束サイン波", converging_vibration)

# --- DataFrame作成 ---
df = composer.create_dataframe()
print(df.head())

# 初期化 (100秒 = 100,000ms / 周期 10ms)
gen_time = 100000
cycle = 10 
composer = SignalComposer(start_time=datetime.now(), cycle_time=cycle, gen_time=gen_time)

# 1. ベース波形（きれいなサイン波）
p_base = composer.generate_curve(
    func="sin", 
    gen_time=gen_time, 
    cycle_time=cycle, 
    freq=5.0, 
    amplitude=5.0, 
    bias=12.0
)

# 2. 50秒(50,000ms)後から「1秒(1000ms)ごとに0.2秒(200ms)間だけ」振幅を2.5倍にする
p_gated = composer.multiply(
    p_base, 
    factor=2.5, 
    interval=1000, 
    duration=200, 
    delay=50000
)

# 3. 物理限界を設定（上限18.0でカットして波形を潰す）
p_final = composer.limit(p_gated, upper=18.0)

# --- 信号の登録とデータ出力 ---

# 信号に名前をつけて登録
composer.add("sensor_value", p_final)

# DataFrameの作成
df_output = composer.create_dataframe()

# CSVとして保存
df_output.to_csv("simulation_data.csv", index=False, encoding='utf-8')

print("CSV出力完了: simulation_data.csv")
print(df_output.head())

# 10分間 (600,000ms) / 周期 100ms
gen_time = 600000
composer = SignalComposer(start_time=datetime.now(), cycle_time=100, gen_time=gen_time)

# --- 1. 車体振動 (故障の起点) ---
# 300秒(300,000ms)後に歯車破損。1秒周期のガタつきが10回続き、その後も微細な振動が残る設定
vibration_raw = composer.generate_curve("sin", freq=20.0, amplitude=15.0)
broken_gear_vib = composer.decay(
    vibration_raw, 
    rate=0.1, 
    interval=100, 
    cycle=1000, 
    delay=300000, 
    max_repeats=10, 
    fill_after=2.0  # 激しいガタつきの後は2.0の継続振動が残る
)
composer.add("車体振動", broken_gear_vib)

# --- 2. 車速 (故障により低下) ---
speed_base = composer.generate_trend(start=0.0, end=100.0)
# 300秒後に振動が発生した影響で、速度が0.8倍のペースで減衰（低下）し始める
speed = composer.decay(speed_base, rate=0.8, interval=100000, delay=300000)
composer.add("車速(km/h)", speed)

# --- 3. エンジン回転数 (故障により不安定化) ---
rpm_trend = speed * 30 + 800
rpm_noise = composer.generate_noise(level=(-30, 30), interval=200)
# 300秒後からノイズ（ブレ）を5倍に増幅
rpm_instability = composer.multiply(rpm_noise, factor=5.0, interval=1000, duration=1000, delay=300000)
engine_rpm = composer.merge([rpm_trend, rpm_instability])
composer.add("エンジン回転数", engine_rpm)

# --- 4. エンジン温度 (故障により急上昇) ---
temp_base = composer.generate_trend(start=70.0, end=90.0) # 通常上昇
# 300秒後から摩擦熱でさらに20度上昇するトレンドを合成
temp_anomaly = composer.generate_trend(start=0.0, end=20.0, delay=300000)
engine_temp = composer.merge([temp_base, temp_anomaly])
composer.add("エンジン温度(℃)", engine_temp)

# --- 出力と描画 ---
df = composer.create_dataframe()
df.to_csv("vehicle_fault_data.csv", index=False)

composer = SignalComposer(start_time, cycle_time=100, gen_time=10000)

composer.add("Sensor_A", series_a)
df = composer.create_dataframe()

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from functools import reduce
import math
import random

class SignalComposer:
    """
    信号生成および合成を行うコンポーザー。
    
    すべての時間に関する引数は「ミリ秒 (ms)」で統一されています。
    サンプリング周期が異なる信号同士でも、最小周期に合わせて自動で補間・合成を行います。
    """

    def __init__(self, start_time: datetime, cycle_time: int = 100, gen_time: int = 10000):
        """
        SignalComposerを初期化します。

        Args:
            start_time (datetime): 生成する全信号の基準となる開始時刻。
            cycle_time (int): デフォルトのサンプリング周期 (ms)。省略時の基本値。
            gen_time (int): デフォルトの信号生成期間 (ms)。省略時の基本値。
        """
        self.start_time = start_time
        self.default_cycle_time = cycle_time
        self.default_gen_time = gen_time
        self.signals = {}
        self.cycle_base = None

    def _update_cycle_base(self, cycle_time: int):
        """
        マージ処理の基準となる最小サンプリング周期を更新します。

        Args:
            cycle_time (int): 使用されたサンプリング周期 (ms)。
        """
        if self.cycle_base is None:
            self.cycle_base = cycle_time
        else:
            self.cycle_base = min(self.cycle_base, cycle_time)

    def _apply_delay(self, series: pd.Series, delay: int, cycle_time: int) -> pd.Series:
        """
        信号に対して時間遅延を適用し、空いた先頭を0で埋めます。

        Args:
            series (pd.Series): 遅延を適用する信号データ。
            delay (int): 遅延させる時間 (ms)。
            cycle_time (int): その信号のサンプリング周期 (ms)。

        Returns:
            pd.Series: 遅延適用後の信号。
        """
        if delay <= 0:
            return series
        shift_steps = int(delay // cycle_time)
        return series.shift(shift_steps).fillna(0.0)

    def generate_pattern(self, pattern: str, cycle_time: int = None, gen_time: int = None, bias: float = 0.0, scale: float = 1.0, delay: int = 0) -> pd.Series:
        """
        16進数パターンの文字列に基づいた繰り返し波形を生成します。

        Args:
            pattern (str): "01AF"等の16進数。各文字がその時点の値（0-15）を表します。
            cycle_time (int, optional): サンプリング周期 (ms)。
            gen_time (int, optional): 信号の長さ (ms)。
            bias (float): 全体へ加算するオフセット値。
            scale (float): 各値に乗算する倍率。
            delay (int): 生成開始の遅延時間 (ms)。

        Returns:
            pd.Series: 生成されたパターン信号。
        """
        c_time = cycle_time or self.default_cycle_time
        g_time = gen_time or self.default_gen_time
        data_count = int(g_time // c_time)
        hex_map = {c: int(c, 16) for c in "0123456789ABCDEF"}
        pattern_values = [hex_map[c] * scale + bias for c in pattern.upper()]
        repeat_count = math.ceil(data_count / len(pattern_values))
        full_pattern = (pattern_values * repeat_count)[:data_count]
        timestamps = [self.start_time + timedelta(milliseconds=i * c_time) for i in range(data_count)]
        series = pd.Series(full_pattern, index=pd.to_datetime(timestamps))
        self._update_cycle_base(c_time)
        return self._apply_delay(series, delay, c_time)

    def generate_trend(self, cycle_time: int = None, gen_time: int = None, start: float = 0.0, end: float = 1.0, delay: int = 0) -> pd.Series:
        """
        開始値から終了値まで一定速度で変化する直線（トレンド）を生成します。

        Args:
            cycle_time (int, optional): サンプリング周期 (ms)。
            gen_time (int, optional): 信号の長さ (ms)。
            start (float): 開始時の値。
            end (float): 終了時の値。
            delay (int): 生成開始の遅延時間 (ms)。

        Returns:
            pd.Series: 生成されたトレンド信号。
        """
        c_time = cycle_time or self.default_cycle_time
        g_time = gen_time or self.default_gen_time
        data_count = int(g_time // c_time)
        values = np.linspace(start, end, data_count)
        timestamps = [self.start_time + timedelta(milliseconds=i * c_time) for i in range(data_count)]
        series = pd.Series(values, index=pd.to_datetime(timestamps))
        self._update_cycle_base(c_time)
        return self._apply_delay(series, delay, c_time)

    def generate_noise(self, cycle_time: int = None, gen_time: int = None, level=1.0, interval=1000, delay: int = 0) -> pd.Series:
        """
        ランダムに値が変化するステップ状のノイズを生成します。

        Args:
            cycle_time (int, optional): サンプリング周期 (ms)。
            gen_time (int, optional): 信号の長さ (ms)。
            level (float or tuple): 強度。単数値なら0~N、タプルなら(min, max)。
            interval (int or tuple): 値を保持する期間 (ms)。タプルなら(min, max)でランダム。
            delay (int): 生成開始の遅延時間 (ms)。

        Returns:
            pd.Series: 生成されたノイズ信号。
        """
        c_time = cycle_time or self.default_cycle_time
        g_time = gen_time or self.default_gen_time
        data_count = int(g_time // c_time)
        low, high = (0.0, float(level)) if isinstance(level, (int, float)) else (float(level[0]), float(level[1]))
        values = []
        while len(values) < data_count:
            v = random.uniform(low, high)
            step_ms = interval if isinstance(interval, (int, float)) else random.uniform(*interval)
            step = max(1, int(step_ms // c_time))
            values.extend([v] * step)
        timestamps = [self.start_time + timedelta(milliseconds=i * c_time) for i in range(data_count)]
        series = pd.Series(values[:data_count], index=pd.to_datetime(timestamps))
        self._update_cycle_base(c_time)
        return self._apply_delay(series, delay, c_time)

    def generate_spike(self, cycle_time: int = None, gen_time: int = None, level=(1.0, 5.0), interval=(1000, 5000), base_level: float = 0.0, delay: int = 0) -> pd.Series:
        """
        一定間隔（またはランダム間隔）で発生する突発的なスパイク信号を生成します。

        Args:
            cycle_time (int, optional): サンプリング周期 (ms)。
            gen_time (int, optional): 信号の長さ (ms)。
            level (tuple): スパイク時の値の範囲 (min, max)。
            interval (tuple): 発生間隔の範囲 (min, max) (ms)。
            base_level (float): スパイクが起きていない時のベース値。
            delay (int): 生成開始の遅延時間 (ms)。

        Returns:
            pd.Series: 生成されたスパイク信号。
        """
        c_time = cycle_time or self.default_cycle_time
        g_time = gen_time or self.default_gen_time
        data_count = int(g_time // c_time)
        values = np.full(data_count, float(base_level))
        i = 0
        while i < data_count:
            values[i] = random.uniform(*level)
            step_ms = interval if isinstance(interval, (int, float)) else random.uniform(*interval)
            i += max(1, int(step_ms // c_time))
        timestamps = [self.start_time + timedelta(milliseconds=i * c_time) for i in range(data_count)]
        series = pd.Series(values, index=pd.to_datetime(timestamps))
        self._update_cycle_base(c_time)
        return self._apply_delay(series, delay, c_time)

    def generate_curve(self, func: str, cycle_time: int = None, freq: float = 1.0, gen_time: int = None, phase: float = 0.0, amplitude: float = 1.0, bias: float = 0.0, delay: int = 0) -> pd.Series:
        """
        サイン波、コサイン波、矩形波、鋸歯状波などの周期関数を生成します。

        Args:
            func (str): 種類 ("sin", "cos", "square", "sawtooth")。
            cycle_time (int, optional): サンプリング周期 (ms)。
            freq (float): 周波数 (Hz)。1.0で1秒間に1周期。
            gen_time (int, optional): 信号の長さ (ms)。
            phase (float): 初期位相 (ラジアン)。
            amplitude (float): 振幅（中心から山までの高さ）。
            bias (float): 全体へのオフセット加算値。
            delay (int): 生成開始の遅延時間 (ms)。

        Returns:
            pd.Series: 生成された周期波形。
        """
        c_time = cycle_time or self.default_cycle_time
        g_time = gen_time or self.default_gen_time
        data_count = int(g_time // c_time)
        t = np.arange(data_count) * (c_time / 1000.0)
        x = 2 * np.pi * freq * t + phase
        f = func.lower()
        if f == "sin": values = amplitude * np.sin(x)
        elif f == "cos": values = amplitude * np.cos(x)
        elif f == "square": values = amplitude * np.sign(np.sin(x))
        elif f == "sawtooth": values = amplitude * (2 * (x / (2 * np.pi) % 1) - 1)
        else: raise ValueError(f"未対応関数: {func}")
        values += bias 
        timestamps = [self.start_time + timedelta(milliseconds=i * c_time) for i in range(data_count)]
        series = pd.Series(values, index=pd.to_datetime(timestamps))
        self._update_cycle_base(c_time)
        return self._apply_delay(series, delay, c_time)

    def limit(self, series: pd.Series, upper: float = None, lower: float = None) -> pd.Series:
        """
        信号の値を指定した上下限でクリッピング（カット）します。

        Args:
            series (pd.Series): 対象の信号。
            upper (float, optional): 上限値。これを超える値は上限値に固定されます。
            lower (float, optional): 下限値。これより小さい値は下限値に固定されます。

        Returns:
            pd.Series: クリッピング後の信号。
        """
        return series.clip(lower=lower, upper=upper)

    def multiply(self, series: pd.Series, factor: float, interval: int, duration: int, delay: int = 0, max_repeats: int = None, fill_after: float = 1.0) -> pd.Series:
            """
            Args:
                max_repeats (int): 指定回数。
                fill_after (float): 指定回数終了後の倍率。ここを 0.0 にすれば4回目以降消えます。
            """
            t_ms = (series.index - series.index[0]).total_seconds() * 1000
            t_values = t_ms.values
            
            mask = ((t_values - delay) % interval < duration) & (t_values >= delay)
            
            if max_repeats is not None:
                limit_time = delay + (interval * max_repeats)
                # 指定回数以内の場合のみ mask を適用し、それ以降は fill_after の値を倍率にする
                multipliers = np.where(t_values < limit_time, 
                                    np.where(mask, factor, 1.0), 
                                    fill_after)
            else:
                multipliers = np.where(mask, factor, 1.0)
                
            return series * multipliers

    def decay(self, series: pd.Series, rate: float = 0.5, interval: int = 10000, cycle: int = 600000, delay: int = 0, max_repeats: int = None, fill_after: float = 1.0) -> pd.Series:
            """
            指定した周期(cycle)ごとに減衰をリセットして適用します。指定回数で終了可能です。

            Args:
                series (pd.Series): 対象の信号。
                rate (float): 減衰率。
                interval (int): 減衰のステップ時間 (ms)。
                cycle (int): 減衰をリセットする周期 (ms)。
                delay (int): 開始遅延 (ms)。
                max_repeats (int, optional): 最大繰り返し回数。
                fill_after (float): 指定回数終了後の倍率。0.0にすると以降の信号が消えます。
            """
            t_ms = (series.index - series.index[0]).total_seconds() * 1000
            t_values = t_ms.values
            
            # 周期内での経過時間を計算
            relative_t = (t_values - delay) % cycle
            relative_t = np.maximum(0, relative_t)
            
            multipliers = np.power(rate, relative_t / interval)
            
            # delay前は 1.0 (元の信号を維持)
            multipliers[t_values < delay] = 1.0
            
            # 回数制限の適用
            if max_repeats is not None:
                limit_time = delay + (cycle * max_repeats)
                # 指定回数終了後は fill_after の値を適用（0.0にすれば消える）
                multipliers[t_values >= limit_time] = fill_after
                
            return series * multipliers

    def merge(self, series_list: list[pd.Series]) -> pd.Series:
        """
        複数の信号を1つの時間軸上で重ね合わせ（加算）ます。
        サンプリング周期が異なる場合は、最小周期に合わせて補間されます。

        Args:
            series_list (list[pd.Series]): 加算するSeriesのリスト。

        Returns:
            pd.Series: 加算合成された信号。
        """
        if not series_list: return pd.Series()
        all_indices = [s.index for s in series_list if not s.empty]
        common_index = pd.date_range(start=min(idx.min() for idx in all_indices), 
                                     end=max(idx.max() for idx in all_indices), 
                                     freq=f"{self.cycle_base}ms")
        aligned = [s.reindex(common_index).interpolate().ffill().bfill() for s in series_list]
        return reduce(lambda a, b: a.add(b, fill_value=0), aligned)

    def add(self, name: str, series: pd.Series):
        """
        生成・加工した信号に名前を付けて、コンポーザーに登録します。

        Args:
            name (str): カラム名（例："Sensor_01"）。
            series (pd.Series): 登録する信号データ。
        """
        self.signals[name] = series

    def create_dataframe(self) -> pd.DataFrame:
        """
        add()で登録されたすべての信号を統合し、1つのDataFrameを生成します。

        Returns:
            pd.DataFrame: timestamp列を主軸に、各信号がカラムとなったデータ。
        """
        df = pd.DataFrame(self.signals)
        df.index.name = "timestamp"
        return df.reset_index()