Rotating Machinery & Industrial Assets

A.2 Rotating Machinery & Industrial Assets (Vibration / Acoustic)

Critical rotating equipment (pumps, fans, compressors, gearboxes, turbines) is instrumented heavily in modern industrial facilities. Traditional "predictive maintenance" stacks often revolve around vendor-specific vibration analyzers, ad-hoc thresholds, and periodic offline analysis. However, the semantics of what operators actually care about are higher-level:

• Which machines are exhibiting early-stage anomalies (imbalance, misalignment, looseness, bearing defects)?
• How do these anomalies correlate with load, process state, and environmental conditions?
• How should maintenance be scheduled, derates applied, or process conditions adjusted to minimize risk?

Spaxiom can serve as a machinery health cortex, fusing vibration, acoustic, electrical, and thermal measurements into structured INTENT events that agents and engineers can reason over consistently across vendors and plants.

Sensors and deployment

For a rotating asset m (e.g., pump P-101), typical sensors include:

• Vibration sensors a_m(t): accelerometers at strategic points (e.g., bearing housing, motor feet), often sampled at kHz and processed into spectral features;
• Acoustic sensors s_m(t): microphones capturing bearing noise, cavitation, and other signatures;
• Electrical sensors i_m(t): motor current and voltage measurements, providing load and power quality context;
• Speed / phase sensors ω_m(t): shaft speed and phase markers (e.g., tachometer pulses);
• Thermal sensors T_m(t): surface or winding temperature.

Spaxiom does not need to operate at raw waveform level; instead, it consumes features precomputed either at the edge or inside a preprocessor:

• RMS vibration levels in bands (overall, 1X, 2X, bearing defect bands),
• Spectral kurtosis and crest factor,
• Harmonic content or envelopes of specific fault frequencies,
• Rolling statistics over electrical and thermal load.

INTENT events and health indices

From these features, Spaxiom defines higher-level INTENT events such as:

• BearingAnomaly (spectral patterns consistent with bearing defect frequencies),
• ImbalanceDetected (elevated 1X vibration at speed-correlated frequency),
• CavitationRisk (acoustic + vibration signature in pumps),
• LoadSpike (current and torque exceeding normal envelopes),
• ThermalRunawayRisk (sustained temperature rise faster than expected under load).

For a machine m and window [t₀, t₁], suppose we have:

• v^1X_m(t): vibration amplitude at the running speed (imbalance indicator),
• v^BD_m(t): vibration amplitude in bearing defect bands,
• k_m(t): spectral kurtosis (impulsiveness),
• T_m(t): temperature,
• L_m(t): mechanical or electrical load fraction.

Normalized feature integrals. Define normalized, windowed feature integrals:

where Δt = t₁ - t₀ and σ^·_m,base are baseline (healthy) standard deviations.

Similarly define:

where T^base_m(L) is the expected temperature as a function of load (learned from historical data).

Composite health score. We can define a simple machine health "anomaly score":

with weights w_i ≥ 0 tuned per machine or class. High H_m indicates abnormal behavior over the window.

Discrete INTENT events. We can then create discrete INTENT events when:

yielding MachineHealthWarning or MachineHealthAlarm for machine m.

Spaxiom-style implementation sketch

In the DSL, a rotating machine entity might encapsulate feature histories and health computation:

from spaxiom import Condition
from spaxiom.temporal import within
from spaxiom.logic import on

class RotatingMachine:
    def __init__(self, name, feat_source):
        self.name = name
        self.feat_source = feat_source  # e.g., streaming feature vectors

        # Baseline stats & thresholds (could be learned)
        self.sigma_1x = 0.12
        self.sigma_bd = 0.08
        self.warn_threshold = 2.0
        self.alarm_threshold = 4.0

    def feature_history(self, window_s: float):
        """
        Returns a list of (dt, features) where features is a dict like:
        {
            "v_1x": ...,
            "v_bd": ...,
            "kurtosis": ...,
            "temp": ...,
            "load": ...,
        }
        """
        return self.feat_source.history(window_s=window_s)

    def expected_temp(self, load: float) -> float:
        # Simple linear model as placeholder; could be learned per machine
        return 40.0 + 25.0 * load

    def health_score(self, window_s: float) -> float:
        series = self.feature_history(window_s)
        if not series:
            return 0.0

        sum_v1x = sum_vbd = sum_k = sum_texcess = 0.0
        total_dt = 0.0

        for dt, f in series:
            v1x = f["v_1x"] / max(self.sigma_1x, 1e-6)
            vbd = f["v_bd"] / max(self.sigma_bd, 1e-6)
            k   = f["kurtosis"]
            temp = f["temp"]
            load = f["load"]

            texcess = max(0.0, temp - self.expected_temp(load))

            sum_v1x     += v1x * dt
            sum_vbd     += vbd * dt
            sum_k       += k * dt
            sum_texcess += texcess * dt
            total_dt    += dt

        if total_dt <= 0:
            return 0.0

        V1x = sum_v1x / total_dt
        Vbd = sum_vbd / total_dt
        K   = sum_k / total_dt
        Tex = sum_texcess / total_dt

        w1, w2, w3, w4 = 0.4, 0.3, 0.2, 0.1
        return w1 * V1x + w2 * Vbd + w3 * K + w4 * Tex

# Example machine instance
pump_101 = RotatingMachine("P-101", feat_source=pump_101_feat_stream)

# Conditions that trigger warnings/alarms over the last 24 hours
warn_condition = Condition(
    lambda: pump_101.health_score(window_s=24 * 3600) > pump_101.warn_threshold
)
alarm_condition = Condition(
    lambda: pump_101.health_score(window_s=24 * 3600) > pump_101.alarm_threshold
)

@on(within(3600, warn_condition))   # evaluate every hour
def pump_warning_agent():
    score = pump_101.health_score(window_s=24 * 3600)
    emit_intent_event({
        "type": "MachineHealthWarning",
        "machine": pump_101.name,
        "score": score,
        "window_h": 24,
    })
    # Optional: hand off to LLM for recommended maintenance actions.

@on(within(600, alarm_condition))   # evaluate every 10 minutes
def pump_alarm_agent():
    score = pump_101.health_score(window_s=24 * 3600)
    emit_intent_event({
        "type": "MachineHealthAlarm",
        "machine": pump_101.name,
        "score": score,
        "window_h": 24,
    })
    # Optionally, trigger interlocks, derates, or automated shutdown logic.

This sketch illustrates how Spaxiom:

• treats feature streams as first-class objects (feat_source.history),
• encapsulates machine-specific health logic in reusable classes,
• promotes a scalar health_score to be consumed by agents and workflows.