SIENTIAPDE-1094

Update settings for language services, adjust simulator port, enhance error handling in ingestor, and refine Redis feeder script

.vscode/settings.json

@@ -8,5 +8,6 @@
   },
   "python.languageServer": "Pylance",
   "python.analysis.typeCheckingMode": "standard",
-  "editor.suggestSelection": "first"
+  "editor.suggestSelection": "first",
+  "windsurfPyright.disableLanguageServices": true
 }

docker-compose.yaml

@@ -88,7 +88,7 @@ services:
         GIT_BRANCH: ${SIMULATOR_GIT_BRANCH}
     container_name: simulator
     ports:
-      - "4840:4840"
+      - "4841:4840"
     depends_on:
       - kafka
       - redis

ingestor/app.py

@@ -16,14 +16,21 @@ POD_ID = os.getenv("HOSTNAME", "localhost")
 def main():
     start_prometheus_server()
     ingestor = Ingestor()
-    ingestor.prepare_ingestor()
+    try:
+        ingestor.prepare_ingestor()
+    except Exception as e:
+        metrics.APP_ERRORS_TOTAL.labels(
+            pod_id=POD_ID).inc()  # Increment errors
+        print(f"Failed to prepare ingestor: {e}")
+        exit_signal.set()
     ingestor.logger.info("Ingestor prepared. Starting main loop.")
 
     while not exit_signal.is_set():
         start_time = time()  # Start loop timer
         try:
             ingestor.loop()
-            metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc()  # Increment loop counter
+            metrics.APP_LOOP_COUNT.labels(
+                pod_id=POD_ID).inc()  # Increment loop counter
 
             exit_signal.wait(ingestor.poll_interval)
 
@@ -33,7 +40,8 @@ def main():
         except Exception:
             print("Exception in main loop. Setting exit_signal flag.")
             traceback.print_exc()
-            metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()  # Increment errors
+            metrics.APP_ERRORS_TOTAL.labels(
+                pod_id=POD_ID).inc()  # Increment errors
             exit_signal.set()
         finally:
             # Record loop duration

ingestor/ingestor.py

@@ -49,13 +49,14 @@ class Ingestor:
         self.notification_handler = NotificationHandler(
             servers=self.kafka_servers,
             logger=self.logger,
-            project_name="OPC_INGESTOR",
-            pipeline_name="-",
-            trigger_name="-",
-            model_name="-",
-            model="-",
+            project_name="OPC_INGESTOR"
         )
 
+        self.notification_handler.base_notification.pipeline = 'OPC_INGESTOR'
+        self.notification_handler.base_notification.trigger = 'INGESTOR'
+        self.notification_handler.base_notification.model_name = '-'
+        self.notification_handler.base_notification.model_id = '-'
+
         self.ingestor_manager = None
 
     def shutdown(self):
@@ -79,7 +80,8 @@ class Ingestor:
         logger = getLogger(__name__)
         logger.setLevel(getenv("LOG_LEVEL", "INFO"))
         handler = StreamHandler()
-        formatter = Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+        formatter = Formatter(
+            "%(asctime)s - %(name)s - %(levelname)s - %(message)s")
         handler.setFormatter(formatter)
 
         logger.addHandler(handler)

requirements.txt

@@ -1,4 +1,4 @@
 asyncua==1.1.5
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14
 prometheus_client

simulator/redis-feeder.py

@@ -1,5 +1,6 @@
-import redis
 import json
+import redis
+
 
 # Redis connection settings
 REDIS_HOST = "localhost"
@@ -7,19 +8,19 @@ REDIS_PORT = 6379
 REDIS_USERNAME = None  # "default"
 REDIS_PASSWORD = None  # "bdnZOpcyiL"
 
-OPC_URL = "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840"
-OPC_URL = "opc.tcp://localhost:4840"
+OPC_URL = "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4841"
+OPC_URL = "opc.tcp://localhost:4841"
 
 # Connect to Redis
 r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT,
                 decode_responses=True, username=REDIS_USERNAME, password=REDIS_PASSWORD)
 
 # Define the key pattern to target
-pattern = "slot:opc_tags:*"
+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):
+print("🔍 Searching for keys matching:", PATTERN)
+for key in r.scan_iter(match=PATTERN):
     r.delete(key)
     print(f"❌ Deleted: {key}")