API REST Ignition : Créez des services web pour vos données SCADA

# Verify the WebDev module is installed and running
# Navigate to: https://your-gateway:8088/system/webdev/
#
# The module adds a new "WebDev" section in the Ignition Designer
# where you can create and manage your REST API resources.
#
# Default base URL for your endpoints:
# https://your-gateway:8088/system/webdev/your-project/

# WebDev Resource: doGet handler
# Endpoint: GET /system/webdev/my-project/api/tags
#
def doGet(request, session):
    """Read tag values and return as JSON."""
    import json

    # Read multiple tags at once
    tag_paths = [
        "[default]Plant/Temperature",
        "[default]Plant/Pressure",
        "[default]Plant/FlowRate",
        "[default]Plant/Status"
    ]

    values = system.tag.readBlocking(tag_paths)

    # Build response object
    response_data = {}
    for path, qv in zip(tag_paths, values):
        tag_name = path.split("/")[-1]
        response_data[tag_name] = {
            "value": qv.value,
            "quality": str(qv.quality),
            "timestamp": str(qv.timestamp)
        }

    # Return JSON response
    return {
        "json": response_data,
        "contentType": "application/json"
    }

# WebDev Resource: doPost handler
# Endpoint: POST /system/webdev/my-project/api/write
#
def doPost(request, session):
    """Write tag values from an incoming JSON payload."""
    import json

    # Parse the JSON body
    body = request["data"]
    payload = json.loads(body)

    # Validate required fields
    if "writes" not in payload:
        return {
            "json": {"error": "Missing 'writes' field"},
            "code": 400
        }

    tag_paths = []
    tag_values = []

    for write in payload["writes"]:
        tag_paths.append(write["path"])
        tag_values.append(write["value"])

    # Execute tag writes
    results = system.tag.writeBlocking(tag_paths, tag_values)

    # Build response with write results
    write_results = []
    for path, result in zip(tag_paths, results):
        write_results.append({
            "path": path,
            "success": result.isGood()
        })

    return {
        "json": {
            "status": "completed",
            "results": write_results
        },
        "contentType": "application/json"
    }

# Shared script: project.auth.validateRequest
#
def validateRequest(request):
    """Validate API key from request headers."""
    auth_header = request.get("headers", {}).get("Authorization", "")

    if not auth_header.startswith("Bearer "):
        return False, {
            "json": {"error": "Missing or invalid Authorization header"},
            "code": 401
        }

    token = auth_header[7:]  # Remove "Bearer " prefix
    valid_tokens = system.tag.readBlocking(
        ["[default]Config/API/ValidTokens"]
    )[0].value

    if token not in valid_tokens.split(","):
        return False, {
            "json": {"error": "Invalid API token"},
            "code": 403
        }

    return True, None

# Updated doGet with auth and CORS
def doGet(request, session):
    """Secured GET endpoint with CORS support."""
    import json

    # CORS headers for cross-origin requests
    cors_headers = {
        "Access-Control-Allow-Origin": "https://your-app.com",
        "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
        "Access-Control-Allow-Headers": "Authorization, Content-Type"
    }

    # Handle preflight OPTIONS request
    if request.get("method") == "OPTIONS":
        return {"code": 204, "headers": cors_headers}

    # Validate authentication
    is_valid, error_response = project.auth.validateRequest(request)
    if not is_valid:
        error_response["headers"] = cors_headers
        return error_response

    # Proceed with tag reading
    tag_paths = ["[default]Plant/Temperature"]
    values = system.tag.readBlocking(tag_paths)

    return {
        "json": {"Temperature": values[0].value},
        "headers": cors_headers,
        "contentType": "application/json"
    }