Predict Equipment Failure with a Maintenance Agent

1. 1
   Ingest Sensor Data

   The Predictive Maintenance Agent requires a stream of data from equipment sensors. We'll simulate sending a JSON payload with vibration, temperature, and acoustic readings for a specific asset (e.g., 'Pump-07B') to the agent's ingestion endpoint.
2. 2
   Check Asset Health Status

   After ingesting data, query the agent for the asset's current health. The Predictive Maintenance Agent analyzes the data against its trained models to provide a real-time anomaly score and overall status.
3. 3
   Get Failure Prediction & RUL

   If the status is not 'Normal', ask the agent for a specific diagnosis. It can predict the most likely failure mode and estimate the Remaining Useful Life (RUL) in operating hours or days.
4. 4
   Trigger Automated Maintenance

   Leverage the agent's integration capabilities to take action. This simulates a request for the Predictive Maintenance Agent to create a work order in a CMMS and recommend the necessary spare parts.

import time
import json

# This is a mock class to simulate the Predictive Maintenance Agent's API.
# In a real-world application, you would use an HTTP client to interact with a live API.
class MockPredictiveMaintenanceAgent:
    """Simulates the SparkCognition Predictive Maintenance Agent"""

    def __init__(self):
        self.asset_data = {}

    def ingest_data(self, data):
        asset_id = data.get('assetId')
        if not asset_id:
            return {"status": "error", "message": "assetId is required"}
        self.asset_data[asset_id] = data
        print(f"[Agent] Ingested data for {asset_id}")
        return {"status": "success", "assetId": asset_id}

    def get_health_status(self, asset_id):
        if asset_id not in self.asset_data:
            return {"status": "error", "message": "No data for asset"}
        print(f"[Agent] Analyzing health for {asset_id}...")
        # Simulate an anomaly based on ingested temperature
        temp = self.asset_data[asset_id]['readings']['temperature_c']
        if temp > 75:
            return {
                "assetId": asset_id,
                "status": "Warning",
                "anomalyScore": 0.82,
                "lastUpdated": int(time.time())
            }
        return {
            "assetId": asset_id,
            "status": "Normal",
            "anomalyScore": 0.15,
            "lastUpdated": int(time.time())
        }

    def get_failure_prediction(self, asset_id):
        print(f"[Agent] Generating failure prediction for {asset_id}...")
        return {
            "assetId": asset_id,
            "predictedFailureMode": "Bearing Wear",
            "confidence": 0.91,
            "rul_hours": 120 # Remaining Useful Life in hours
        }

    def create_work_order(self, request):
        asset_id = request.get('assetId')
        print(f"[Agent] Creating work order for {asset_id} in CMMS...")
        return {
            "workOrderId": "WO-983451",
            "cmmsSystem": "Simulated-Maximo",
            "status": "Created",
            "summary": f"Inspect {request.get('failureMode')} on {asset_id}",
            "recommendedParts": ["SKF-6203-2Z", "Gasket-5512"]
        }

# --- Main Execution ---
def main():
    # 1. Initialize the agent
    agent = MockPredictiveMaintenanceAgent()

    # 2. Define the asset and its sensor data
    asset_id = "Pump-07B"
    sensor_data = {
      "assetId": asset_id,
      "timestamp": int(time.time()),
      "readings": {
        "vibration_mm_s": 1.25, # slightly elevated
        "temperature_c": 75.6, # above normal threshold
        "acoustic_db": 68.1,
        "current_a": 15.2
      }
    }

    print("--- Step 1: Ingesting Sensor Data ---")
    ingest_status = agent.ingest_data(sensor_data)
    print(f"Response: {json.dumps(ingest_status)}\n")

    print("--- Step 2: Checking Asset Health Status ---")
    health_status = agent.get_health_status(asset_id)
    print(f"Response: {json.dumps(health_status, indent=2)}\n")

    if health_status.get("status") != "Normal":
        print("--- Step 3: Getting Failure Prediction ---")
        prediction = agent.get_failure_prediction(asset_id)
        print(f"Response: {json.dumps(prediction, indent=2)}\n")

        print("--- Step 4: Triggering Automated Maintenance ---")
        work_order_request = {
            "assetId": asset_id,
            "failureMode": prediction['predictedFailureMode'],
            "priority": "High"
        }
        work_order_response = agent.create_work_order(work_order_request)
        print(f"Response: {json.dumps(work_order_response, indent=2)}\n")

if __name__ == "__main__":
    main()