63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
import yaml
|
|
from opcua_connector import opcua_connector
|
|
import time
|
|
import argparse
|
|
|
|
|
|
|
|
def check_opcua_server(OPCcon):
|
|
"""State 0: Check if OPC UA server is reachable."""
|
|
print("---- State 0 ----")
|
|
try:
|
|
OPCcon.opcuaclient.connect() # Connection happens here
|
|
print("OPC UA server is reachable.")
|
|
return 1 # Move to state 1
|
|
except Exception as e:
|
|
print(f"Error connecting to OPC UA server: {e}")
|
|
|
|
def establish_connection(OPCcon):
|
|
"""State 1: Establish connection to OPC UA server."""
|
|
print("---- State 1 ----")
|
|
try:
|
|
if OPCcon.opcuaclient.uaclient:
|
|
print("OPC UA connection established.")
|
|
return 2 # Move to state 2
|
|
else:
|
|
print(f"Error establishing connection")
|
|
except Exception as e:
|
|
print(f"Error establishing connection: {e}")
|
|
|
|
def push_btn(OPCcon, btn_name):
|
|
"""State 2:"""
|
|
print("---- State 2 ----")
|
|
try:
|
|
OPCcon.press_btn(btn_name)
|
|
print("Data collection complete.")
|
|
return 99 # Return to state 2
|
|
except Exception as e:
|
|
print(f"Error in data collection: {e}.")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# Argument parsing
|
|
parser = argparse.ArgumentParser(description='Run OPC UA button press script.')
|
|
parser.add_argument('-n', '--name', type=str, choices=['BTN_START', 'BTN_STOP'], required=True, help='Button name to press (BTN_START or BTN_STOP)')
|
|
args = parser.parse_args()
|
|
btn_name = args.name # Get button name from arguments
|
|
|
|
# INIT
|
|
config = yaml.safe_load(open('./cfg.yaml'))
|
|
OPCcon = opcua_connector(config)
|
|
|
|
# State machine loop
|
|
state = 0
|
|
if state == 0:
|
|
state = check_opcua_server(OPCcon)
|
|
elif state == 1:
|
|
state = establish_connection(OPCcon)
|
|
elif state == 2:
|
|
state = push_btn(OPCcon, btn_name)
|
|
|
|
# Close the InfluxDB connection when done
|
|
OPCcon.influxclient.close()
|
|
OPCcon.opcuaclient.disconnect() # Disconnect from OPC UA after state machine finishes |