def start_opc_ssh_tunnel(self) -> None:
"""
Поднимает SSH-туннель до OPC сервера: пробрасывает порт OPC на runner.
После этого OPC доступен на runner'е как opc.tcp://localhost:<port>.
"""
from constants.architecture_constants import OpcUaConstants as OpcConst
opc_host, opc_port = self._parse_opc_target()
self._opc_tunnel_port = opc_port
process = self._stand_client.open_ssh_tunnel(opc_port, opc_host, opc_port)
if process is None:
raise RuntimeError("[SETUP] [ERROR] Не удалось запустить OPC SSH-туннель")
self._opc_tunnel_process = process
time.sleep(OpcConst.TUNNEL_START_DELAY_S)
check_cmd = (
f"timeout 5 bash -c 'cat < /dev/null > /dev/tcp/localhost/{opc_port}' "
f"&& echo {Im_const.CMD_STATUS_OK} || echo {Im_const.CMD_STATUS_FAIL}"
)
result = self._stand_client.run_cmd(check_cmd, need_output=True, use_ssh=False)
if result != Im_const.CMD_STATUS_OK:
self.stop_opc_ssh_tunnel()
raise RuntimeError(
f"[SETUP] [ERROR] OPC SSH-туннель не поднялся: localhost:{opc_port} недоступен с runner'а"
)
logger.info("[SETUP] [OK] OPC SSH-туннель поднят: localhost:%s -> %s:%s", opc_port, opc_host, opc_port)
def stop_opc_ssh_tunnel(self) -> None:
"""Закрывает SSH-туннель до OPC сервера."""
if self._opc_tunnel_process is None:
return
SubprocessClient.terminate_process(self._opc_tunnel_process, timeout=5.0)
self._opc_tunnel_process = None
self._opc_tunnel_port = None
logger.info("[TEARDOWN] [OK] OPC SSH-туннель закрыт")
@property
def opc_tunnel_url(self) -> Optional[str]:
"""URL OPC через SSH-туннель (localhost) либо None, если туннель не поднят."""
if self._opc_tunnel_port:
return f"opc.tcp://localhost:{self._opc_tunnel_port}"
return None