EsoCore Integration Guide

Comprehensive guide for integrating EsoCore with existing industrial systems including CMMS, SCADA, MES, ERP, and industrial protocols. Learn how to connect EsoCore monitoring data to your existing workflows, automate maintenance processes, and maximize the value of equipment health insights.

Integration Overview

EsoCore supports multiple integration pathways to fit into your existing industrial infrastructure:

Industrial Protocols:

Modbus RTU/TCP
PROFINET Device
PROFIBUS DP
OPC UA (planned)

IT Systems:

REST API for custom integrations
Webhook notifications for events
Database access for analytics
File exports (CSV, JSON, Parquet)

Common Integrations:

CMMS (Computerized Maintenance Management Systems)
SCADA (Supervisory Control and Data Acquisition)
MES (Manufacturing Execution Systems)
ERP (Enterprise Resource Planning)
BI Tools (Business Intelligence and Analytics)

REST API Integration

API Overview

EsoCore provides a REST API for programmatic access to platform features.

Note: The management API is under active development. Some integration patterns described below (webhooks, alert acknowledgment, config REST API) are planned but not yet implemented. Refer to the live Swagger documentation at /api/docs/ for the current endpoint list.

Base URL: https://[SERVER]/api/

Authentication: Token-based authentication via Authorization header, or API key via X-Auth-Key header (also accepts X-API-Key for compatibility):

Authorization: Token your-token-here
# or for device-facing IoT endpoints:
X-Auth-Key: your-device-api-key

Response Format: JSON

Authentication

Users authenticate via the login endpoint:

import requests

# Get auth token
response = requests.post(
    "https://server/api/auth/login/",
    json={"email": "[email protected]", "password": "password"}
)
token = response.json()["token"]

# Use token for subsequent requests
headers = {
    "Authorization": f"Token {token}",
    "Content-Type": "application/json"
}

response = requests.get(
    "https://server/api/telemetry/points/",
    headers=headers
)

Key API Endpoints

Management Endpoints (token auth):

GET /api/devices/devices/          # Device list
GET /api/devices/sites/            # Site list
GET /api/devices/firmware/         # Firmware bundles
GET /api/telemetry/packets/        # Telemetry packets
GET /api/telemetry/points/         # Telemetry data points
GET /api/telemetry/windows/        # Telemetry windows
GET /api/events/events/            # System events
GET /api/events/rules/             # Alert rules
GET /api/events/notifications/     # Notification queue
GET /api/assets/assets/            # Asset management
GET /api/assets/cycles/            # Asset cycles
GET /api/users/profiles/           # User profiles
GET /api/dashboard/summary/        # Dashboard summary

IoT Device-Facing Endpoints (API key auth via X-Auth-Key):

POST /api/iot/v1/auth/handshake    # Device authentication
POST /api/iot/v1/telemetry/batch   # Telemetry data ingestion
GET  /api/iot/v1/config            # Device configuration
GET  /api/iot/v1/ota/check         # OTA update check
POST /api/iot/v1/ota/report        # OTA update status report

Common query parameters for list endpoints:

page: Page number for pagination
page_size: Results per page
ordering: Sort field
search: Text search filter

Example Integrations

Python Integration:

import requests
from datetime import datetime, timedelta

class EsoCoreClient:
    def __init__(self, base_url, token):
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Token {token}",
            "Content-Type": "application/json"
        }

    def get_devices(self):
        """Get all devices"""
        response = requests.get(
            f"{self.base_url}/devices/devices/",
            headers=self.headers
        )
        return response.json()

    def get_telemetry_points(self, **params):
        """Get telemetry data points"""
        response = requests.get(
            f"{self.base_url}/telemetry/points/",
            headers=self.headers,
            params=params
        )
        return response.json()

    def get_events(self, **params):
        """Get system events"""
        response = requests.get(
            f"{self.base_url}/events/events/",
            headers=self.headers,
            params=params
        )
        return response.json()

# Usage
client = EsoCoreClient(
    base_url="https://your-server/api",
    token="your-auth-token"
)

# Get devices
devices = client.get_devices()
for device in devices["results"]:
    print(f"{device['name']}: {device['status']}")

# Get recent events
events = client.get_events()
if events["results"]:
    print(f"INFO: {len(events['results'])} recent events")

JavaScript/Node.js Integration:

const axios = require('axios')

