Files
sientia-dataops-opc-ingestor/simulator/redis-feeder.py
vitor-aignosi 8674a710af SIENTIAPDE-1017
Update liveness and readiness probes, enhance Redis connection, and refine OPC tag data structure
2025-04-25 16:57:55 -03:00

56 lines
1.5 KiB
Python

import redis
import json
import os
# Redis connection settings
redis_host = "localhost"
redis_port = 6379
# Connect to Redis
r = redis.Redis(host=redis_host, port=redis_port,
decode_responses=True, username='default', password='bdnZOpcyiL')
# Define the key pattern to target
pattern = "slot:opc_tags:*"
# Step 1: Find and delete matching keys
print("🔍 Searching for keys matching:", pattern)
for key in r.scan_iter(match=pattern):
r.delete(key)
print(f"❌ Deleted: {key}")
# Step 2: Insert new data
# Example new OPC tag data
new_data = {
"slot:opc_tags:1": {
"server1": {
"name": "server1",
"url": "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840",
"server_uri": "http://opcua-server.simulator",
"tags": {
'ns=2;i=2': {
'tag_name': 'Counter',
'frequency': 1000,
'topics': ['opcua', 'counter'],
},
'ns=2;i=3': {
'tag_name': 'Rollout',
'frequency': 1000,
"topics": ['opcua', 'rollout'],
},
'ns=2;i=4': {
'tag_name': 'Square',
'frequency': 1000,
"topics": ['opcua'],
},
}
}
}
}
for key, val in new_data.items():
r.set(key, json.dumps(val))
print(f"✅ Set: {key} -> {val}")
print("🚀 OPC tag keys replaced successfully.")