feat: 初始化HEFA-L2 PDI管理系统项目
添加前端Vue2项目结构，包括ElementUI集成、路由配置和API模块 实现后端FastAPI服务，包含Oracle数据库连接和PDI CRUD接口 添加OPC-UA轮询服务，支持跟踪图数据同步到Oracle 提供SQLite镜像数据库用于本地开发和快速查询 包含完整的部署脚本和文档说明
This commit is contained in:
244
backend/opc_service.py
Normal file
244
backend/opc_service.py
Normal file
@@ -0,0 +1,244 @@
|
||||
"""
|
||||
OPC-UA polling service.
|
||||
|
||||
Logic:
|
||||
1. Connect to the OPC-UA server at `opc_url`.
|
||||
2. Read the counter node (`counter_node`) every `poll_interval` seconds.
|
||||
3. When the counter value changes, read the trackmap nodes listed in
|
||||
`trackmap_nodes` (a dict: {oracle_column -> node_id}).
|
||||
4. For each row returned by the trackmap nodes build an UPDATE statement
|
||||
and apply it to PLTM.CMPT_PL_TRACKMAP.
|
||||
|
||||
`trackmap_nodes` example (stored in .env or configured via UI):
|
||||
{
|
||||
"COILID": "ns=2;s=PL.TRACKMAP.P01.COILID",
|
||||
"BEF_ES": "ns=2;s=PL.TRACKMAP.P01.BEF_ES",
|
||||
...
|
||||
}
|
||||
For multi-position setups, define one entry per position column and handle
|
||||
positional logic accordingly. The current implementation reads a flat set of
|
||||
nodes and stores them against the POSITION value also read from OPC.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class OpcService:
|
||||
def __init__(self):
|
||||
self.config_path = os.path.join(os.path.dirname(__file__), "opc_config.json")
|
||||
self.opc_url: str = os.getenv("OPC_URL", "opc.tcp://192.168.1.100:4840")
|
||||
self.counter_node: str = os.getenv(
|
||||
"OPC_COUNTER_NODE", "ns=2;s=PL.TRACKMAP.COUNTER"
|
||||
)
|
||||
self.poll_interval: int = int(os.getenv("OPC_POLL_INTERVAL", "2"))
|
||||
# Mapping: oracle_column_name -> OPC node id
|
||||
# Populated from .env or via API
|
||||
self.trackmap_nodes: Dict[str, str] = self._load_trackmap_nodes()
|
||||
# Load persisted config if present
|
||||
self._load_persisted_config()
|
||||
self.running: bool = False
|
||||
self.last_counter: Optional[Any] = None
|
||||
self.last_update: Optional[str] = None
|
||||
self.event_log: List[str] = []
|
||||
self._stop_event = asyncio.Event()
|
||||
self._task: Optional[asyncio.Task] = None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def _load_trackmap_nodes(self) -> Dict[str, str]:
|
||||
"""Load trackmap node mapping from env vars prefixed OPC_NODE_."""
|
||||
nodes = {}
|
||||
for key, val in os.environ.items():
|
||||
if key.startswith("OPC_NODE_"):
|
||||
col = key[len("OPC_NODE_"):].lower()
|
||||
nodes[col] = val
|
||||
return nodes
|
||||
|
||||
def _load_persisted_config(self):
|
||||
"""Load OPC config from local json file if exists."""
|
||||
if not os.path.exists(self.config_path):
|
||||
return
|
||||
try:
|
||||
with open(self.config_path, "r", encoding="utf-8") as f:
|
||||
cfg = json.load(f)
|
||||
self.opc_url = cfg.get("opc_url", self.opc_url)
|
||||
self.counter_node = cfg.get("counter_node", self.counter_node)
|
||||
self.poll_interval = int(cfg.get("poll_interval", self.poll_interval))
|
||||
self.trackmap_nodes = cfg.get("trackmap_nodes", self.trackmap_nodes) or {}
|
||||
self._log(f"Loaded OPC config from {self.config_path}")
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to load OPC config %s: %s", self.config_path, exc)
|
||||
|
||||
def save_config(self):
|
||||
"""Persist current OPC config to local json file."""
|
||||
cfg = {
|
||||
"opc_url": self.opc_url,
|
||||
"counter_node": self.counter_node,
|
||||
"poll_interval": self.poll_interval,
|
||||
"trackmap_nodes": self.trackmap_nodes,
|
||||
}
|
||||
os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
|
||||
with open(self.config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(cfg, f, ensure_ascii=False, indent=2)
|
||||
|
||||
def _log(self, msg: str):
|
||||
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
entry = f"[{ts}] {msg}"
|
||||
logger.info(entry)
|
||||
self.event_log.append(entry)
|
||||
if len(self.event_log) > 500:
|
||||
self.event_log = self.event_log[-500:]
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
async def start_polling(self):
|
||||
if self.running:
|
||||
return
|
||||
self._stop_event.clear()
|
||||
self._task = asyncio.create_task(self._poll_loop())
|
||||
|
||||
async def stop_polling(self):
|
||||
self._stop_event.set()
|
||||
if self._task:
|
||||
self._task.cancel()
|
||||
try:
|
||||
await self._task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self.running = False
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
async def _poll_loop(self):
|
||||
self.running = True
|
||||
self._log(f"OPC polling started: {self.opc_url}")
|
||||
try:
|
||||
from opcua import Client # type: ignore
|
||||
except ImportError:
|
||||
self._log("opcua package not installed – running in SIMULATION mode")
|
||||
await self._simulate_loop()
|
||||
return
|
||||
|
||||
while not self._stop_event.is_set():
|
||||
try:
|
||||
client = Client(self.opc_url)
|
||||
client.connect()
|
||||
self._log(f"Connected to OPC server: {self.opc_url}")
|
||||
try:
|
||||
while not self._stop_event.is_set():
|
||||
await self._tick(client)
|
||||
await asyncio.sleep(self.poll_interval)
|
||||
finally:
|
||||
client.disconnect()
|
||||
self._log("Disconnected from OPC server")
|
||||
except Exception as exc:
|
||||
self._log(f"OPC connection error: {exc}. Retrying in 5s...")
|
||||
await asyncio.sleep(5)
|
||||
|
||||
self.running = False
|
||||
self._log("OPC polling stopped")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
async def _tick(self, client):
|
||||
"""Read counter; if changed, fetch trackmap nodes and update Oracle."""
|
||||
try:
|
||||
counter_node = client.get_node(self.counter_node)
|
||||
current_counter = counter_node.get_value()
|
||||
except Exception as exc:
|
||||
self._log(f"Failed to read counter node: {exc}")
|
||||
return
|
||||
|
||||
if current_counter == self.last_counter:
|
||||
return # nothing changed
|
||||
|
||||
self._log(
|
||||
f"Counter changed: {self.last_counter} -> {current_counter}. "
|
||||
"Fetching trackmap data..."
|
||||
)
|
||||
self.last_counter = current_counter
|
||||
self.last_update = datetime.now().isoformat()
|
||||
|
||||
if not self.trackmap_nodes:
|
||||
self._log("No trackmap nodes configured – skipping DB update")
|
||||
return
|
||||
|
||||
# Read all configured nodes
|
||||
data: Dict[str, Any] = {}
|
||||
for col, node_id in self.trackmap_nodes.items():
|
||||
try:
|
||||
node = client.get_node(node_id)
|
||||
data[col] = node.get_value()
|
||||
except Exception as exc:
|
||||
self._log(f"Failed to read node {node_id}: {exc}")
|
||||
|
||||
if not data:
|
||||
return
|
||||
|
||||
# Determine POSITION from data (must be one of the mapped columns)
|
||||
position = data.get("position")
|
||||
if position is None:
|
||||
self._log("'position' not in trackmap_nodes data – cannot update row")
|
||||
return
|
||||
|
||||
await self._update_oracle(position, data)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
async def _update_oracle(self, position: Any, data: Dict[str, Any]):
|
||||
"""Write fetched OPC values into PLTM.CMPT_PL_TRACKMAP."""
|
||||
import threading
|
||||
|
||||
def _do_update():
|
||||
try:
|
||||
from database import get_connection
|
||||
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
try:
|
||||
updatable = {k: v for k, v in data.items() if k != "position"}
|
||||
if not updatable:
|
||||
return
|
||||
set_clause = ", ".join(
|
||||
[f"{k.upper()} = :{k}" for k in updatable.keys()]
|
||||
)
|
||||
updatable["position_"] = position
|
||||
sql = (
|
||||
f"UPDATE PLTM.CMPT_PL_TRACKMAP "
|
||||
f"SET {set_clause} WHERE POSITION = :position_"
|
||||
)
|
||||
cursor.execute(sql, updatable)
|
||||
conn.commit()
|
||||
self._log(
|
||||
f"Updated CMPT_PL_TRACKMAP POSITION={position}: "
|
||||
+ ", ".join(f"{k}={v}" for k, v in updatable.items() if k != "position_")
|
||||
)
|
||||
finally:
|
||||
cursor.close()
|
||||
conn.close()
|
||||
except Exception as exc:
|
||||
self._log(f"Oracle update failed: {exc}")
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
await loop.run_in_executor(None, _do_update)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
async def _simulate_loop(self):
|
||||
"""Simulation mode when opcua is not installed."""
|
||||
counter = 0
|
||||
while not self._stop_event.is_set():
|
||||
await asyncio.sleep(self.poll_interval)
|
||||
counter += 1
|
||||
self.last_counter = counter
|
||||
self.last_update = datetime.now().isoformat()
|
||||
self._log(f"[SIM] Counter tick: {counter}")
|
||||
self.running = False
|
||||
|
||||
|
||||
opc_service = OpcService()
|
||||
Reference in New Issue
Block a user