class EsoCoreClient {
  constructor(baseUrl, token) {
    this.baseUrl = baseUrl
    this.headers = {
      Authorization: `Token ${token}`,
      'Content-Type': 'application/json',
    }
  }

  async getDevices() {
    const response = await axios.get(`${this.baseUrl}/devices/devices/`, {
      headers: this.headers,
    })
    return response.data
  }

  async getTelemetryPoints(params = {}) {
    const response = await axios.get(`${this.baseUrl}/telemetry/points/`, {
      headers: this.headers,
      params,
    })
    return response.data
  }
}

// Usage
const client = new EsoCoreClient('https://your-server/api', 'your-auth-token')

client
  .getDevices()
  .then(data => console.log(data.results))
  .catch(error => console.error(error))

CMMS Integration

Overview

Integrate EsoCore with Computerized Maintenance Management Systems to automatically create work orders based on equipment condition.

Supported CMMS Platforms:

SAP PM
IBM Maximo
Infor EAM
eMaint
Fiix
Generic via REST API

Integration Methods

Method 1: Webhook Notifications

Configure EsoCore to send alerts directly to CMMS:

In EsoCore:
- Settings → Integrations → Webhooks
- Add new webhook
- Webhook URL: https://cmms.company.com/api/work-orders
- Authentication: Bearer token or API key
- Event types: Select alert severities to trigger
- Payload template: Map EsoCore data to CMMS fields
Payload Example:

{
  "equipment_id": "{{device_id}}",
  "alert_type": "{{alert_severity}}",
  "description": "{{alert_message}}",
  "priority": "{{priority_mapping}}",
  "detected_at": "{{timestamp}}",
  "sensor_data": {
    "sensor": "{{sensor_name}}",
    "value": "{{current_value}}",
    "threshold": "{{threshold_value}}"
  }
}

Test integration with sample alert

Method 2: Scheduled Polling

CMMS polls EsoCore API periodically:

# CMMS-side script (runs every 5 minutes)
import requests

ESOCORE_API = "https://your-server/api"
ESOCORE_TOKEN = "your-auth-token"
CMMS_API = "https://cmms/api/work-orders"
CMMS_KEY = "cmms-api-key"

# Get recent events from EsoCore
events = requests.get(
    f"{ESOCORE_API}/events/events/",
    headers={"Authorization": f"Token {ESOCORE_TOKEN}"},
).json()

# Create work orders in CMMS for new events
for event in events["results"]:
    work_order = {
        "equipment_id": event["device"],
        "title": f"Alert: {event['event_type']}",
        "description": event["message"],
        "priority": map_priority(event["severity"]),
        "due_date": calculate_due_date(event["severity"])
    }

    # Create work order
    response = requests.post(
        CMMS_API,
        headers={"Authorization": f"Bearer {CMMS_KEY}"},
        json=work_order
    )

Method 3: Email-to-Ticket

If CMMS supports email-to-ticket:

In EsoCore:
- Settings → Notifications → Email
- Add CMMS email address: [email protected]
- Configure email template with required CMMS fields in subject/body
Email Template:

Subject: [EQUIPMENT:{{device_name}}][PRIORITY:{{severity}}] {{alert_type}}

Equipment: {{device_name}}
Location: {{location}}
Alert: {{alert_type}}
Severity: {{severity}}
Detected: {{timestamp}}

Description:
{{alert_message}}

Sensor Data:
{{sensor_readings}}

Recommended Action:
{{recommended_action}}

SAP PM Integration Example

# SAP PM work order creation
from pyrfc import Connection

def create_sap_notification(alert):
    # Connect to SAP
    conn = Connection(
        ashost='sap.company.com',
        sysnr='00',
        client='100',
        user='INTERFACE_USER',
        passwd='password'
    )

    # Call BAPI to create notification
    result = conn.call('BAPI_ALM_NOTIF_CREATE', {
        'NOTIFHEADER': {
            'NOTIF_TYPE': 'M2',  # Maintenance notification
            'EQUIPMENT': alert['equipment_id'],
            'PRIORITY': map_to_sap_priority(alert['severity'])
        },
        'NOTIFTASK': [{
            'TASK_TEXT': alert['message'],
            'TASK_CODE': 'PRED',  # Predictive maintenance
            'TASK_CODEGRP': 'PM'
        }]
    })

    if result['RETURN']['TYPE'] == 'S':
        notification_id = result['NOTIFHEADER']['NOTIF_NO']
        return notification_id
    else:
        raise Exception(f"SAP Error: {result['RETURN']['MESSAGE']}")

