csv_opcua_server.py

import asyncio
import threading
import socket
from http.server import HTTPServer, BaseHTTPRequestHandler

import pandas as pd
import requests
from asyncua import Server, ua

# =========================
# CONFIGURATION
# =========================
BIND_IP = "::"            # IPv6 wildcard
OPCUA_PORT = 4840
HTTP_PORT = 7860
CSV_FILE = "data.csv"
ROW_INTERVAL = 2  # seconds

# =========================
# PUBLIC IP DETECTION (IPv6)
# =========================
def get_public_ip():
    try:
        # This returns IPv6 if available
        return requests.get("https://api64.ipify.org", timeout=5).text.strip()
    except Exception:
        return "::1"

PUBLIC_IP = get_public_ip()

# Wrap IPv6 in brackets for URL usage
def format_host(ip):
    if ":" in ip:
        return f"[{ip}]"
    return ip

FORMATTED_BIND = format_host(BIND_IP)
FORMATTED_PUBLIC = format_host(PUBLIC_IP)

ENDPOINT = f"opc.tcp://{FORMATTED_BIND}:{OPCUA_PORT}/csv-opcua-server/"
PUBLIC_ENDPOINT = f"opc.tcp://{FORMATTED_PUBLIC}:{OPCUA_PORT}/csv-opcua-server/"

# =========================
# IPV6 HTTP SERVER
# =========================
class IPv6HTTPServer(HTTPServer):
    address_family = socket.AF_INET6

class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-type", "text/plain")
        self.end_headers()
        self.wfile.write(b"OK - Server is reachable via browser\n")

    def log_message(self, format, *args):
        return

def start_http_server():
    httpd = IPv6HTTPServer((BIND_IP, HTTP_PORT), HealthHandler)
    print(f"HTTP server running on http://{FORMATTED_PUBLIC}:{HTTP_PORT}")
    httpd.serve_forever()

# =========================
# MAIN ASYNC OPC UA SERVER
# =========================
async def main():
    df = pd.read_csv(CSV_FILE)
    if df.empty:
        raise ValueError("CSV file is empty")

    server = Server()
    await server.init()

    server.set_endpoint(ENDPOINT)
    server.set_server_name("IPv6 CSV OPC UA Async Server")

    uri = "http://example.org/public-csv-opcua"
    idx = await server.register_namespace(uri)

    objects = server.nodes.objects
    csv_obj = await objects.add_object(idx, "CSV_Data")

    variables = {}
    for col in df.columns:
        var = await csv_obj.add_variable(
            idx,
            col,
            float(df[col].iloc[0]),
            ua.VariantType.Double,
        )
        await var.set_writable()
        variables[col] = var

    print("====================================")
    print(" OPC UA SERVER RUNNING (IPv6)")
    print(f" Internal bind : {ENDPOINT}")
    print(f" Public access : {PUBLIC_ENDPOINT}")
    print(f" Publishing 1 row every {ROW_INTERVAL} seconds")
    print("====================================")

    row_index = 0
    row_count = len(df)

    async with server:
        while True:
            row = df.iloc[row_index]
            print(f"Publishing row {row_index + 1}/{row_count}")

            for col, var in variables.items():
                value = float(row[col])
                await var.write_value(value)
                print(f"  {col} = {value}")

            row_index = (row_index + 1) % row_count
            await asyncio.sleep(ROW_INTERVAL)

# =========================
# START EVERYTHING
# =========================
if __name__ == "__main__":
    threading.Thread(target=start_http_server, daemon=True).start()
    asyncio.run(main())