The Interview Simulation

import numpy as np
from dataclasses import dataclass
from typing import Optional

@dataclass
class EndpointResult:
    endpoint_idx: Optional[int]
    confidence: float
    algorithm: str  # state for debugging: monitoring / rapid_drop_warning / confirmed_drop

class EndpointDetector:
    """
    Real-time endpoint detection for plasma etch.

    Physics: OES 520nm (silicon emission) rises as SiO2 clears.
    We detect the drop in the SiO2 signal when silicon is exposed.

    Causal: uses only past and current data (backward difference, causal smoothing).
    O(1) per update: buffers grow linearly but processing is constant-time.
    """
    def __init__(self,
                 window_smooth: int = 10,          # 100ms at 100Hz
                 threshold_relative: float = 0.15, # 15% drop = endpoint
                 min_step_duration: int = 500,      # ignore first 5 seconds
                 confirmation_count: int = 3):      # require 3 consecutive confirms
        self.window_smooth = window_smooth
        self.threshold_relative = threshold_relative
        self.min_step_duration = min_step_duration
        self.confirmation_count = confirmation_count
        self.buffer: list = []
        self.endpoint_confirmed = 0
        self.found_endpoint: Optional[int] = None

    def _causal_smooth(self, data: np.ndarray) -> np.ndarray:
        """Causal moving average. Pads with first value to maintain length."""
        if len(data) < self.window_smooth:
            return data
        kernel = np.ones(self.window_smooth) / self.window_smooth
        smoothed = np.convolve(data, kernel, mode='valid')
        pad = np.full(self.window_smooth - 1, data[0])
        return np.concatenate([pad, smoothed])

    def update(self, timestamp: float, oes_intensity: float) -> EndpointResult:
        """Process one new OES sample. Call at 100Hz."""
        self.buffer.append(oes_intensity)
        idx = len(self.buffer) - 1

        if idx < self.min_step_duration:
            return EndpointResult(None, 0.0, "insufficient_data")
        if self.found_endpoint is not None:
            return EndpointResult(self.found_endpoint, 1.0, "already_confirmed")

        smoothed = self._causal_smooth(np.array(self.buffer))
        current = smoothed[-1]
        derivative = smoothed[-1] - smoothed[-2]  # backward difference
        baseline = smoothed[:self.min_step_duration].max()

        relative_drop = (baseline - current) / baseline if baseline > 0 else 0.0

        if relative_drop > self.threshold_relative and derivative < 0:
            self.endpoint_confirmed += 1
            if self.endpoint_confirmed >= self.confirmation_count:
                self.found_endpoint = idx
                return EndpointResult(idx,
                                      min(relative_drop / self.threshold_relative, 1.0),
                                      "confirmed_drop")
        else:
            self.endpoint_confirmed = max(0, self.endpoint_confirmed - 1)

        if derivative < -0.05 * baseline:
            return EndpointResult(None, 0.5, "rapid_drop_warning")
        return EndpointResult(None, 0.0, "monitoring")

# Validation with synthetic data
def generate_oes(n=1000, endpoint_at=600, noise=0.02):
    sig = np.ones(n) + np.random.normal(0, noise, n)
    if endpoint_at < n:
        drop = np.exp(-np.linspace(0, 5, n - endpoint_at))
        sig[endpoint_at:] = 0.3 + 0.7 * drop[0] * (1 - drop)
    return sig

if __name__ == "__main__":
    oes = generate_oes(endpoint_at=600)
    detector = EndpointDetector()
    for i, intensity in enumerate(oes):
        result = detector.update(i / 100.0, intensity)
        if result.endpoint_idx is not None:
            error_ms = (result.endpoint_idx - 600) * 10  # 10ms per sample at 100Hz
            print(f"Detected at sample {result.endpoint_idx} | error: {error_ms:.0f}ms")
            break