SCADA Integration

Modbus TCP Integration

Expose EsoCore data to SCADA via Modbus TCP:

Configuration:

EsoCore Web UI → Settings → Fieldbus → Modbus TCP
Enable Modbus TCP Server
Port: 502 (default) or custom
Configure register mapping:
- Starting register: 40001
- Data type: Float (2 registers) or Integer (1 register)
- Map sensors to register ranges

Register Map Example:

Register	Parameter	Type	Unit
40001-40002	Motor 1 Vibration RMS	Float	mm/s
40003-40004	Motor 1 Temperature	Float	°C
40005-40006	Motor 1 Current Phase A	Float	A
40007	Motor 1 Status	Int	0=OK, 1=Warning, 2=Alert
40008	Motor 1 Alert Code	Int	See docs
...	...	...	...

SCADA Configuration:

Allen-Bradley / Rockwell:

Controller: Modbus/TCP
IP Address: 192.168.1.100
Port: 502
Unit ID: 1
Register Map: Import CSV
Scan Rate: 1000ms

Siemens WinCC:

Driver: ModbusTCP.chn
IP: 192.168.1.100
Port: 502
Register Configuration: Import from XML
Update Cycle: 1s

Ignition SCADA:

# Add device
Name: EsoCore_Edge1
Driver: Modbus TCP
Hostname: 192.168.1.100
Port: 502

# Configure tags
Motor1_Vibration: Float, 40001, 2 registers
Motor1_Temperature: Float, 40003, 2 registers
Motor1_Status: Int16, 40007, 1 register

OPC UA Integration (Planned)

Future release will include OPC UA server:

Standard OPC UA information model
TLS-encrypted communication
User authentication
Browse-able address space
Subscriptions for real-time updates

MES Integration

Manufacturing Execution System Integration

Connect equipment health data to production systems:

Use Cases:

Display equipment health on production dashboards
Prevent production scheduling on degraded equipment
Correlate quality issues with equipment condition
Track equipment availability for OEE calculation

Integration Example:

# MES Integration Script
import requests
from datetime import datetime

class MESIntegration:
    def __init__(self, esocore_url, esocore_key, mes_url, mes_key):
        self.esocore_url = esocore_url
        self.esocore_headers = {"Authorization": f"Token {esocore_key}"}
        self.mes_url = mes_url
        self.mes_headers = {"Authorization": f"Bearer {mes_key}"}

    def update_equipment_status(self):
        """Update MES with equipment health status"""
        # Get equipment health from EsoCore
        devices = requests.get(
            f"{self.esocore_url}/devices/devices/",
            headers=self.esocore_headers
        ).json()

        for device in devices:
            # Get current health status
            health = self.get_device_health(device['id'])

            # Update MES equipment status
            requests.put(
                f"{self.mes_url}/equipment/{device['equipment_id']}/status",
                headers=self.mes_headers,
                json={
                    "health_score": health['score'],
                    "status": health['status'],
                    "alerts": health['active_alerts'],
                    "last_updated": datetime.now().isoformat()
                }
            )

    def check_production_readiness(self, equipment_id):
        """Check if equipment is ready for production"""
        device = self.get_device_by_equipment_id(equipment_id)

        # Get active alerts
        events = requests.get(
            f"{self.esocore_url}/events/events/",
            headers=self.esocore_headers,
            params={"device": device['id']}
        ).json()

        # Check for critical events
        critical_alerts = [e for e in events["results"] if e.get('severity') == 'critical']

        return {
            "ready": len(critical_alerts) == 0,
            "alerts": critical_alerts,
            "recommendation": "Schedule maintenance" if critical_alerts else "OK"
        }

ERP Integration

Enterprise Resource Planning Integration

Connect equipment data to business systems:

Use Cases:

Automatic spare parts ordering based on predictions
Maintenance cost tracking and budgeting
Asset lifecycle management
Warranty and vendor management

Integration Points:

Procurement:

# Trigger spare parts order based on alert
def trigger_parts_order(alert):
    # Map alert to required spare parts
    parts = get_required_parts(alert['device_type'], alert['alert_type'])

    # Create purchase requisition in ERP
    for part in parts:
        create_purchase_req(
            part_number=part['number'],
            quantity=1,
            priority="Normal" if alert['severity'] != 'critical' else "Urgent",
            requester="Predictive Maintenance System",
            notes=f"Based on alert: {alert['message']}"
        )

Finance:

# Track actual maintenance costs vs predicted
def track_maintenance_roi():
    # Get prevented failures
    prevented = get_prevented_failures()

    # Calculate avoided costs
    avoided_costs = sum([
        failure['estimated_downtime_hours'] * facility_hourly_rate +
        failure['estimated_repair_cost']
        for failure in prevented
    ])

    # Get actual monitoring costs
    monitoring_costs = get_monitoring_system_costs()

    # Calculate ROI
    roi = (avoided_costs - monitoring_costs) / monitoring_costs * 100

    # Post to ERP financial module
    post_cost_savings(
        category="Predictive Maintenance",
        avoided_costs=avoided_costs,
        actual_costs=monitoring_costs,
        roi=roi
    )

Business Intelligence Integration

Data Export for Analytics

CSV Export:

# Export historical data for analysis
def export_to_csv(device_id, start_date, end_date):
    data = requests.get(
        f"{API_URL}/telemetry/points/",
        headers=headers,
        params={
            "device_id": device_id,
            "start_time": start_date,
            "end_time": end_date,
            "limit": 100000
        }
    ).json()

    # Convert to pandas DataFrame
    import pandas as pd
    df = pd.DataFrame(data)
    df.to_csv(f"esocore_export_{device_id}.csv", index=False)

Database Integration:

# Sync to data warehouse
import psycopg2

def sync_to_warehouse():
    # Connect to data warehouse
    conn = psycopg2.connect(
        host="warehouse.company.com",
        database="analytics",
        user="esocore_sync",
        password="password"
    )

    # Get latest data from EsoCore
    latest_data = get_latest_telemetry()

    # Insert into warehouse
    cursor = conn.cursor()
    for reading in latest_data:
        cursor.execute("""
            INSERT INTO equipment_telemetry
            (timestamp, device_id, sensor_id, metric, value, unit)
            VALUES (%s, %s, %s, %s, %s, %s)
            ON CONFLICT (timestamp, device_id, sensor_id, metric)
            DO UPDATE SET value = EXCLUDED.value
        """, (
            reading['timestamp'],
            reading['device_id'],
            reading['sensor_id'],
            reading['metric'],
            reading['value'],
            reading['unit']
        ))

    conn.commit()
    cursor.close()
    conn.close()

Power BI Integration:

# Power BI REST API integration
def publish_to_powerbi(dataset_id, table_name, data):
    import requests

    # Get Power BI access token (OAuth2)
    token = get_powerbi_token()

    # Push data to Power BI
    response = requests.post(
        f"https://api.powerbi.com/v1.0/myorg/datasets/{dataset_id}/tables/{table_name}/rows",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        },
        json={"rows": data}
    )

    return response.ok

Best Practices

Security

API Key Management:

Generate separate keys for each integration
Use read-only keys when possible
Rotate keys regularly (quarterly recommended)
Store keys securely (vault, environment variables)
Never commit keys to source control

Network Security:

Use TLS/HTTPS for all API connections
Implement firewall rules limiting access
Use VPN for remote integrations
Monitor API usage for anomalies

Performance

Optimize API Usage:

Cache data when appropriate
Use bulk endpoints vs many individual calls
Implement exponential backoff for retries
Set reasonable polling intervals (1-5 minutes typical)
Use webhooks instead of polling when possible

Data Management:

Only query data you need (use filters)
Limit historical queries to necessary time ranges
Implement pagination for large datasets
Archive old data regularly

Reliability

Error Handling:

import time
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def get_session_with_retries():
    session = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('https://', adapter)
    return session

# Usage
session = get_session_with_retries()
response = session.get(url, headers=headers)

Monitoring:

Log all integration errors
Track API response times
Monitor data freshness
Alert on integration failures
Maintain fallback procedures

Support and Resources

Documentation

Examples and Code

GitHub repository: Sample integrations and code examples
Community contributions: Share your integrations
Professional services: Custom integration development available

Support

Community forum: GitHub Discussions
Commercial support: [email protected]
Integration consulting: [email protected]

Successful integration amplifies the value of equipment monitoring by connecting insights to action. Choose the integration methods that best fit your existing infrastructure and workflows.

API Documentation | Installation Guide | Quick Start