diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml
index 843eec5..f9b92b6 100644
--- a/.github/workflows/docker-image.yml
+++ b/.github/workflows/docker-image.yml
@@ -588,7 +588,7 @@ jobs:
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- - name: Build and push oracle_plugin
+ - name: Build and push ckn_plugin
uses: docker/build-push-action@v5
with:
context: external_plugins/image_detecting_plugin
@@ -597,23 +597,11 @@ jobs:
tags: tapis/image_detecting_plugin:${{ env.TRAPS_REL }}
build-args: REL=${{ env.TRAPS_REL }}
- power_measuring_plugin:
+ power_measuring_plugin_deps:
runs-on: ubuntu-latest
environment: main
needs: camera_traps_py_313
steps:
- - name: Set TRAPS_REL
- run: |
- if [[ ${{ github.event_name == 'workflow_dispatch' }} == true ]]; then
- TRAPS_REL=${{ inputs.release_tag }}
- elif [[ ${{ github.event_name == 'push' }} == true ]]; then
- TRAPS_REL=latest
- elif [[ ${{ github.event_name == 'pull_request' }} == true ]]; then
- TRAPS_REL=latest
- else
- TRAPS_REL=${{ github.event.release.tag_name }}
- fi
- echo "TRAPS_REL=$TRAPS_REL" >> "$GITHUB_ENV"
- name: Checkout
uses: actions/checkout@v4
- name: Set up QEMU
@@ -641,6 +629,35 @@ jobs:
file: external_plugins/power_measuring_plugin/Dockerfile-scaphandre
platforms: linux/amd64,linux/arm64
tags: tapis/scaphandre
+
+ power_measuring_plugin:
+ runs-on: ubuntu-latest
+ environment: main
+ needs: power_measuring_plugin_deps
+ steps:
+ - name: Set TRAPS_REL
+ run: |
+ if [[ ${{ github.event_name == 'workflow_dispatch' }} == true ]]; then
+ TRAPS_REL=${{ inputs.release_tag }}
+ elif [[ ${{ github.event_name == 'push' }} == true ]]; then
+ TRAPS_REL=latest
+ elif [[ ${{ github.event_name == 'pull_request' }} == true ]]; then
+ TRAPS_REL=latest
+ else
+ TRAPS_REL=${{ github.event.release.tag_name }}
+ fi
+ echo "TRAPS_REL=$TRAPS_REL" >> "$GITHUB_ENV"
+ - name: Checkout
+ uses: actions/checkout@v4
+ - name: Set up QEMU
+ uses: docker/setup-qemu-action@v3
+ - name: Set up Docker Buildx
+ uses: docker/setup-buildx-action@v3
+ - name: Login to Docker Hub
+ uses: docker/login-action@v3
+ with:
+ username: ${{ secrets.DOCKERHUB_USERNAME }}
+ password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Build and push power_measuring_plugin
uses: docker/build-push-action@v5
with:
@@ -650,7 +667,7 @@ jobs:
tags: tapis/power_measuring_plugin_py:${{ env.TRAPS_REL }}
build-args: REL=${{ env.TRAPS_REL }}
- oracle_plugin:
+ ckn_plugin:
runs-on: ubuntu-latest
environment: main
needs: camera_traps_py_313
@@ -678,13 +695,13 @@ jobs:
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- - name: Build and push oracle_plugin
+ - name: Build and push ckn_plugin
uses: docker/build-push-action@v5
with:
- context: external_plugins/oracle_plugin
+ context: external_plugins/ckn_plugin
platforms: linux/amd64,linux/arm64
push: true
- tags: tapis/oracle_plugin:${{ env.TRAPS_REL }}
+ tags: tapis/ckn_plugin:${{ env.TRAPS_REL }}
build-args: REL=${{ env.TRAPS_REL }}
detection_reporter_plugin:
@@ -715,7 +732,7 @@ jobs:
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- - name: Build and push oracle_plugin
+ - name: Build and push ckn_plugin
uses: docker/build-push-action@v5
with:
context: external_plugins/detection_reporter_plugin
@@ -752,7 +769,7 @@ jobs:
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- - name: Build and push oracle_plugin
+ - name: Build and push ckn_plugin
uses: docker/build-push-action@v5
with:
context: external_plugins/video_generating_plugin
@@ -822,11 +839,11 @@ jobs:
trigger_tests:
runs-on: ubuntu-latest
environment: main
- needs: [engine, image_scoring_plugin_server_py_313, image_scoring_plugin_ultralytics_py_313, image_scoring_plugin_yolov5_py_38, image_scoring_plugin_py_nano_38, image_generating_plugin, image_detecting_plugin, power_measuring_plugin, oracle_plugin, detection_reporter_plugin, video_generating_plugin, custom_install]
+ needs: [engine, image_scoring_plugin_server_py_313, image_scoring_plugin_ultralytics_py_313, image_scoring_plugin_yolov5_py_38, image_scoring_plugin_py_nano_38, image_generating_plugin, image_detecting_plugin, power_measuring_plugin_deps, power_measuring_plugin, ckn_plugin, detection_reporter_plugin, video_generating_plugin, custom_install]
steps:
- name: Check if tests are disabled
run: |
- if [[ ${{ github.event_name == 'workflow_dispatch' }} == true ]] && [ "${{ inputs.run_tests }}" == "false" ];
+ if [[ "${{ github.event_name }}" == "workflow_dispatch" ]] && [ "${{ inputs.run_tests }}" == "false" ];
then
echo "Tests are disabled"
exit 1
diff --git a/Cargo.lock b/Cargo.lock
index 3602cb7..efa81c0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,6 +1,6 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
-version = 3
+version = 4
[[package]]
name = "android_system_properties"
@@ -613,9 +613,9 @@ dependencies = [
[[package]]
name = "traitobject"
-version = "0.1.0"
+version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "efd1f82c56340fdf16f2a953d7bda4f8fdffba13d93b00844c25572110b26079"
+checksum = "04a79e25382e2e852e8da874249358d382ebaf259d0d34e75d8db16a7efabbc7"
[[package]]
name = "typemap"
diff --git a/Dockerfile b/Dockerfile
index ef32b2a..5a4c4d1 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -5,7 +5,7 @@
# -------------------
# First build phase: In this phase we do the build for release with an intermedidate layer that caches the dependencies
# --------------------
-FROM rust:1.75 as builder
+FROM rust:1.90 as builder
# install libzmq
RUN USER=root apt-get update && apt-get install -y libzmq3-dev
diff --git a/Makefile b/Makefile
index 7e5c19c..a9842e3 100644
--- a/Makefile
+++ b/Makefile
@@ -17,7 +17,7 @@
# Example: 'make build' builds all targets.
clean:
- cd releases/${TRAPS_REL}; rm -rf power_output_dir/*; rm -rf images_output_dir/*; rm -rf oracle_plugin_dir/*;
+ cd releases/${TRAPS_REL}; rm -rf power_output_dir/*; rm -rf images_output_dir/*; rm -rf ckn_plugin_dir/*;
build-engine:
docker build -t tapis/camera_traps_engine:${TRAPS_REL} --build-arg TRAPS_REL=${TRAPS_REL} .
@@ -47,8 +47,8 @@ build-generating:
build-power:
cd external_plugins/power_measuring_plugin/ && docker build -t tapis/powerjoular -f Dockerfile-powerjoular . && docker build -t tapis/scaphandre -f Dockerfile-scaphandre . && docker build -t tapis/power_measuring_plugin_py:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
-build-oracle:
- cd external_plugins/oracle_plugin/ && docker build -t tapis/oracle_plugin:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
+build-ckn:
+ cd external_plugins/ckn_plugin/ && docker build -t tapis/ckn_plugin:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
build-detection:
cd external_plugins/detection_reporter_plugin && docker build -t tapis/detection_reporter_plugin:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
@@ -59,7 +59,7 @@ build-detecting:
build-video:
cd external_plugins/video_generating_plugin/ && docker build -t tapis/video_generating_plugin:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
-build-py-plugins: build-camerapy build-scoring-server build-generating build-power build-oracle build-detection build-detecting build-video
+build-py-plugins: build-camerapy build-scoring-server build-generating build-power build-ckn build-detection build-detecting build-video
build-installer:
cd installer && docker build -t tapis/camera-traps-installer:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../
@@ -67,10 +67,10 @@ build-installer:
build: build-engine build-py-plugins build-installer
tag: build
- docker tag tapis/camera_traps_py:${TRAPS_REL} tapis/camera_traps_py & docker tag tapis/camera_traps_py_3.8:${TRAPS_REL} tapis/camera_traps_py_3.8 & docker tag tapis/image_scoring_plugin_py_3.8:${TRAPS_REL} tapis/image_scoring_plugin_py_3.8 & docker tag tapis/image_scoring_plugin_py_nano_3.8:${TRAPS_REL} tapis/image_scoring_plugin_py_nano_3.8 & docker tag tapis/image_generating_plugin_py:${TRAPS_REL} tapis/image_generating_plugin_py & docker tag tapis/power_measuring_plugin_py:${TRAPS_REL} tapis/power_measuring_plugin_py & docker tag tapis/camera_traps_engine:${TRAPS_REL} tapis/camera_traps_engine & docker tag tapis/oracle_plugin:${TRAPS_REL} tapis/oracle_plugin & docker tag tapis/camera-traps-installer:${TRAPS_REL} tapis/camera-traps-installer
+ docker tag tapis/camera_traps_py:${TRAPS_REL} tapis/camera_traps_py & docker tag tapis/camera_traps_py_3.8:${TRAPS_REL} tapis/camera_traps_py_3.8 & docker tag tapis/image_scoring_plugin_py_3.8:${TRAPS_REL} tapis/image_scoring_plugin_py_3.8 & docker tag tapis/image_scoring_plugin_py_nano_3.8:${TRAPS_REL} tapis/image_scoring_plugin_py_nano_3.8 & docker tag tapis/image_generating_plugin_py:${TRAPS_REL} tapis/image_generating_plugin_py & docker tag tapis/power_measuring_plugin_py:${TRAPS_REL} tapis/power_measuring_plugin_py & docker tag tapis/camera_traps_engine:${TRAPS_REL} tapis/camera_traps_engine & docker tag tapis/ckn_plugin:${TRAPS_REL} tapis/ckn_plugin & docker tag tapis/camera-traps-installer:${TRAPS_REL} tapis/camera-traps-installer
push: tag
- docker push tapis/camera_traps_py & docker push tapis/camera_traps_py:${TRAPS_REL} & docker push tapis/camera_traps_py_3.8 & docker push tapis/camera_traps_py_3.8:${TRAPS_REL} & docker push tapis/image_scoring_plugin_py_3.8 & docker push tapis/image_scoring_plugin_py_3.8:${TRAPS_REL} & docker push tapis/image_scoring_plugin_py_nano_3.8 & docker push tapis/image_scoring_plugin_py_nano_3.8:${TRAPS_REL} & docker push tapis/image_generating_plugin_py & docker push tapis/image_generating_plugin_py:${TRAPS_REL} & docker push tapis/power_measuring_plugin_py:${TRAPS_REL} & docker push tapis/oracle_plugin:${TRAPS_REL} & docker push tapis/camera_traps_engine & docker push tapis/camera_traps_engine:${TRAPS_REL} & docker push tapis/camera-traps-installer:${TRAPS_REL} & docker push tapis/camera-traps-installer
+ docker push tapis/camera_traps_py & docker push tapis/camera_traps_py:${TRAPS_REL} & docker push tapis/camera_traps_py_3.8 & docker push tapis/camera_traps_py_3.8:${TRAPS_REL} & docker push tapis/image_scoring_plugin_py_3.8 & docker push tapis/image_scoring_plugin_py_3.8:${TRAPS_REL} & docker push tapis/image_scoring_plugin_py_nano_3.8 & docker push tapis/image_scoring_plugin_py_nano_3.8:${TRAPS_REL} & docker push tapis/image_generating_plugin_py & docker push tapis/image_generating_plugin_py:${TRAPS_REL} & docker push tapis/power_measuring_plugin_py:${TRAPS_REL} & docker push tapis/ckn_plugin:${TRAPS_REL} & docker push tapis/camera_traps_engine & docker push tapis/camera_traps_engine:${TRAPS_REL} & docker push tapis/camera-traps-installer:${TRAPS_REL} & docker push tapis/camera-traps-installer
push-all:
docker buildx build --platform linux/arm64,linux/amd64 --push -t tapis/camera_traps_engine:${TRAPS_REL} --build-arg TRAPS_REL=${TRAPS_REL} .
@@ -79,5 +79,5 @@ push-all:
cd external_plugins/image_scoring_plugin/ && docker buildx build --platform linux/arm64 -t tapis/image_scoring_plugin_py_nano_3.8:${TRAPS_REL} --build-arg REL=${TRAPS_REL} -f Dockerfile-3.8-nano .; cd ../..
cd external_plugins/image_generating_plugin/ && docker buildx build --platform linux/arm64,linux/amd64 --push -t tapis/image_generating_plugin_py:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
cd external_plugins/power_measuring_plugin/ && docker buildx build --platform linux/arm64,linux/amd64 --push -t tapis/power_measuring_plugin_py:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
- cd external_plugins/oracle_plugin/ && docker buildx build --platform linux/arm64,linux/amd64 --push -t tapis/oracle_plugin:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
+ cd external_plugins/ckn_plugin/ && docker buildx build --platform linux/arm64,linux/amd64 --push -t tapis/ckn_plugin:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../..
cd installer && docker buildx build --platform linux/arm64,linux/amd64 --push -t tapis/camera-traps-installer:${TRAPS_REL} --build-arg REL=${TRAPS_REL} .; cd ../
diff --git a/README.md b/README.md
index 45beb2b..1ffe05e 100644
--- a/README.md
+++ b/README.md
@@ -36,7 +36,7 @@ In general, plugins can also depend on their own environment variables and/or co
| detection_reporter_plugin| TRAPS_DETECTION_REPORTER_* | /traps-detection.toml |
| image_store_plugin | TRAPS_IMAGE_STORE_FILE | ~/traps-image-store.toml | |
| power_measure_plugin | TRAPS_POWER_LOG_PATH | ~/logs | |
-| oracle_monitor_plugin | TRAPS_ORACLE_OUTPUT_PATH | ~/output | |
+| ckn_plugin | TRAPS_CKN_OUTPUT_PATH | ~/output | |
| integration tests | TRAPS_INTEGRATION_CONFIG_FILE | ~/traps-integration.toml | |
| logger | TRAPS_LOG4RS_CONFIG_FILE | resources/log4rs.yml | Packaged with application |
@@ -123,7 +123,7 @@ Camera-traps uses a [TOML](https://toml.io/en/) file to configure the internal a
> "PluginTerminateEvent"
> ]
> [[plugins.external]]
-> plugin_name = "ext_oracle_monitor_plugin"
+> plugin_name = "ext_ckn_plugin"
> id = "6e153711-9823-4ee6-b608-58e2e801db51"
> external_port = 6011
> subscriptions = [
diff --git a/external_plugins/ckn_plugin/Dockerfile b/external_plugins/ckn_plugin/Dockerfile
new file mode 100644
index 0000000..9c8303d
--- /dev/null
+++ b/external_plugins/ckn_plugin/Dockerfile
@@ -0,0 +1,12 @@
+ARG REL=use-build-arg
+FROM tapis/camera_traps_py_3.13:$REL
+RUN find / -path /proc -prune -o -path /sys -prune -o -path /dev -prune -o -writable -print0 | xargs -0 -P $(nproc) -I {} chmod 0777 {} || true
+
+# Install confluent-kafka for Kafka streaming
+RUN pip install --no-cache-dir confluent-kafka
+
+ADD ckn_plugin.py /ckn_plugin.py
+ADD power_processor.py /power_processor.py
+RUN chmod -R 0777 /ckn_plugin.py /power_processor.py || true
+
+ENTRYPOINT [ "python", "-u", "/ckn_plugin.py" ]
diff --git a/external_plugins/ckn_plugin/ckn_plugin.py b/external_plugins/ckn_plugin/ckn_plugin.py
new file mode 100644
index 0000000..74f1392
--- /dev/null
+++ b/external_plugins/ckn_plugin/ckn_plugin.py
@@ -0,0 +1,905 @@
+"""
+CKN Plugin - Combined Oracle and CKN Daemon functionality.
+
+This plugin:
+1. Listens on ZMQ for image events (ImageReceived, ImageScored, ImageStored, ImageDeleted, PluginTerminating)
+2. Keeps mapping state in memory (no image_mapping_final.json)
+3. Computes running experiment metrics (TP/FP/FN, precision, recall, f1, IoU, mAP)
+4. Streams per-image events to Kafka when image_decision is set (when CKN is configured)
+5. Streams power summary to Kafka at shutdown (when power monitoring is enabled)
+"""
+
+import os
+import sys
+import zmq
+import logging
+import json
+import time
+
+from pyevents.events import get_plugin_socket, get_next_msg, send_quit_command
+from ctevents.ctevents import socket_message_to_typed_event, send_terminate_plugin_fb_event
+from ctevents import ImageStoredEvent, ImageDeletedEvent, ImageScoredEvent, ImageReceivedEvent, PluginTerminatingEvent
+
+# Try to import Kafka producer, handle gracefully if not available
+KAFKA_AVAILABLE = False
+try:
+ from confluent_kafka import Producer, KafkaError
+ from confluent_kafka.admin import AdminClient
+ KAFKA_AVAILABLE = True
+except ImportError:
+ pass
+
+# ============================================================================
+# Logging setup
+# ============================================================================
+log_level = os.environ.get("CKN_LOG_LEVEL", "INFO")
+LOG_FILE = os.environ.get("CKN_LOG_FILE", "/output/ckn_plugin.log")
+
+logger = logging.getLogger("CKN Plugin")
+if log_level == "DEBUG":
+ logger.setLevel(logging.DEBUG)
+elif log_level == "INFO":
+ logger.setLevel(logging.INFO)
+elif log_level == "WARN":
+ logger.setLevel(logging.WARN)
+elif log_level == "ERROR":
+ logger.setLevel(logging.ERROR)
+
+if not logger.handlers:
+ formatter = logging.Formatter('%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]')
+
+ # Console handler
+ console_handler = logging.StreamHandler()
+ console_handler.setFormatter(formatter)
+ logger.addHandler(console_handler)
+
+ # File handler
+ try:
+ log_dir = os.path.dirname(LOG_FILE)
+ if log_dir and not os.path.exists(log_dir):
+ os.makedirs(log_dir, exist_ok=True)
+ file_handler = logging.FileHandler(LOG_FILE)
+ file_handler.setFormatter(formatter)
+ logger.addHandler(file_handler)
+ logger.info(f"Logging to file: {LOG_FILE}")
+ except Exception as e:
+ logger.warning(f"Could not create log file {LOG_FILE}: {e}")
+
+# ============================================================================
+# Configuration from environment
+# ============================================================================
+PORT = int(os.environ.get('CKN_PLUGIN_PORT', 6011))
+OUTPUT_DIR = os.environ.get('TRAPS_CKN_OUTPUT_PATH', "/output/")
+MODEL_ID = os.environ.get("MODEL_ID")
+
+# Ground truth file (written by image generating plugin, read-only for us)
+uuid_image_mapping_path = os.path.join(OUTPUT_DIR, "uuid_image_mapping.json")
+
+# Kafka configuration
+KAFKA_BROKER = os.environ.get('CKN_KAFKA_BROKER', '')
+KAFKA_TOPIC = os.environ.get('CKN_KAFKA_TOPIC', 'oracle-events')
+KAFKA_SECURITY_PROTOCOL = os.environ.get('CKN_KAFKA_SECURITY_PROTOCOL', 'SSL')
+DEVICE_ID = os.environ.get('CAMERA_TRAPS_DEVICE_ID', '')
+CKN_DOMAIN = os.environ.get('CKN_DOMAIN', 'animal-ecology')
+USER_ID = os.environ.get('USER_ID', '')
+EXPERIMENT_ID = os.environ.get('EXPERIMENT_ID', '')
+
+# Power monitoring configuration
+ENABLE_POWER_MONITORING = os.environ.get('ENABLE_POWER_MONITORING', 'false').lower()
+POWER_SUMMARY_FILE = os.environ.get('POWER_SUMMARY_FILE', '')
+POWER_SUMMARY_TOPIC = os.environ.get('POWER_SUMMARY_TOPIC', 'cameratraps-power-summary')
+POWER_SUMMARY_TIMEOUT = int(os.environ.get('POWER_SUMMARY_TIMEOUT', 10))
+POWER_SUMMARY_MAX_TRIES = int(os.environ.get('POWER_SUMMARY_MAX_TRIES', 5))
+
+# Special UUID used to signal experiment end
+EXPERIMENT_END_SIGNAL = os.getenv('EXPERIMENT_END_SIGNAL', '6e153711-9823-4ee6-b608-58e2e801db51')
+
+SOCKET_TIMEOUT = 2000
+
+# ============================================================================
+# Global state
+# ============================================================================
+# In-memory mapping (replaces image_mapping_final.json)
+existing_image_mapping = {}
+
+# Number of images generated by the entire application
+total_images_generated = 0
+
+# Number of images processed (have image_decision set)
+total_images_processed = 0
+
+# Whether we received the PluginTerminating event from image generating plugin
+received_terminating_signal = False
+
+# UUIDs we couldn't initially retrieve from uuid_image_mapping
+uuids_with_errors = []
+
+# Set of UUIDs already streamed to Kafka
+processed_uuids = set()
+
+# Kafka producer (initialized if KAFKA_BROKER is set)
+kafka_producer = None
+
+# Running experiment metrics
+experiment_metrics = {
+ "total_images": 0,
+ "total_predictions": 0,
+ "total_ground_truth_objects": 0,
+ "true_positives": 0,
+ "false_positives": 0,
+ "false_negatives": 0,
+ "precision": 0.0,
+ "recall": 0.0,
+ "f1_score": 0.0,
+ "mean_iou": None,
+ "map_50": None,
+ "map_50_95": None,
+ # Internal accumulators
+ "sum_iou": 0.0,
+ "num_iou_pairs": 0,
+ "gt_boxes_count": 0,
+}
+
+# For mAP: per-threshold prediction lists of (score, is_tp)
+map_thresholds = [round(t, 2) for t in [0.50, 0.55, 0.60, 0.65, 0.70, 0.75, 0.80, 0.85, 0.90, 0.95]]
+map_data = {t: [] for t in map_thresholds}
+
+
+# ============================================================================
+# Kafka Connect JSON schema envelopes
+# ============================================================================
+# The JDBC sink connectors on cknkafkaconnect run with
+# value.converter.schemas.enable=true, so JsonConverter requires each record
+# to be a {"schema": ..., "payload": ...} envelope rather than bare JSON.
+EVENT_FIELD_TYPES = {
+ "domain": "string",
+ "device_id": "string",
+ "experiment_id": "string",
+ "user_id": "string",
+ "model_id": "string",
+ "image_count": "int32",
+ "UUID": "string",
+ "image_name": "string",
+ "ground_truth": "string",
+ "image_receiving_timestamp": "string",
+ "image_scoring_timestamp": "string",
+ "image_store_delete_time": "string",
+ "label": "string",
+ "probability": "double",
+ "image_decision": "string",
+ "flattened_scores": "string",
+ "total_images": "int32",
+ "total_predictions": "int32",
+ "total_ground_truth_objects": "int32",
+ "true_positives": "int32",
+ "false_positives": "int32",
+ "false_negatives": "int32",
+ "precision": "double",
+ "recall": "double",
+ "f1_score": "double",
+ "mean_iou": "double",
+ "map_50": "double",
+ "map_50_95": "double",
+}
+EVENT_REQUIRED_FIELDS = {"UUID", "experiment_id", "user_id"}
+
+
+def build_connect_envelope(payload, field_types, required_fields=None):
+ """Wrap a flat dict in a Kafka Connect JSON schema envelope (schema+payload)."""
+ required_fields = required_fields or set()
+ fields = [
+ {"field": name, "type": conn_type, "optional": name not in required_fields}
+ for name, conn_type in field_types.items()
+ ]
+ schema = {"type": "struct", "fields": fields, "optional": False}
+ return {"schema": schema, "payload": payload}
+
+
+def build_power_field_types(flattened_event):
+ """Power summary keys are dynamic (per-plugin names), so infer types from the event."""
+ return {
+ name: "string" if name == "experiment_id" else "double"
+ for name in flattened_event
+ }
+
+
+# ============================================================================
+# Kafka helpers
+# ============================================================================
+def validate_broker_address(broker_address):
+ """Validate the broker address format."""
+ if not broker_address:
+ return False, "Broker address is empty"
+
+ # Check basic format (host:port)
+ if ':' not in broker_address:
+ return False, f"Invalid broker format '{broker_address}' - expected 'host:port'"
+
+ parts = broker_address.rsplit(':', 1)
+ host = parts[0]
+ port_str = parts[1]
+
+ # Validate port
+ try:
+ port = int(port_str)
+ if port < 1 or port > 65535:
+ return False, f"Invalid port {port} - must be between 1 and 65535"
+ except ValueError:
+ return False, f"Invalid port '{port_str}' - must be a number"
+
+ # Validate host is not empty
+ if not host:
+ return False, "Host cannot be empty"
+
+ return True, f"Broker address '{broker_address}' is valid (host={host}, port={port})"
+
+
+def test_kafka_connection(configuration, timeout=10, num_tries=5):
+ """Check if the Kafka broker is up and running."""
+ if not KAFKA_AVAILABLE:
+ logger.error("Kafka client library (confluent-kafka) not available")
+ return False
+
+ broker = configuration.get('bootstrap.servers', 'unknown')
+ security_protocol = configuration.get('security.protocol', 'unknown')
+
+ logger.info(f"Testing Kafka connection to {broker} with protocol {security_protocol}")
+
+ for i in range(num_tries):
+ try:
+ logger.info(f"Connection attempt {i+1}/{num_tries}...")
+ admin_client = AdminClient(configuration)
+ topics = admin_client.list_topics(timeout=timeout)
+ topic_list = list(topics.topics.keys())
+ logger.info(f"Successfully connected! Available topics: {topic_list[:10]}{'...' if len(topic_list) > 10 else ''}")
+ return True
+ except Exception as e:
+ error_msg = str(e)
+ logger.warning(f"Kafka connection attempt {i+1}/{num_tries} failed: {error_msg}")
+ if i < num_tries - 1:
+ logger.info(f"Retrying in 5 seconds...")
+ time.sleep(5)
+
+ logger.error(f"Could not connect to Kafka broker after {num_tries} attempts")
+ return False
+
+
+def init_kafka_producer():
+ """Initialize Kafka producer if configured."""
+ global kafka_producer
+
+ logger.info("="*60)
+ logger.info("Initializing Kafka Producer")
+ logger.info("="*60)
+
+ # Check if Kafka library is available
+ if not KAFKA_AVAILABLE:
+ logger.error("Kafka client library (confluent-kafka) not installed. CKN streaming disabled.")
+ return False
+
+ # Check if broker is configured
+ if not KAFKA_BROKER:
+ logger.warning("CKN_KAFKA_BROKER environment variable not set. CKN streaming disabled.")
+ return False
+
+ # Validate broker address format
+ is_valid, validation_msg = validate_broker_address(KAFKA_BROKER)
+ logger.info(f"Broker validation: {validation_msg}")
+ if not is_valid:
+ logger.error(f"Invalid broker address: {validation_msg}")
+ return False
+
+ # Log configuration
+ logger.info(f"Kafka Configuration:")
+ logger.info(f" - Broker: {KAFKA_BROKER}")
+ logger.info(f" - Security Protocol: {KAFKA_SECURITY_PROTOCOL}")
+ logger.info(f" - Topic: {KAFKA_TOPIC}")
+ logger.info(f" - Experiment ID: {EXPERIMENT_ID}")
+ logger.info(f" - User ID: {USER_ID}")
+ logger.info(f" - Device ID: {DEVICE_ID}")
+
+ kafka_conf = {
+ 'bootstrap.servers': KAFKA_BROKER,
+ 'log_level': 0,
+ 'security.protocol': KAFKA_SECURITY_PROTOCOL
+ }
+
+ logger.info(f"Attempting to connect to Kafka broker at {KAFKA_BROKER}...")
+
+ if not test_kafka_connection(kafka_conf):
+ logger.error("Failed to connect to Kafka broker. CKN streaming disabled.")
+ logger.info("="*60)
+ return False
+
+ logger.info(f"Successfully connected to Kafka broker!")
+ kafka_producer = Producer(**kafka_conf)
+ logger.info("Kafka producer initialized successfully")
+ logger.info("="*60)
+ return True
+
+
+# ============================================================================
+# Metrics computation (ported from oracle_ckn_daemon)
+# ============================================================================
+def _compute_iou(box_a, box_b):
+ """Compute IoU between two boxes in [x, y, w, h] format with normalized coordinates."""
+ ax, ay, aw, ah = box_a
+ bx, by, bw, bh = box_b
+ ax2, ay2 = ax + aw, ay + ah
+ bx2, by2 = bx + bw, by + bh
+ inter_x1 = max(ax, bx)
+ inter_y1 = max(ay, by)
+ inter_x2 = min(ax2, bx2)
+ inter_y2 = min(ay2, by2)
+ inter_w = max(0.0, inter_x2 - inter_x1)
+ inter_h = max(0.0, inter_y2 - inter_y1)
+ inter_area = inter_w * inter_h
+ area_a = max(0.0, aw) * max(0.0, ah)
+ area_b = max(0.0, bw) * max(0.0, bh)
+ union = area_a + area_b - inter_area
+ return (inter_area / union) if union > 0 else 0.0
+
+
+def _greedy_match(preds, gts, iou_threshold):
+ """
+ Greedy 1-1 matching between predictions and ground truths by IoU threshold.
+ Returns list of (pred_idx, gt_idx, iou) for matches.
+ """
+ matches = []
+ used_preds = set()
+ used_gts = set()
+ # Precompute IoU matrix
+ candidates = []
+ for pi, p in enumerate(preds):
+ pbox = p.get("bounding_box")
+ if pbox is None:
+ continue
+ for gi, g in enumerate(gts):
+ gbox = g.get("bounding_box")
+ if gbox is None:
+ continue
+ iou = _compute_iou(pbox, gbox)
+ candidates.append((iou, pi, gi))
+ # Sort by descending IoU
+ candidates.sort(reverse=True)
+ for iou, pi, gi in candidates:
+ if iou < iou_threshold:
+ break
+ if pi in used_preds or gi in used_gts:
+ continue
+ matches.append((pi, gi, iou))
+ used_preds.add(pi)
+ used_gts.add(gi)
+ return matches, used_preds, used_gts
+
+
+def _update_map_structures(predictions, gt_boxes):
+ """Update per-threshold AP structures using label-aware matches."""
+ global map_data
+ for thr in map_thresholds:
+ matches, used_preds, used_gts = _greedy_match(predictions, gt_boxes, thr)
+ matched_gt_by_pred = {pi: gi for pi, gi, _ in matches}
+ for pi, pred in enumerate(predictions):
+ score = float(pred.get("probability", 0.0))
+ if pi in matched_gt_by_pred:
+ gi = matched_gt_by_pred[pi]
+ if str(pred.get("label")) == str(gt_boxes[gi].get("label")):
+ map_data[thr].append((score, True))
+ else:
+ map_data[thr].append((score, False))
+ else:
+ map_data[thr].append((score, False))
+
+
+def _compute_ap(preds_list, num_gt):
+ """
+ Compute AP given list of (score, is_tp) and total GT count.
+ Uses precision envelope integration.
+ """
+ if num_gt <= 0 or not preds_list:
+ return 0.0
+ preds_sorted = sorted(preds_list, key=lambda x: x[0], reverse=True)
+ tp_cum = 0
+ fp_cum = 0
+ precisions = []
+ recalls = []
+ for score, is_tp in preds_sorted:
+ if is_tp:
+ tp_cum += 1
+ else:
+ fp_cum += 1
+ precision = tp_cum / max(tp_cum + fp_cum, 1)
+ recall = tp_cum / num_gt
+ precisions.append(precision)
+ recalls.append(recall)
+ # Precision envelope
+ for i in range(len(precisions) - 2, -1, -1):
+ if precisions[i] < precisions[i + 1]:
+ precisions[i] = precisions[i + 1]
+ # Integrate AP
+ ap = 0.0
+ prev_recall = 0.0
+ for p, r in zip(precisions, recalls):
+ if r > prev_recall:
+ ap += p * (r - prev_recall)
+ prev_recall = r
+ return ap
+
+
+def _update_experiment_metrics(ground_truth_label, scores_list, ground_truth_boxes=None):
+ """
+ Update running experiment metrics based on a single image's ground truth label and predictions.
+ If ground-truth bounding boxes are provided, compute IoU and mAP.
+ """
+ global experiment_metrics, map_data
+
+ # Normalize inputs
+ gt = None if ground_truth_label is None else str(ground_truth_label)
+ has_gt_object = gt is not None and gt.lower() not in ["empty", "unknown"]
+
+ predictions = scores_list if isinstance(scores_list, list) else []
+ num_predictions = len(predictions)
+ predicted_labels = [str(p.get("label")) for p in predictions if p and p.get("label") is not None]
+
+ # Detection mode vs classification mode
+ detection_mode = isinstance(ground_truth_boxes, list) and len(ground_truth_boxes) > 0
+ increment_true_positive = 0
+ increment_false_negative = 0
+ increment_false_positive = 0
+
+ if detection_mode:
+ gt_boxes = [g for g in ground_truth_boxes if g and g.get("bounding_box") is not None]
+ experiment_metrics["gt_boxes_count"] += len(gt_boxes)
+ _update_map_structures(predictions, gt_boxes)
+ matches, used_preds, used_gts = _greedy_match(predictions, gt_boxes, 0.5)
+ tp_count = 0
+ sum_iou_img = 0.0
+ for pi, gi, iou in matches:
+ if str(predictions[pi].get("label")) == str(gt_boxes[gi].get("label")):
+ tp_count += 1
+ sum_iou_img += iou
+ fp_count = max(num_predictions - len(used_preds), 0)
+ fn_count = max(len(gt_boxes) - len(used_gts), 0)
+ increment_true_positive = tp_count
+ increment_false_positive = fp_count
+ increment_false_negative = fn_count
+ if tp_count > 0:
+ experiment_metrics["sum_iou"] += sum_iou_img
+ experiment_metrics["num_iou_pairs"] += tp_count
+ else:
+ has_correct_prediction = has_gt_object and any(lbl == gt for lbl in predicted_labels)
+ increment_true_positive = 1 if has_correct_prediction else 0
+ increment_false_negative = 1 if has_gt_object and not has_correct_prediction else 0
+ increment_false_positive = max(num_predictions - (1 if has_correct_prediction else 0), 0)
+
+ # Update totals
+ experiment_metrics["total_images"] += 1
+ experiment_metrics["total_predictions"] += num_predictions
+ if detection_mode:
+ experiment_metrics["total_ground_truth_objects"] += len(ground_truth_boxes)
+ else:
+ experiment_metrics["total_ground_truth_objects"] += 1 if has_gt_object else 0
+ experiment_metrics["true_positives"] += increment_true_positive
+ experiment_metrics["false_negatives"] += increment_false_negative
+ experiment_metrics["false_positives"] += increment_false_positive
+
+ # Derived metrics
+ tp = float(experiment_metrics["true_positives"])
+ fp = float(experiment_metrics["false_positives"])
+ fn = float(experiment_metrics["false_negatives"])
+ precision = (tp / (tp + fp)) if (tp + fp) > 0 else 0.0
+ recall = (tp / (tp + fn)) if (tp + fn) > 0 else 0.0
+ f1 = (2.0 * precision * recall / (precision + recall)) if (precision + recall) > 0 else 0.0
+ experiment_metrics["precision"] = precision
+ experiment_metrics["recall"] = recall
+ experiment_metrics["f1_score"] = f1
+
+ # Mean IoU
+ if experiment_metrics["num_iou_pairs"] > 0:
+ experiment_metrics["mean_iou"] = experiment_metrics["sum_iou"] / experiment_metrics["num_iou_pairs"]
+
+ # mAP
+ if experiment_metrics["gt_boxes_count"] > 0:
+ ap50 = _compute_ap(map_data[0.5], experiment_metrics["gt_boxes_count"]) if 0.5 in map_data else 0.0
+ ap_values = [_compute_ap(map_data[t], experiment_metrics["gt_boxes_count"]) for t in map_thresholds]
+ map_50_95 = sum(ap_values) / len(ap_values) if ap_values else 0.0
+ experiment_metrics["map_50"] = ap50
+ experiment_metrics["map_50_95"] = map_50_95
+
+
+# ============================================================================
+# Kafka streaming
+# ============================================================================
+def build_event_payload(uuid):
+ """
+ Build event payload from in-memory entry for Kafka streaming.
+
+ Event structure matches publish_test_event.py SAMPLE_EVENT for Neo4j Kafka Connector:
+ - Identity fields first (device_id, experiment_id, user_id, model_id)
+ - Image metadata (image_count, UUID, image_name, ground_truth)
+ - Timestamps (image_receiving_timestamp, image_scoring_timestamp, image_store_delete_time)
+ - Prediction result (label, probability, image_decision)
+ - Flattened scores
+ - Running experiment metrics
+ """
+ global existing_image_mapping, experiment_metrics
+
+ entry = existing_image_mapping.get(uuid)
+ if not entry:
+ return None
+
+ # Extract fields
+ image_count = entry.get("image_count")
+ image_name = entry.get("image_name")
+ ground_truth = entry.get("ground_truth")
+ image_receiving_timestamp = entry.get("image_receiving_timestamp")
+ image_scoring_timestamp = entry.get("image_scoring_timestamp")
+ image_store_delete_time = entry.get("image_store_delete_time") or entry.get("image_delete_time")
+ image_decision = entry.get("image_decision")
+ model_id = entry.get("model_id") or MODEL_ID
+
+ # Extract scores
+ scores = entry.get("score", [])
+ flattened_scores = json.dumps(scores) if scores else None
+
+ # Extract highest probability label
+ label = None
+ probability = 0.0
+ if scores:
+ highest_score = max(scores, key=lambda x: x.get("probability", 0.0))
+ label = highest_score.get("label")
+ probability = highest_score.get("probability", 0.0)
+
+ # Build event payload - SAME STRUCTURE AS publish_test_event.py SAMPLE_EVENT
+ event = {
+ # Identity fields
+ "domain": CKN_DOMAIN,
+ "device_id": DEVICE_ID,
+ "experiment_id": EXPERIMENT_ID,
+ "user_id": USER_ID,
+ "model_id": model_id,
+ # Image metadata
+ "image_count": image_count,
+ "UUID": uuid,
+ "image_name": image_name,
+ "ground_truth": ground_truth,
+ # Timestamps
+ "image_receiving_timestamp": image_receiving_timestamp,
+ "image_scoring_timestamp": image_scoring_timestamp,
+ "image_store_delete_time": image_store_delete_time,
+ # Prediction result (highest probability label)
+ "label": label,
+ "probability": probability,
+ "image_decision": image_decision,
+ # Flattened scores as JSON string
+ "flattened_scores": flattened_scores,
+ # Running experiment metrics
+ "total_images": experiment_metrics["total_images"],
+ "total_predictions": experiment_metrics["total_predictions"],
+ "total_ground_truth_objects": experiment_metrics["total_ground_truth_objects"],
+ "true_positives": experiment_metrics["true_positives"],
+ "false_positives": experiment_metrics["false_positives"],
+ "false_negatives": experiment_metrics["false_negatives"],
+ "precision": experiment_metrics["precision"],
+ "recall": experiment_metrics["recall"],
+ "f1_score": experiment_metrics["f1_score"],
+ "mean_iou": experiment_metrics["mean_iou"],
+ "map_50": experiment_metrics["map_50"],
+ "map_50_95": experiment_metrics["map_50_95"],
+ }
+
+ return event
+
+
+def stream_event_to_kafka(uuid):
+ """Stream event to Kafka broker."""
+ global kafka_producer, processed_uuids
+
+ if not kafka_producer:
+ logger.debug(f"Kafka producer not initialized, skipping stream for UUID: {uuid}")
+ return
+
+ if uuid in processed_uuids:
+ logger.debug(f"UUID {uuid} already processed, skipping")
+ return
+
+ event_data = build_event_payload(uuid)
+ if not event_data:
+ logger.warning(f"Could not build event payload for UUID: {uuid}")
+ return
+
+ try:
+ # Event already includes device_id, experiment_id, user_id from build_event_payload
+ row_json = json.dumps(event_data, indent=2)
+
+ # Log the event before publishing
+ logger.info("="*60)
+ logger.info(f"KAFKA EVENT - Publishing to topic '{KAFKA_TOPIC}'")
+ logger.info("="*60)
+ logger.info(f"Key: {EXPERIMENT_ID}")
+ logger.info(f"Event payload:")
+ logger.info(row_json)
+ logger.info("="*60)
+
+ # Produce to Kafka wrapped in a Connect schema envelope (use non-indented JSON for actual message)
+ envelope = build_connect_envelope(event_data, EVENT_FIELD_TYPES, EVENT_REQUIRED_FIELDS)
+ kafka_producer.produce(KAFKA_TOPIC, key=EXPERIMENT_ID, value=json.dumps(envelope))
+ kafka_producer.flush()
+
+ processed_uuids.add(uuid)
+ logger.info(f"Successfully streamed event to Kafka for UUID: {uuid}")
+
+ except Exception as e:
+ logger.error(f"Error streaming to Kafka: {e}")
+
+
+# ============================================================================
+# Power summary processing
+# ============================================================================
+def process_power_summary():
+ """Read power summary file and stream to Kafka."""
+ global kafka_producer
+
+ if ENABLE_POWER_MONITORING != 'true':
+ logger.info("Power monitoring not enabled, skipping power summary.")
+ return
+
+ if not kafka_producer:
+ logger.info("Kafka producer not available, skipping power summary streaming.")
+ return
+
+ if not POWER_SUMMARY_FILE:
+ logger.info("Power summary file not configured, skipping.")
+ return
+
+ logger.info(f"Waiting for power summary file: {POWER_SUMMARY_FILE}")
+
+ attempt = 0
+ while attempt < POWER_SUMMARY_MAX_TRIES:
+ if os.path.exists(POWER_SUMMARY_FILE):
+ try:
+ with open(POWER_SUMMARY_FILE, 'r') as f:
+ data = json.load(f)
+
+ power_summary = data.get("plugin power summary report", [])
+
+ # Build flattened event - SAME STRUCTURE AS publish_test_event.py POWER_SUMMARY_EVENT
+ # experiment_id first, then per-plugin fields, then totals
+ flattened_event = {
+ "experiment_id": EXPERIMENT_ID,
+ }
+
+ total_cpu = 0.0
+ total_gpu = 0.0
+
+ for plugin_data in power_summary:
+ plugin_name = plugin_data.get("plugin", "unknown")
+ cpu = plugin_data.get("cpu_power_consumption", 0.0)
+ gpu = plugin_data.get("gpu_power_consumption", 0.0)
+
+ flattened_event[f"{plugin_name}_cpu_power_consumption"] = cpu
+ flattened_event[f"{plugin_name}_gpu_power_consumption"] = gpu
+ total_cpu += cpu
+ total_gpu += gpu
+
+ flattened_event["total_cpu_power_consumption"] = total_cpu
+ flattened_event["total_gpu_power_consumption"] = total_gpu
+
+ # Log the event before publishing
+ power_json = json.dumps(flattened_event, indent=2)
+ logger.info("="*60)
+ logger.info(f"KAFKA POWER EVENT - Publishing to topic '{POWER_SUMMARY_TOPIC}'")
+ logger.info("="*60)
+ logger.info(f"Key: {EXPERIMENT_ID}")
+ logger.info(f"Power summary payload:")
+ logger.info(power_json)
+ logger.info("="*60)
+
+ # Stream to Kafka wrapped in a Connect schema envelope
+ power_envelope = build_connect_envelope(
+ flattened_event, build_power_field_types(flattened_event), {"experiment_id"}
+ )
+ kafka_producer.produce(POWER_SUMMARY_TOPIC, key=EXPERIMENT_ID, value=json.dumps(power_envelope))
+ kafka_producer.flush()
+
+ logger.info("Power summary successfully streamed to Kafka.")
+ return
+
+ except Exception as e:
+ logger.error(f"Error processing power summary: {e}")
+ return
+
+ attempt += 1
+ logger.info(f"Power summary file not found, attempt {attempt}/{POWER_SUMMARY_MAX_TRIES}")
+ time.sleep(POWER_SUMMARY_TIMEOUT)
+
+ logger.warning("Power summary file not found after all attempts.")
+
+
+# ============================================================================
+# In-memory state management (replaces file-based image_mapping_final.json)
+# ============================================================================
+def get_socket():
+ context = zmq.Context()
+ return get_plugin_socket(context, PORT)
+
+
+def compute_total_images_generated():
+ """Reads uuid_image_mapping file to determine how many images were generated."""
+ try:
+ with open(uuid_image_mapping_path, 'r') as f:
+ d = json.load(f)
+ return len(d.keys())
+ except Exception as e:
+ logger.error(f"Error reading uuid_image_mapping: {e}")
+ return -1
+
+
+def compute_total_images_processed():
+ """Count images in memory with image_decision set."""
+ global existing_image_mapping
+ total = 0
+ for _, v in existing_image_mapping.items():
+ if v.get("image_decision"):
+ total += 1
+ return total
+
+
+def update_mapping(uuid, updated_data):
+ """
+ Update the in-memory mapping for a given image UUID.
+ If UUID is new, fetch base info from uuid_image_mapping.json.
+ """
+ global existing_image_mapping, uuids_with_errors
+
+ # If UUID not yet in mapping, fetch from uuid_image_mapping.json
+ if uuid not in existing_image_mapping:
+ logger.debug(f"Fetching {uuid} from {uuid_image_mapping_path}")
+ uuid_image_mapping = {}
+ try:
+ with open(uuid_image_mapping_path, 'r') as f:
+ uuid_image_mapping = json.load(f)
+ # Try to recover any previously failed UUIDs
+ for failed_uuid in list(uuids_with_errors):
+ if failed_uuid in uuid_image_mapping:
+ if failed_uuid in existing_image_mapping:
+ existing_image_mapping[failed_uuid].update(uuid_image_mapping[failed_uuid])
+ else:
+ existing_image_mapping[failed_uuid] = uuid_image_mapping[failed_uuid]
+ uuids_with_errors.remove(failed_uuid)
+ except json.JSONDecodeError as e:
+ logger.error(f"JSON decode error reading uuid_image_mapping for {uuid}: {e}")
+ uuids_with_errors.append(uuid)
+ except FileNotFoundError:
+ logger.error(f"uuid_image_mapping file not found: {uuid_image_mapping_path}")
+
+ # Get base info or create minimal entry
+ existing_image_mapping[uuid] = uuid_image_mapping.get(uuid, {"UUID": uuid})
+
+ # Update with new data
+ for key, value in updated_data.items():
+ existing_image_mapping[uuid][key] = value
+
+ # If this update set image_decision, update metrics and stream to Kafka
+ if "image_decision" in updated_data:
+ entry = existing_image_mapping[uuid]
+ ground_truth = entry.get("ground_truth")
+ scores = entry.get("score", [])
+ ground_truth_boxes = entry.get("ground_truth_boxes") or entry.get("ground_truth_bboxes")
+
+ # Update experiment metrics
+ _update_experiment_metrics(ground_truth, scores, ground_truth_boxes)
+
+ # Stream to Kafka
+ stream_event_to_kafka(uuid)
+
+
+# ============================================================================
+# Main event loop
+# ============================================================================
+def main():
+ """
+ Main loop for CKN plugin.
+ Waits for new messages on ZMQ and processes accordingly:
+ 1. ImageReceived/ImageScored/ImageStored/ImageDeleted: update in-memory mapping
+ 2. PluginTerminating: compute total images, trigger shutdown when done
+ """
+ global received_terminating_signal, total_images_generated, total_images_processed
+
+ # Initialize Kafka if configured
+ init_kafka_producer()
+
+ done = False
+ while not done:
+ socket = get_socket()
+ try:
+ message = get_next_msg(socket)
+ except zmq.error.Again:
+ logger.debug(f"Got zmq.error.Again; waited {SOCKET_TIMEOUT} ms without message")
+ continue
+ except Exception as e:
+ logger.debug(f"Got exception from get_next_msg: {type(e)}: {e}")
+ done = True
+ logger.info("CKN plugin stopping due to timeout limit...")
+ continue
+
+ if not message:
+ logger.info("No message found in get_next_msg")
+ continue
+
+ logger.info("Got a message from the event socket")
+ event = socket_message_to_typed_event(message)
+
+ if isinstance(event, ImageReceivedEvent):
+ uuid = event.ImageUuid().decode('utf-8')
+ timestamp = event.EventCreateTs().decode('utf-8').strip("'")
+ logger.info(f"Image received {uuid} {timestamp}")
+ update_mapping(uuid, {"image_receiving_timestamp": timestamp})
+
+ elif isinstance(event, ImageScoredEvent):
+ uuid = event.ImageUuid().decode('utf-8')
+ scores = []
+ for i in range(event.ScoresLength()):
+ label = event.Scores(i).Label().decode('utf-8')
+ prob = event.Scores(i).Probability()
+ scores.append({"label": label, "probability": prob})
+ timestamp = event.EventCreateTs().decode('utf-8')
+ logger.info(f"Image scored {uuid} {scores} {timestamp}")
+ update_mapping(uuid, {"image_scoring_timestamp": timestamp, "score": scores})
+
+ elif isinstance(event, ImageStoredEvent):
+ uuid = event.ImageUuid().decode('utf-8')
+ timestamp = event.EventCreateTs().decode('utf-8')
+ destination = event.Destination().decode('utf-8')
+ logger.info(f"Image stored {uuid} {timestamp} {destination}")
+ update_mapping(uuid, {"image_store_delete_time": timestamp, "image_decision": destination})
+
+ elif isinstance(event, ImageDeletedEvent):
+ uuid = event.ImageUuid().decode('utf-8')
+ timestamp = event.EventCreateTs().decode('utf-8')
+ logger.info(f"Image deleted {uuid} {timestamp}")
+ update_mapping(uuid, {"image_delete_time": timestamp, "image_decision": "Deleted"})
+
+ elif isinstance(event, PluginTerminatingEvent):
+ plugin_name = event.PluginName().decode('utf-8')
+ if plugin_name == 'ext_image_gen_plugin':
+ logger.info("Received Terminating signal from image generating plugin")
+ received_terminating_signal = True
+ total_images_generated = compute_total_images_generated()
+ logger.info(f"Total images generated: {total_images_generated}")
+
+ # Check if all images processed
+ if received_terminating_signal:
+ total_images_processed = compute_total_images_processed()
+ logger.info(f"CKN plugin has processed: {total_images_processed} out of {total_images_generated}")
+
+ if total_images_generated < 0:
+ total_images_generated = compute_total_images_generated()
+
+ # Shutdown when all images processed
+ if (received_terminating_signal
+ and total_images_generated > 0
+ and total_images_generated == total_images_processed):
+ logger.info("All images processed. Initiating shutdown...")
+
+ # Send PluginTerminate first: the power monitor plugin only writes
+ # power_summary_report.json when it receives this event, so it must
+ # be told to stop before we wait on the summary file.
+ send_terminate_plugin_fb_event(socket, "*", EXPERIMENT_END_SIGNAL)
+ logger.info("Sent PluginTerminate * event")
+
+ # Now wait for the power summary file and stream it to Kafka
+ process_power_summary()
+
+ time.sleep(1)
+ send_quit_command(socket)
+ logger.info("Sent quit command.")
+ sys.exit()
+ else:
+ logger.debug(f"Event: {event}")
+
+
+if __name__ == '__main__':
+ logger.info("CKN plugin starting...")
+ main()
+ logger.info("CKN plugin exiting...")
diff --git a/external_plugins/ckn_plugin/consume_test_event.py b/external_plugins/ckn_plugin/consume_test_event.py
new file mode 100644
index 0000000..0e375a8
--- /dev/null
+++ b/external_plugins/ckn_plugin/consume_test_event.py
@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+"""
+Consume events from the oracle-events Kafka topic.
+"""
+
+import json
+import signal
+import sys
+from confluent_kafka import Consumer, KafkaError, KafkaException
+from confluent_kafka.admin import AdminClient
+
+# Kafka configuration
+KAFKA_BROKER = "cknbroker.pods.icicleai.tapis.io:443"
+KAFKA_TOPIC = "oracle-events"
+KAFKA_SECURITY_PROTOCOL = "SSL"
+CONSUMER_GROUP = "ckn-test-consumer-group"
+
+# Flag to control graceful shutdown
+running = True
+
+
+def signal_handler(sig, frame):
+ """Handle Ctrl+C for graceful shutdown."""
+ global running
+ print("\nShutting down consumer...")
+ running = False
+
+
+def test_connection(kafka_conf):
+ """Test connection to Kafka broker."""
+ print(f"Testing connection to {KAFKA_BROKER}...")
+ try:
+ admin_client = AdminClient(kafka_conf)
+ topics = admin_client.list_topics(timeout=10)
+ print(f"Connected! Available topics: {list(topics.topics.keys())}")
+ return True
+ except Exception as e:
+ print(f"Connection failed: {e}")
+ return False
+
+
+def consume_events():
+ """Consume events from the Kafka topic."""
+ global running
+
+ # Base config for testing connection
+ base_conf = {
+ 'bootstrap.servers': KAFKA_BROKER,
+ 'security.protocol': KAFKA_SECURITY_PROTOCOL,
+ }
+
+ # Test connection first
+ if not test_connection(base_conf):
+ print("Aborting: Could not connect to Kafka broker")
+ return False
+
+ # Consumer config
+ consumer_conf = {
+ 'bootstrap.servers': KAFKA_BROKER,
+ 'security.protocol': KAFKA_SECURITY_PROTOCOL,
+ 'group.id': CONSUMER_GROUP,
+ 'auto.offset.reset': 'earliest', # Start from beginning if no committed offset
+ 'enable.auto.commit': True,
+ 'auto.commit.interval.ms': 5000,
+ }
+
+ print(f"\nCreating Kafka consumer...")
+ print(f"Consumer group: {CONSUMER_GROUP}")
+ print(f"Topic: {KAFKA_TOPIC}")
+ print(f"Starting from: earliest (will read all available messages)")
+
+ consumer = Consumer(**consumer_conf)
+ consumer.subscribe([KAFKA_TOPIC])
+
+ print(f"\n{'='*60}")
+ print(f"Listening for events on topic: {KAFKA_TOPIC}")
+ print(f"Press Ctrl+C to stop")
+ print(f"{'='*60}\n")
+
+ # Set up signal handler for graceful shutdown
+ signal.signal(signal.SIGINT, signal_handler)
+ signal.signal(signal.SIGTERM, signal_handler)
+
+ message_count = 0
+
+ try:
+ while running:
+ # Poll for messages with 1 second timeout
+ msg = consumer.poll(timeout=1.0)
+
+ if msg is None:
+ continue
+
+ if msg.error():
+ if msg.error().code() == KafkaError._PARTITION_EOF:
+ # End of partition, not an error
+ print(f"Reached end of partition {msg.partition()}")
+ continue
+ else:
+ raise KafkaException(msg.error())
+
+ # Process the message
+ message_count += 1
+
+ print(f"\n{'='*60}")
+ print(f"MESSAGE #{message_count}")
+ print(f"{'='*60}")
+ print(f"Topic: {msg.topic()}")
+ print(f"Partition: {msg.partition()}")
+ print(f"Offset: {msg.offset()}")
+ print(f"Key: {msg.key().decode('utf-8') if msg.key() else 'None'}")
+ print(f"Timestamp: {msg.timestamp()}")
+ print(f"{'='*60}")
+
+ # Parse and pretty-print the JSON payload
+ try:
+ value = msg.value().decode('utf-8')
+ event = json.loads(value)
+ print("Event payload:")
+ print(json.dumps(event, indent=2))
+ except json.JSONDecodeError:
+ print(f"Raw value (not JSON): {msg.value()}")
+ except Exception as e:
+ print(f"Error decoding message: {e}")
+
+ print(f"{'='*60}\n")
+
+ except Exception as e:
+ print(f"Error consuming messages: {e}")
+ finally:
+ # Close consumer
+ print(f"\nClosing consumer... Total messages received: {message_count}")
+ consumer.close()
+
+ return True
+
+
+if __name__ == "__main__":
+ consume_events()
diff --git a/external_plugins/ckn_plugin/power_processor.py b/external_plugins/ckn_plugin/power_processor.py
new file mode 100644
index 0000000..b5776b5
--- /dev/null
+++ b/external_plugins/ckn_plugin/power_processor.py
@@ -0,0 +1,125 @@
+"""
+Power Processor - Reads power summary from file and streams to Kafka.
+
+This module can be used standalone or imported by ckn_plugin.
+The main ckn_plugin.py has this logic inlined, but this module
+is kept for compatibility and testing.
+"""
+
+import json
+import os
+import time
+import logging
+
+
+def build_connect_envelope(payload, field_types, required_fields=None):
+ """Wrap a flat dict in a Kafka Connect JSON schema envelope (schema+payload).
+
+ Mirrors the helper in ckn_plugin.py - the JDBC sink connectors run with
+ value.converter.schemas.enable=true, so bare JSON is silently dropped.
+ """
+ required_fields = required_fields or set()
+ fields = [
+ {"field": name, "type": conn_type, "optional": name not in required_fields}
+ for name, conn_type in field_types.items()
+ ]
+ schema = {"type": "struct", "fields": fields, "optional": False}
+ return {"schema": schema, "payload": payload}
+
+
+def build_power_field_types(flattened_event):
+ """Power summary keys are dynamic (per-plugin names), so infer types from the event."""
+ return {
+ name: "string" if name == "experiment_id" else "double"
+ for name in flattened_event
+ }
+
+
+class PowerProcessor:
+ """
+ Processes the power events and sends events to the CKN Broker.
+ """
+ def __init__(self, power_summary_file, kafka_producer, topic, experiment_id, max_attempts=5, timeout=10):
+ self.power_summary_file = power_summary_file
+ self.kafka_producer = kafka_producer
+ self.topic = topic
+ self.experiment_id = experiment_id
+ self.max_attempts = int(max_attempts) if max_attempts is not None else 5
+ self.timeout = int(timeout) if timeout is not None else 10
+
+ def get_power_summary(self):
+ """
+ Reads the power summary from the power summary file.
+ Returns a flattened event dictionary.
+ """
+ with open(self.power_summary_file, 'r') as file:
+ data = json.load(file)
+
+ # Extract the plugin power summary report
+ power_summary = data.get("plugin power summary report", [])
+
+ # Initialize a dictionary for the flattened event
+ flattened_event = {}
+
+ # Initialize total CPU and GPU consumption
+ total_cpu_consumption = 0.0
+ total_gpu_consumption = 0.0
+
+ # Iterate over each plugin's data and add it to the flattened event
+ for plugin_data in power_summary:
+ plugin_name = plugin_data.get("plugin", "unknown")
+
+ # Add plugin's CPU and GPU consumption to the flattened event
+ cpu_consumption = plugin_data.get("cpu_power_consumption", 0.0)
+ gpu_consumption = plugin_data.get("gpu_power_consumption", 0.0)
+
+ flattened_event[f"{plugin_name}_cpu_power_consumption"] = cpu_consumption
+ flattened_event[f"{plugin_name}_gpu_power_consumption"] = gpu_consumption
+
+ # Accumulate total CPU and GPU consumption
+ total_cpu_consumption += cpu_consumption
+ total_gpu_consumption += gpu_consumption
+
+ # Add total CPU and GPU consumption and experiment ID to the flattened event
+ flattened_event["total_cpu_power_consumption"] = total_cpu_consumption
+ flattened_event["total_gpu_power_consumption"] = total_gpu_consumption
+ flattened_event["experiment_id"] = self.experiment_id
+
+ return flattened_event
+
+ def process_summary_events(self):
+ """
+ Waits for the summary to be available and processes it.
+ """
+ attempt = 0
+ while attempt < self.max_attempts:
+ if os.path.exists(self.power_summary_file):
+ logging.info("Reading the power summary file...")
+
+ try:
+ # Read the power summary
+ power_summary = self.get_power_summary()
+ envelope = build_connect_envelope(
+ power_summary, build_power_field_types(power_summary), {"experiment_id"}
+ )
+ power_summary_json = json.dumps(envelope)
+
+ # Send the event to Kafka
+ if self.kafka_producer:
+ self.kafka_producer.produce(self.topic, key=self.experiment_id, value=power_summary_json)
+ self.kafka_producer.flush()
+ logging.info("Power summary streamed to Kafka.")
+ else:
+ logging.info("Kafka producer not available; read power summary but did not stream.")
+ logging.debug(f"Power summary payload: {power_summary_json}")
+ return
+ except Exception as e:
+ logging.error(f"Error processing power summary: {e}")
+ return
+
+ # Increment the attempt count and wait before trying again
+ attempt += 1
+ logging.info(f"Power summary file not found, attempt {attempt}/{self.max_attempts}")
+ time.sleep(self.timeout)
+
+ logging.warning("No power summary file found after all attempts.")
diff --git a/external_plugins/ckn_plugin/publish_test_event.py b/external_plugins/ckn_plugin/publish_test_event.py
new file mode 100644
index 0000000..fa12120
--- /dev/null
+++ b/external_plugins/ckn_plugin/publish_test_event.py
@@ -0,0 +1,333 @@
+#!/usr/bin/env python3
+"""
+Publish test events to Kafka topics:
+- oracle-events: Image processing events
+- cameratraps-power-summary: Power consumption summary
+"""
+
+import json
+import uuid
+import sys
+from confluent_kafka import Producer
+from confluent_kafka.admin import AdminClient
+
+# Kafka configuration
+KAFKA_BROKER = "cknbroker.pods.icicleai.tapis.io:443"
+KAFKA_SECURITY_PROTOCOL = "SSL"
+
+# Topics
+ORACLE_EVENTS_TOPIC = "oracle-events"
+POWER_SUMMARY_TOPIC = "cameratraps-power-summary"
+
+# The JDBC sink connectors run with value.converter.schemas.enable=true, so
+# JsonConverter requires each record to be a {"schema": ..., "payload": ...}
+# envelope rather than bare JSON - mirrors the helper in ckn_plugin.py.
+EVENT_FIELD_TYPES = {
+ "device_id": "string",
+ "experiment_id": "string",
+ "user_id": "string",
+ "model_id": "string",
+ "image_count": "int32",
+ "UUID": "string",
+ "image_name": "string",
+ "ground_truth": "string",
+ "image_receiving_timestamp": "string",
+ "image_scoring_timestamp": "string",
+ "image_store_delete_time": "string",
+ "label": "string",
+ "probability": "double",
+ "image_decision": "string",
+ "flattened_scores": "string",
+ "total_images": "int32",
+ "total_predictions": "int32",
+ "total_ground_truth_objects": "int32",
+ "true_positives": "int32",
+ "false_positives": "int32",
+ "false_negatives": "int32",
+ "precision": "double",
+ "recall": "double",
+ "f1_score": "double",
+ "mean_iou": "double",
+ "map_50": "double",
+ "map_50_95": "double",
+}
+EVENT_REQUIRED_FIELDS = {"UUID", "experiment_id", "user_id"}
+
+
+def build_connect_envelope(payload, field_types, required_fields=None):
+ """Wrap a flat dict in a Kafka Connect JSON schema envelope (schema+payload)."""
+ required_fields = required_fields or set()
+ fields = [
+ {"field": name, "type": conn_type, "optional": name not in required_fields}
+ for name, conn_type in field_types.items()
+ ]
+ schema = {"type": "struct", "fields": fields, "optional": False}
+ return {"schema": schema, "payload": payload}
+
+
+def build_power_field_types(flattened_event):
+ """Power summary keys are dynamic (per-plugin names), so infer types from the event."""
+ return {
+ name: "string" if name == "experiment_id" else "double"
+ for name in flattened_event
+ }
+
+# Generate unique IDs for this run
+EXPERIMENT_ID = str(uuid.uuid4())
+
+# Sample event matching the ckn_plugin event structure for Neo4j Kafka Connector
+# All fields at top level as expected by the Cypher query
+SAMPLE_EVENT = {
+ # Identity fields - must be pre-registered in patradb (users/edge_devices/models)
+ # for the CKN ingest trigger (fn_ingest_camera_trap_event) to accept the row.
+ "device_id": "example_device",
+ "experiment_id": EXPERIMENT_ID,
+ "user_id": "example_user",
+ "model_id": "1",
+
+ # Image metadata
+ "image_count": 1,
+ "UUID": str(uuid.uuid4()),
+ "image_name": "/example_images/blank01.jpeg",
+ "ground_truth": "empty",
+
+ # Timestamps
+ "image_receiving_timestamp": "2026-02-01T03:52:29.593154603+00:00",
+ "image_scoring_timestamp": "2026-02-01T03:52:42.308346",
+ "image_store_delete_time": "2026-02-01T03:52:42.309816675+00:00",
+
+ # Prediction result (highest probability label)
+ "label": "animal",
+ "probability": 0.019999999552965164,
+ "image_decision": "Deleted",
+
+ # Flattened scores as JSON string
+ "flattened_scores": '[{"label": "animal", "probability": 0.019999999552965164}, {"label": "animal", "probability": 0.019999999552965164}, {"label": "person", "probability": 0.019999999552965164}, {"label": "person", "probability": 0.009999999776482582}, {"label": "animal", "probability": 0.009999999776482582}, {"label": "animal", "probability": 0.009999999776482582}, {"label": "person", "probability": 0.009999999776482582}]',
+
+ # Running experiment metrics
+ "total_images": 1,
+ "total_predictions": 7,
+ "total_ground_truth_objects": 0,
+ "true_positives": 0,
+ "false_positives": 7,
+ "false_negatives": 0,
+ "precision": 0.0,
+ "recall": 0.0,
+ "f1_score": 0.0,
+ "mean_iou": None,
+ "map_50": None,
+ "map_50_95": None
+}
+
+# Power summary event for cameratraps-power-summary topic
+# FLATTENED structure as expected by Neo4j Kafka Connector Cypher query
+POWER_SUMMARY_EVENT = {
+ "experiment_id": EXPERIMENT_ID,
+ # Per-plugin power consumption (flattened)
+ "image_generating_plugin_cpu_power_consumption": 2.6314074074074068,
+ "image_generating_plugin_gpu_power_consumption": 0.07603703703703703,
+ "power_monitor_plugin_cpu_power_consumption": 2.5915555555555554,
+ "power_monitor_plugin_gpu_power_consumption": 0.07137037037037038,
+ "image_scoring_plugin_cpu_power_consumption": 2.5690384615384616,
+ "image_scoring_plugin_gpu_power_consumption": 0.08219230769230768,
+ # Totals
+ "total_cpu_power_consumption": 7.792001433501424, # sum of all CPU
+ "total_gpu_power_consumption": 0.22959971509971508 # sum of all GPU
+}
+
+class PowerProcessor:
+ """
+ Processes the power events and sends events to the CKN Broker.
+ """
+ def __init__(self, power_summary_file, kafka_producer, topic, experiment_id, max_attempts=5, timeout=10):
+ self.power_summary_file = power_summary_file
+ self.kafka_producer = kafka_producer
+ self.topic = topic
+ self.experiment_id = experiment_id
+ self.max_attempts = max_attempts
+ self.timeout = timeout
+
+ def get_power_summary(self):
+ """
+ Reads the power summary from the power summary file.
+ :return:
+ """
+ with open(self.power_summary_file, 'r') as file:
+ data = json.load(file)
+
+ # Extract the plugin power summary report
+ power_summary = data["plugin power summary report"]
+
+ # Initialize a dictionary for the flattened event
+ flattened_event = {}
+
+ # Initialize total CPU and GPU consumption
+ total_cpu_consumption = 0.0
+ total_gpu_consumption = 0.0
+
+ # Iterate over each plugin's data and add it to the flattened event
+ for plugin_data in power_summary:
+ plugin_name = plugin_data["plugin"]
+
+ # Add plugin's CPU and GPU consumption to the flattened event
+ cpu_consumption = plugin_data["cpu_power_consumption"]
+ gpu_consumption = plugin_data["gpu_power_consumption"]
+
+ flattened_event[f"{plugin_name}_cpu_power_consumption"] = cpu_consumption
+ flattened_event[f"{plugin_name}_gpu_power_consumption"] = gpu_consumption
+
+ # Accumulate total CPU and GPU consumption
+ total_cpu_consumption += cpu_consumption
+ total_gpu_consumption += gpu_consumption
+
+ # Add total CPU and GPU consumption and experiment ID to the flattened event
+ flattened_event["total_cpu_power_consumption"] = total_cpu_consumption
+ flattened_event["total_gpu_power_consumption"] = total_gpu_consumption
+ flattened_event["experiment_id"] = self.experiment_id
+
+ return flattened_event
+
+ def process_summary_events(self):
+ """
+ Waits for the summary to be available and processes it.
+ :return:
+ """
+ attempt = 0
+ # read the file if it's available. total wait time
+ while attempt < self.max_attempts:
+ if os.path.exists(self.power_summary_file):
+ logging.info("Reading the power summary file...")
+
+ # read the power summary
+ power_summary = self.get_power_summary()
+ power_summary_json = json.dumps(power_summary)
+
+ # send the event to kafka
+ self.kafka_producer.produce(self.topic, key=self.experiment_id, value=power_summary_json)
+ self.kafka_producer.flush()
+ return
+ # Increment the attempt count and wait before trying again
+ attempt += 1
+ time.sleep(self.timeout)
+
+ logging.info("No power summary file found...")
+
+
+def delivery_callback(err, msg):
+ """Callback for message delivery confirmation."""
+ if err:
+ print(f"ERROR: Message delivery failed: {err}")
+ else:
+ print(f"SUCCESS: Message delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")
+
+
+def test_connection(kafka_conf):
+ """Test connection to Kafka broker."""
+ print(f"Testing connection to {KAFKA_BROKER}...")
+ try:
+ admin_client = AdminClient(kafka_conf)
+ topics = admin_client.list_topics(timeout=10)
+ print(f"Connected! Available topics: {list(topics.topics.keys())}")
+ return True
+ except Exception as e:
+ print(f"Connection failed: {e}")
+ return False
+
+
+def publish_event(topic, event, key, envelope):
+ """Publish an event to the specified Kafka topic, wrapped in a Connect schema envelope."""
+ kafka_conf = {
+ 'bootstrap.servers': KAFKA_BROKER,
+ 'security.protocol': KAFKA_SECURITY_PROTOCOL,
+ }
+
+ # Test connection first
+ if not test_connection(kafka_conf):
+ print("Aborting: Could not connect to Kafka broker")
+ return False
+
+ # Create producer
+ print(f"\nCreating Kafka producer...")
+ producer = Producer(**kafka_conf)
+
+ # Serialize event
+ event_json = json.dumps(envelope)
+
+ print(f"\n{'='*60}")
+ print(f"Publishing event to topic: {topic}")
+ print(f"Key: {key}")
+ print(f"{'='*60}")
+ print(f"Event payload:")
+ print(json.dumps(event, indent=2))
+ print(f"{'='*60}\n")
+
+ # Produce message
+ producer.produce(
+ topic,
+ key=key,
+ value=event_json,
+ callback=delivery_callback
+ )
+
+ # Wait for delivery
+ print("Waiting for delivery confirmation...")
+ producer.flush(timeout=30)
+
+ print("\nDone!")
+ return True
+
+
+def publish_oracle_event():
+ """Publish sample oracle event to oracle-events topic."""
+ print("\n" + "="*60)
+ print("PUBLISHING ORACLE EVENT")
+ print("="*60)
+ envelope = build_connect_envelope(SAMPLE_EVENT, EVENT_FIELD_TYPES, EVENT_REQUIRED_FIELDS)
+ return publish_event(ORACLE_EVENTS_TOPIC, SAMPLE_EVENT, EXPERIMENT_ID, envelope)
+
+
+def publish_power_summary():
+ """Publish power summary event to cameratraps-power-summary topic."""
+ print("\n" + "="*60)
+ print("PUBLISHING POWER SUMMARY EVENT")
+ print("="*60)
+ envelope = build_connect_envelope(
+ POWER_SUMMARY_EVENT, build_power_field_types(POWER_SUMMARY_EVENT), {"experiment_id"}
+ )
+ return publish_event(POWER_SUMMARY_TOPIC, POWER_SUMMARY_EVENT, EXPERIMENT_ID, envelope)
+
+
+def print_usage():
+ """Print usage instructions."""
+ print("""
+Usage: python3 publish_test_event.py [option]
+
+Options:
+ oracle - Publish oracle event to 'oracle-events' topic
+ power - Publish power summary to 'cameratraps-power-summary' topic
+ both - Publish both events (default)
+ help - Show this help message
+
+Examples:
+ python3 publish_test_event.py oracle
+ python3 publish_test_event.py power
+ python3 publish_test_event.py both
+""")
+
+
+if __name__ == "__main__":
+ option = sys.argv[1] if len(sys.argv) > 1 else "both"
+
+ if option == "help" or option == "-h" or option == "--help":
+ print_usage()
+ elif option == "oracle":
+ publish_oracle_event()
+ elif option == "power":
+ publish_power_summary()
+ elif option == "both":
+ publish_oracle_event()
+ publish_power_summary()
+ else:
+ print(f"Unknown option: {option}")
+ print_usage()
+ sys.exit(1)
diff --git a/external_plugins/ckn_plugin/run_e2e_test.sh b/external_plugins/ckn_plugin/run_e2e_test.sh
new file mode 100755
index 0000000..b6b81c8
--- /dev/null
+++ b/external_plugins/ckn_plugin/run_e2e_test.sh
@@ -0,0 +1,194 @@
+#!/usr/bin/env bash
+#
+# Runs a full camera-traps experiment end-to-end against the real ckn_plugin
+# in this branch, streaming oracle events + a power summary through cknbroker
+# into patradb. Automates the runbook in PR #65
+# (https://github.com/tapis-project/camera-traps/pull/65).
+#
+# Requirements: Docker + Compose v2 on a Linux host (x86_64 or arm64), ~25 GB
+# free disk, outbound HTTPS. The user/device/model below must already be
+# registered in patradb (users.username / edge_devices.device_id / models.id)
+# or the ingest trigger will reject every event.
+#
+# Usage:
+# ./run_e2e_test.sh # real powerjoular (0 W readings on most VMs)
+# SYNTHETIC_POWER=1 ./run_e2e_test.sh # swap in a fake powerjoular with nonzero wattage
+#
+# Override identities/version if needed:
+# USER_ID=example_user DEVICE_ID=example_device MODEL_ID=2 TRAPS_REL=0.6.0 ./run_e2e_test.sh
+
+set -euo pipefail
+
+TRAPS_REL="${TRAPS_REL:-0.6.0}"
+USER_ID="${USER_ID:-example_user}"
+DEVICE_ID="${DEVICE_ID:-example_device}"
+MODEL_ID="${MODEL_ID:-2}"
+EXPERIMENT_ID="${EXPERIMENT_ID:-$(python3 -c 'import uuid; print(uuid.uuid4())')}"
+SYNTHETIC_POWER="${SYNTHETIC_POWER:-0}"
+WORKDIR="${WORKDIR:-$HOME/ct-e2e}"
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+
+echo "=== camera-traps CKN e2e test ==="
+echo "Experiment ID: $EXPERIMENT_ID"
+echo "Identity: user=$USER_ID device=$DEVICE_ID model=$MODEL_ID"
+echo "Synthetic power: $SYNTHETIC_POWER"
+echo "Work dir: $WORKDIR"
+echo
+
+mkdir -p "$WORKDIR"
+
+echo "--- 1. Building ckn_plugin image from this branch ---"
+docker build -t "tapis/ckn_plugin:${TRAPS_REL}" --build-arg REL="${TRAPS_REL}" \
+ "$REPO_ROOT/external_plugins/ckn_plugin"
+
+echo "--- 2. Building installer image from this branch ---"
+docker build -t tapis/camera-traps-installer:local "$REPO_ROOT/installer"
+
+echo "--- 3. Writing input.yml ---"
+cat > "$WORKDIR/input.yml" </dev/null
+fi
+
+if [ "$SYNTHETIC_POWER" = "1" ]; then
+ echo "--- 5b. Wiring in synthetic power readings (no RAPL on this host) ---"
+ FPJ_DIR="$WORKDIR/fake-powerjoular"
+ mkdir -p "$FPJ_DIR"
+ cat > "$FPJ_DIR/fake_powerjoular.py" <<'PYEOF'
+#!/usr/bin/env python3
+"""
+Test-only stand-in for powerjoular, for hosts without Intel RAPL (cloud VMs).
+
+Accepts the same CLI the camera-traps powerjoular backend uses
+(`-tp -f `) and appends one CSV row per second in the
+5-column format `convert_powerjoular_csv_to_json` expects, with synthetic
+nonzero CPU/GPU wattage. Runs until the backend force-removes the container.
+"""
+import random
+import sys
+import time
+from datetime import datetime
+
+args = sys.argv[1:]
+pid = args[args.index("-tp") + 1]
+out = args[args.index("-f") + 1]
+
+print(f"fake-powerjoular: monitoring pid {pid}, writing {out}", flush=True)
+
+with open(out, "w") as f:
+ f.write("Date,CPU Utilization,Total Power,CPU Power,GPU Power\n")
+
+while True:
+ ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+ util = random.uniform(0.05, 0.95)
+ cpu_w = random.uniform(2.0, 6.0)
+ gpu_w = random.uniform(0.05, 0.3)
+ total_w = cpu_w + gpu_w
+ with open(out, "a") as f:
+ f.write(f"{ts},{util:.14f},{total_w:.14f},{cpu_w:.14f},{gpu_w:.14f}\n")
+ time.sleep(1)
+PYEOF
+ cat > "$FPJ_DIR/Dockerfile" <<'DOCKEOF'
+FROM python:3.12-alpine
+COPY fake_powerjoular.py /fake_powerjoular.py
+ENTRYPOINT ["python", "-u", "/fake_powerjoular.py"]
+DOCKEOF
+
+ # the power backend always `docker pull`s POWER_JOULAR_IMAGE before running,
+ # so a local-only tag needs to come from somewhere pullable: a throwaway
+ # local registry (localhost is exempt from Docker's TLS requirement).
+ docker rm -f e2e-registry >/dev/null 2>&1 || true
+ docker run -d -p 5000:5000 --name e2e-registry registry:2 >/dev/null
+ sleep 2
+ docker build -t localhost:5000/fake-powerjoular:latest "$FPJ_DIR"
+ docker push localhost:5000/fake-powerjoular:latest
+
+ sed -i '/TRAPS_TEST_POWER_FUNCTION=1/a\ - POWER_JOULAR_IMAGE=localhost:5000/fake-powerjoular:latest' \
+ "$WORKDIR/e2e_test/docker-compose.yml"
+fi
+
+echo "--- 6. Running the experiment ---"
+cd "$WORKDIR/e2e_test"
+docker compose up -d
+
+echo "Waiting for ckn_plugin to process all images and exit..."
+while true; do
+ status=$(docker inspect -f '{{.State.Status}}' ckn_plugin 2>/dev/null || echo "gone")
+ if [ "$status" = "exited" ] || [ "$status" = "gone" ]; then
+ break
+ fi
+ sleep 5
+done
+
+echo
+echo "=== ckn_plugin log tail ==="
+docker logs ckn_plugin 2>&1 | tail -40
+
+streamed=$(docker logs ckn_plugin 2>&1 | grep -c "Successfully streamed event" || true)
+power=$(docker logs ckn_plugin 2>&1 | grep -c "Power summary successfully streamed" || true)
+errs=$(docker logs ckn_plugin 2>&1 | grep -ci "error streaming\|failed to connect\|traceback" || true)
+
+echo
+echo "=== Summary ==="
+echo "Oracle events streamed: $streamed"
+echo "Power summary streamed: $power"
+echo "Errors detected in log: $errs"
+
+if [ "$SYNTHETIC_POWER" = "1" ]; then
+ docker rm -f e2e-registry >/dev/null 2>&1 || true
+fi
+
+echo
+echo "--- 7. Teardown ---"
+docker compose down
+
+echo
+echo "=== Verify in patradb ==="
+echo "export PGPASSWORD="
+cat <= num_images_captured:
done = True
- logger.info(f'Received Terminate event * and shutting down detection reporter plugin')
+ logger.info("Initiating shut down for all other plugins...")
+ send_terminate_plugin_fb_event(socket, "*", "6e153711-9823-4ee6-b608-58e2e801db51")
send_quit_command(socket)
if __name__ == '__main__':
diff --git a/external_plugins/image_detecting_plugin/Dockerfile b/external_plugins/image_detecting_plugin/Dockerfile
index 58adea1..ee8adf5 100644
--- a/external_plugins/image_detecting_plugin/Dockerfile
+++ b/external_plugins/image_detecting_plugin/Dockerfile
@@ -4,7 +4,7 @@ FROM tapis/camera_traps_py_3.13:$REL
RUN find / -path /proc -prune -o -path /sys -prune -o -path /dev -prune -o -writable -print0 | xargs -0 -P $(nproc) -I {} chmod 0777 {} || true
# Package dependencies
-RUN apt update && apt install -y motion vim
+RUN apt update && apt install -y motion vim ffmpeg v4l-utils
RUN pip install watchdog
RUN pip install PILLOW
RUN pip install PyYaml
@@ -22,3 +22,5 @@ RUN chmod -R 0777 /image_detecting_plugin.py || true
ENTRYPOINT [ "/entry.sh" ]
+HEALTHCHECK --interval=10s --timeout=5s --start-period=30s --retries=5 \
+ CMD test -f /tmp/ready || exit 1
diff --git a/external_plugins/image_detecting_plugin/image_detecting_plugin.py b/external_plugins/image_detecting_plugin/image_detecting_plugin.py
index 29ab0bc..28dc387 100644
--- a/external_plugins/image_detecting_plugin/image_detecting_plugin.py
+++ b/external_plugins/image_detecting_plugin/image_detecting_plugin.py
@@ -7,14 +7,15 @@
import zmq
import yaml
import logging
-from subprocess import Popen
+import subprocess
+import sys
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from ctevents import ctevents
from pyevents.events import get_plugin_socket, send_quit_command
-from ctevents.ctevents import send_terminate_plugin_fb_event
+from ctevents.ctevents import send_terminate_plugin_fb_event, send_terminating_plugin_fb_event
# Path to a directory that this plugin "watches" for new image files.
# By default, we set this directory to `/var/lib/motion` in the container, assuming
@@ -23,6 +24,7 @@
DATA_MONITORING_PATH = os.environ.get("DATA_MONITORING_PATH", "/var/lib/motion")
MIN_SECONDS_BETWEEN_IMAGES = float(os.environ.get("MIN_SECONDS_BETWEEN_IMAGES", "2.0"))
MODE = os.environ.get("MODE", "demo")
+DEVICE=os.environ.get("DEVICE")
logger = logging.getLogger("Image Generating Plugin")
if log_level == "DEBUG":
@@ -117,10 +119,23 @@ def check_for_connection(self):
self.last_pos = f.tell()
for line in new_lines:
- if 'motion detection Enabled' in line:
+ if 'device_capability' in line:
self.observer.stop()
return
+def monitor_generating_power():
+ """
+ This function is used to initiate the power monitoring event, if the monitoring flag is set.
+ """
+ monitor_flag = os.getenv('MONITOR_POWER')
+ pid = [os.getpid()]
+ monitor_type = [1]
+ monitor_seconds = 0
+ if monitor_flag:
+ ctevents.send_monitor_power_start_fb_event(socket, pid, monitor_type, monitor_seconds)
+ logger.info(f"Monitoring image detecting power")
+
+
class NewFileHandler(FileSystemEventHandler):
"""
Basic watchdog class to detect new files in the configured directory.
@@ -131,6 +146,7 @@ class NewFileHandler(FileSystemEventHandler):
def __init__(self):
super().__init__()
self.last_image_time = 0
+ self.num_images = 0
def extract_timestamp(self, file_path):
basename = os.path.basename(file_path)
@@ -182,6 +198,7 @@ def process_file(self, file_path):
if uuid:
self.last_image_time = current_time
logging.info(f"Generated uuid ({uuid}) and successfully sent new image event for file: {file_path}")
+ self.num_images = self.num_images + 1
except Exception as e:
logging.error(f"Error processing {file_path}: {e}")
@@ -203,6 +220,36 @@ def get_duration():
else:
return video_info['duration']
+def test_camera(v4l2_device=None):
+ if v4l2_device:
+ sample_img = '/tmp/sample.png'
+ try:
+ result = subprocess.run(
+ ['ffmpeg', '-f', 'v4l2', '-i', v4l2_device, '-frames', '1', sample_img],
+ check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+ )
+ logging.info(f'Captured sample image')
+ except subprocess.CalledProcessError as e:
+ global socket
+ logging.error(f'Error: Failed to capture image from {v4l2_device}. {e.stderr.decode()}')
+ logger.info('Sending quit command')
+ send_terminate_plugin_fb_event(socket, "*", "35f20cdd-a404-4436-8df9-d80a9de91147")
+ send_quit_command(socket)
+ sys.exit()
+
+def update_video_info(num_images):
+ video_info_file = os.environ.get('TRAPS_VIDEO_INFO_PATH', '/video_info.yaml')
+ video_info = {}
+ if os.path.exists(video_info_file):
+ try:
+ with open(video_info_file, 'r') as f:
+ video_info = yaml.safe_load(f)
+ except Exception as e:
+ logging.error(f'Error processing {video_info_file}: {e}')
+ video_info['num_images'] = num_images
+ with open(video_info_file, 'w') as f:
+ yaml.dump(video_info, f)
+
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(message)s',
@@ -215,9 +262,11 @@ def get_duration():
socket = get_socket()
logging.info(f"Image Detecting Plugin starting, monitoring path: {path}")
+ monitor_generating_power()
+
# Startup motion
duration = get_duration()
- motion_proc = Popen(['motion'])
+ motion_proc = subprocess.Popen(['motion'])
# make sure motion is connected to camera
log_observer = Observer()
@@ -226,6 +275,8 @@ def get_duration():
log_observer.start()
log_observer.join()
logger.info('motion has connected to camera')
+ with open('/tmp/ready', 'w') as f:
+ f.write('Application is ready\n')
# instantiate and start the event handler
event_handler = NewFileHandler()
@@ -247,6 +298,7 @@ def get_duration():
observer.stop()
observer.join()
motion_proc.kill()
- logger.info('Sending quit command')
- send_terminate_plugin_fb_event(socket, "*", "35f20cdd-a404-4436-8df9-d80a9de91147")
+ update_video_info(event_handler.num_images)
+ send_terminating_plugin_fb_event(socket,"ext_image_detecting_plugin","35f20cdd-a404-4436-8df9-d80a9de91147")
send_quit_command(socket)
+ logger.info("Image Detecting Plugin shutting down...")
diff --git a/external_plugins/image_generating_plugin/image_generating_plugin.py b/external_plugins/image_generating_plugin/image_generating_plugin.py
index 268b605..2852a30 100644
--- a/external_plugins/image_generating_plugin/image_generating_plugin.py
+++ b/external_plugins/image_generating_plugin/image_generating_plugin.py
@@ -60,6 +60,21 @@ def load_ground_truth():
"""
logger.info("Retrieving the ground truth file in image generating plugin")
ground_truth_data = {}
+ def row_update(row: dict, ground_truth_data: dict):
+ image_name = row['image_name']
+ label = row['ground_truth']
+ try:
+ xc = float(row.get('bbox:x', None))
+ yc = float(row.get('bbox:y', None))
+ width = float(row.get('bbox:w', None))
+ height = float(row.get('bbox:h', None))
+ bbox = [{'label': label, 'bounding_box': [xc, yc, width, height]}]
+ except (TypeError, ValueError):
+ bbox = None
+ if bbox:
+ ground_truth_data[image_name] = {'label': label, 'bbox': bbox}
+ else:
+ ground_truth_data[image_name] = {'label': label}
try:
if use_ground_truth_url:
logger.info(f"Retrieving custom ground truth file: {ground_truth_url}")
@@ -68,14 +83,12 @@ def load_ground_truth():
csv_content = response.content.decode('utf-8').splitlines()
reader = csv.DictReader(csv_content)
for row in reader:
- image_name = row['image_name']
- ground_truth_data[image_name] = row['ground_truth']
+ row_update(row, ground_truth_data)
else:
with open(ground_truth_file, mode='r') as ground_truth:
reader = csv.DictReader(ground_truth)
for row in reader:
- image_name = row['image_name']
- ground_truth_data[image_name] = row['ground_truth']
+ row_update(row, ground_truth_data)
except FileNotFoundError:
logger.error(f"File not found: {ground_truth_file}")
except KeyError as e:
@@ -85,7 +98,7 @@ def load_ground_truth():
return ground_truth_data
-def oracle_monitoring_info(track_image_count, image_uuid, image_name, ground_truth):
+def ckn_monitoring_info(track_image_count, image_uuid, image_name, ground_truth):
"""
This function is called by get_binary for each image processed in the main loop.It
creates and/or updates the uuid_image_mapping.json file that stores information about the image file.
@@ -100,19 +113,22 @@ def oracle_monitoring_info(track_image_count, image_uuid, image_name, ground_tru
OUTPUT_DIR = os.environ.get('TRAPS_MAPPING_OUTPUT_PATH', "/output/")
# look up the image in the ground_truth dictionary
- ground_truth_info = ground_truth.get(image_name, 'unavailable')
+ ground_truth_info = ground_truth.get(image_name, {})
# A value of 'unavailable' is unexpected, as it means there is a missing entry in the dictionary,
# but we do not let that stop the entire program, we just log it as an error:
- if ground_truth_info == 'unavailable':
+ gt_label = ground_truth_info.get('label', 'unavailable')
+ if gt_label == 'unavailable':
logger.error(f"ERROR: Could not determine ground truth value for image {image_name}; no entry in ground_truth dict.")
+ gt_bbox = ground_truth_info.get('bbox')
# build the mapping dictionary for this image
image_mapping_dict = {
"image_count": track_image_count,
"UUID": image_uuid,
"image_name": image_name,
- "ground_truth": ground_truth_info,
+ "ground_truth": gt_label,
"model_id": model_variant,
+ **({"ground_truth_boxes": gt_bbox} if gt_bbox is not None else {}),
}
# Read the uuid image mapping file to build the current map:
@@ -143,7 +159,7 @@ def get_binary(file_name, binary_img, img_format, track_image_count, total_image
logger.info(f"Sending new image with the following data: image:{file_name}; uuid:{image_uuid}; format: {img_format}; type(format): {type(img_format)}")
try:
# first update the uuid_image_mapping.json file
- oracle_monitoring_info(track_image_count, image_uuid, file_name, ground_truth)
+ ckn_monitoring_info(track_image_count, image_uuid, file_name, ground_truth)
# then send the new image event; this should ensure that the mapping file always contains the
# the image prior to a subsequent plugin processing it.
diff --git a/external_plugins/image_scoring_plugin/image_scoring_plugin_server.py b/external_plugins/image_scoring_plugin/image_scoring_plugin_server.py
index 25ad6e5..ca9319e 100644
--- a/external_plugins/image_scoring_plugin/image_scoring_plugin_server.py
+++ b/external_plugins/image_scoring_plugin/image_scoring_plugin_server.py
@@ -34,6 +34,7 @@
DEFAULT_BOX_EXPANSION = 0
CROP_IMAGE = (os.getenv('CROP_IMAGE') == 'true')
DETECTIONS = (os.getenv('DETECTIONS') == 'true')
+EXPANDED_METRICS = (os.getenv('EXPANDED_METRICS') == 'true')
# Whether to force image resizing to a (square) integer size (not recommended to change this)
# None means no resizing.
IMAGE_SIZE = None
@@ -130,17 +131,23 @@ def main():
scores = []
if not results:
- scores.append({"image_uuid": image_uuid, "label": "empty", "probability": 0.0})
+ if EXPANDED_METRICS:
+ scores.append({"image_uuid": image_uuid, "label": "empty", "probability": 0.0, 'bbox': []})
+ else:
+ scores.append({"image_uuid": image_uuid, "label": "empty", "probability": 0.0})
else:
for r in results:
# Each score object should have the format:
# {"image_uuid": image_uuid, "label": "animal", "probability": 0.85}
# Each result returned from detector is a dictionary with `category` and `conf`
# If an image contains multiple detection, we need to append muplitple label and probability for each image.
- scores.append({"image_uuid": image_uuid, "label": r['category'], "probability": r['conf']})
- logger.info(f"Sending image scored event with the following scores: {scores}")
+ if EXPANDED_METRICS:
+ scores.append({"image_uuid": image_uuid, "label": label, "probability": r['conf'], 'bbox': r['bbox']})
+ else:
+ scores.append({"image_uuid": image_uuid, "label": label, "probability": r['conf']})
+ logger.info(f"Sending image scored event with the following scores: {scores}")
send_image_scored_fb_event(socket, image_uuid, image_format, scores)
- logger.info(f"Image Scoring Plugin processing for message {total_messages} complete.")
+ logger.info(f"Image Scoring Plugin processing for message {total_messages} complete.")
diff --git a/external_plugins/oracle_plugin/oracle_plugin.py b/external_plugins/oracle_plugin/oracle_plugin.py
index 0948c14..2d1cc12 100644
--- a/external_plugins/oracle_plugin/oracle_plugin.py
+++ b/external_plugins/oracle_plugin/oracle_plugin.py
@@ -233,7 +233,12 @@ def main():
for i in range(event.ScoresLength()):
label = event.Scores(i).Label().decode('utf-8')
prob = event.Scores(i).Probability()
- scores.append({"label": label, "probability": prob})
+ bbox = event.Scores(i).Bbox()
+ if bbox:
+ bbox_list = [bbox.XCenter(), bbox.YCenter(), bbox.Width(), bbox.Height()]
+ scores.append({"label": label, "probability": prob, "bounding_box": bbox_list})
+ else:
+ scores.append({"label": label, "probability": prob})
timestamp = event.EventCreateTs().decode('utf-8')
logger.info(f"Inside scoring {uuid} {scores} {timestamp}")
update_json(uuid, {"image_scoring_timestamp": timestamp, "score" : scores})
diff --git a/external_plugins/power_measuring_plugin/Dockerfile-scaphandre b/external_plugins/power_measuring_plugin/Dockerfile-scaphandre
index 62b46b9..467d65e 100644
--- a/external_plugins/power_measuring_plugin/Dockerfile-scaphandre
+++ b/external_plugins/power_measuring_plugin/Dockerfile-scaphandre
@@ -1,4 +1,4 @@
ARG REL=use-build-arg
FROM debian:bookworm-slim
-COPY --from=tapis/power_measuring_plugin_py:0.4.0 /scaphandre /scaphandre
+COPY --from=tapis/power_measuring_plugin_py:0.4.0 /scaphandre/target/release/scaphandre /scaphandre/target/release/scaphandre
diff --git a/external_plugins/power_measuring_plugin/power_measuring_plugin.py b/external_plugins/power_measuring_plugin/power_measuring_plugin.py
index bde0a7c..eefa1ff 100644
--- a/external_plugins/power_measuring_plugin/power_measuring_plugin.py
+++ b/external_plugins/power_measuring_plugin/power_measuring_plugin.py
@@ -369,8 +369,14 @@ def get_pids_meta(pids, types):
name = "power_monitor_plugin"
elif "image_scoring_plugin.py" in command_line:
name = "image_scoring_plugin"
- elif "oracle_plugin.py" in command_line:
- name = "oracle_plugin"
+ elif "ckn_plugin.py" in command_line:
+ name = "ckn_plugin"
+ elif "image_detecting_plugin.py" in command_line:
+ name = "image_detecting_plugin"
+ elif "video_generating_plugin.py" in command_line:
+ name = "video_generating_plugin"
+ elif "detection_reporter.py" in command_line:
+ name = "detection_reporter_plugin"
logger.debug(f"Found proc for pid {pid}; name: {name}; cmdline: {command_line}")
procs["name"].append(name)
diff --git a/external_plugins/power_measuring_plugin/validate_schemas.py b/external_plugins/power_measuring_plugin/validate_schemas.py
index 69840ca..253072e 100644
--- a/external_plugins/power_measuring_plugin/validate_schemas.py
+++ b/external_plugins/power_measuring_plugin/validate_schemas.py
@@ -21,7 +21,10 @@ def validate_metadata_schema(metadata_file):
"image_scoring_plugin",
"image_generating_plugin",
"power_monitor_plugin",
- "oracle_plugin",
+ "ckn_plugin",
+ "image_detecting_plugin",
+ "video_generating_plugin",
+ "detection_reporter_plugin",
"engine"
]
},
@@ -150,4 +153,4 @@ def validate_log_schema(log_file):
validate(instance=log_file, schema=log_schema_object)
except Exception as e:
print("Log validation failed:", e)
- exit()
\ No newline at end of file
+ exit()
diff --git a/external_plugins/video_generating_plugin/video_generating_plugin.py b/external_plugins/video_generating_plugin/video_generating_plugin.py
index 516e31a..4be4c42 100644
--- a/external_plugins/video_generating_plugin/video_generating_plugin.py
+++ b/external_plugins/video_generating_plugin/video_generating_plugin.py
@@ -86,7 +86,7 @@ def monitor_generating_power():
monitor_seconds = 0
if monitor_flag:
send_monitor_power_start_fb_event(socket, pid, monitor_type, monitor_seconds)
- logger.info(f"Monitoring image generating power")
+ logger.info(f"Monitoring video generating power")
def is_v4l2loopback_available():
out = run(['v4l2-ctl', '-d', device, '--all'], capture_output=True)
@@ -112,6 +112,7 @@ def process_video(input_video_path, ground_truth):
def main():
global socket
+ socket = get_socket()
ground_truth = load_ground_truth()
stream_proc = None
if mode == 'device':
@@ -119,7 +120,6 @@ def main():
stream_proc = process_video(input_video_path, ground_truth)
done = False
while not done:
- socket = get_socket()
try:
message = get_next_msg(socket)
except zmq.error.Again:
diff --git a/installer/README.md b/installer/README.md
index 84e26e7..ff4c019 100644
--- a/installer/README.md
+++ b/installer/README.md
@@ -206,7 +206,7 @@ users a pre-bundled set of example image. See below:
* Example: false
-* `deploy_oracle`: Whether to deploy the oracle plugin. Default: true.
+* `deploy_ckn`: Whether to deploy the CKN plugin. Default: true.
* Example: false
@@ -224,7 +224,7 @@ This table shows which plugins are enabled and disabled by default and for each
| Image generating | Y | N | Y | N |
| Video generating | N | N | N | Y |
| Image detecting | N | Y | N | Y |
-| Oracle | Y | N | Y | N |
+| CKN | Y | N | Y | N |
| CKN | Y | N | Y | N |
| CKN MQTT | N | Y | N | N |
| Power monitoring | Y | N | Y | Y |
@@ -322,10 +322,10 @@ This is a complete list of all possible configurations.
* Example: 0.4
-* `oracle_plugin_output_dir`: Host directory within the
- host_output_dir where the oracle_plugin's outputs will be written. (Relative to `host_output_dir`)
+* `ckn_plugin_output_dir`: Host directory within the
+ host_output_dir where the ckn_plugin's outputs will be written. (Relative to `host_output_dir`)
- * Example: oracle_output_dir
+ * Example: ckn_output_dir
* `image_generating_log_level`: Log level for image
generating plugin.
@@ -425,15 +425,15 @@ This is a complete list of all possible configurations.
* `image_detecting_log_level`: Log level for image detecting plugin
* Example: DEBUG, INFO
-* `oracle_plugin_image`: The image to use for the oracle plugin, not including the tag. Default: tapis/oracle_plugin.
+* `ckn_plugin_image`: The image to use for the CKN plugin, not including the tag. Default: tapis/ckn_plugin.
- * Example: tapis/oracle_plugin
+ * Example: tapis/ckn_plugin
-* `oracle_plugin_output_dir`: Host directory within the host_output_dir where
- the output of the oracle plugin will be written. (Relative to
+* `ckn_plugin_output_dir`: Host directory within the host_output_dir where
+ the output of the CKN plugin will be written. (Relative to
`host_output_dir`)
- * Example: oracle_output_dir
+ * Example: ckn_output_dir
* `detection_reporter_plugin_image`: The image to use for the detection reporter plugin, not including the tag.
diff --git a/installer/compile.py b/installer/compile.py
index 9e11383..28a2aad 100644
--- a/installer/compile.py
+++ b/installer/compile.py
@@ -85,15 +85,17 @@ def get_vars(input_data, default_data):
'deploy_ckn': False,
'deploy_ckn_mqtt': True,
'deploy_power_monitoring': False,
- 'deploy_oracle': False,
+ 'deploy_ckn': False,
'motion_video_device': '/dev/video0',
+ 'motion_video_type': 'device',
+ 'use_example_video': False,
'inference_server': True}
simulation_defaults = {'deploy_image_generating': True,
'deploy_image_detecting': False,
'deploy_reporter': False,
'deploy_ckn': True,
'deploy_ckn_mqtt': False,
- 'deploy_oracle': True,
+ 'deploy_ckn': True,
'expanded_metrics': True,
'inference_server': False}
video_simulation_defaults = {'deploy_image_generating': False,
@@ -102,7 +104,7 @@ def get_vars(input_data, default_data):
'deploy_reporter': True,
'deploy_ckn': False,
'deploy_ckn_mqtt': False,
- 'deploy_oracle': False,
+ 'deploy_ckn': False,
'use_bundled_example_images': False,
'inference_server': False}
@@ -158,6 +160,7 @@ def get_vars(input_data, default_data):
vars['model_id'] = '41d3ed40-b836-4a62-b3fb-67cee79f33d9-model'
if vars.get('mode') == 'video_simulation':
+ vars['fake_stream'] = True
# for video simulations, determine if motion is using device or netcam
if vars.get('motion_video_device'):
vars['motion_video_type'] = 'device'
@@ -372,9 +375,9 @@ def generate_additional_directories(vars, full_install_dir):
if not os.path.exists(images_output_dir):
os.makedirs(images_output_dir)
- oracle_plugin_output_dir = os.path.join(full_install_dir, vars["oracle_plugin_output_dir"])
- if not os.path.exists(oracle_plugin_output_dir):
- os.makedirs(oracle_plugin_output_dir)
+ ckn_plugin_output_dir = os.path.join(full_install_dir, vars["ckn_plugin_output_dir"])
+ if not os.path.exists(ckn_plugin_output_dir):
+ os.makedirs(ckn_plugin_output_dir)
power_output_dir = os.path.join(full_install_dir, vars["power_output_dir"])
if not os.path.exists(power_output_dir):
diff --git a/installer/defaults.yml b/installer/defaults.yml
index f7e7e5e..7882123 100644
--- a/installer/defaults.yml
+++ b/installer/defaults.yml
@@ -34,6 +34,7 @@ motion_threshold: 100
motion_width: 640
motion_height: 480
image_detecting_log_level: DEBUG
+image_detecting_monitor_power: true
# video generating plugin
deploy_video_generating: false
@@ -41,6 +42,7 @@ video_generating_plugin_image: tapis/video_generating_plugin
video_generating_log_level: DEBUG
video_output_dir: video_output_dir
use_example_video: true
+video_generating_monitor_power: true
source_video_url:
local_video_path:
@@ -72,10 +74,10 @@ power_output_dir: power_output_dir
power_monitor_mount_docker_socket: false
-# Oracle Plugin
-deploy_oracle: true
-oracle_plugin_image: tapis/oracle_plugin
-oracle_plugin_output_dir: oracle_output_dir
+# CKN Plugin
+deploy_ckn: true
+ckn_plugin_image: tapis/ckn_plugin
+ckn_plugin_output_dir: ckn_output_dir
device_id: AAAAAAAAAAAAAAAAAAAA
# Detection Reporter Plugin
@@ -85,12 +87,13 @@ detection_reporter_plugin_output_dir: detection_output_dir
detected_events_file: detections.csv
detection_thresholds:
animal: 0.5
+detection_reporter_monitor_power: true
# CKN Daemon -- for Monitoring Plane Integration
deploy_ckn: true
ckn_daemon_tag: latest
-ckn_kafka_broker_address: https://cknbroker.pods.icicleai.tapis.io
-ckn_kafka_broker_port: 9092
+ckn_kafka_broker_address: cknbroker.pods.icicleai.tapis.io
+ckn_kafka_broker_port: 443
ckn_enable_power_monitoring: true
experiment_id:
user_id:
diff --git a/installer/ground_truth.csv b/installer/ground_truth.csv
index d5c7f64..0bd9362 100644
--- a/installer/ground_truth.csv
+++ b/installer/ground_truth.csv
@@ -1,13 +1,13 @@
-ground_truth,image_name
-animal,/example_images/labrador-pup.jpg
-animal,/example_images/baby-red-fox.jpg
-animal,/example_images/panda-cub.jpg
-animal,/example_images/cheetah-baby.jpg
-animal,/example_images/golden-pup.jpg
-animal,/example_images/german-shep-pup.jpg
+ground_truth,image_name,bbox:x,bbox:y,bbox:w,bbox:h
+animal,/example_images/labrador-pup.jpg,573.5,565.0,813.0,740.0
+animal,/example_images/baby-red-fox.jpg,385.0,295.5,574.0,571.0
+animal,/example_images/panda-cub.jpg,168.0,192.5,302.0,259.0
+animal,/example_images/cheetah-baby.jpg,422.5,268.5,845.0,533.0
+animal,/example_images/golden-pup.jpg,682.5,647.0,947.0,1052.0
+animal,/example_images/german-shep-pup.jpg,591.0,633.5,896.0,1267.0
empty,/example_images/APL_Keyset_800x290.jpg
empty,/example_images/blank02.jpeg
empty,/example_images/blank01.jpeg
empty,/example_images/blank04.jpeg
empty,/example_images/blank03.jpg
-empty,/example_images/blank05.jpeg
\ No newline at end of file
+empty,/example_images/blank05.jpeg
diff --git a/installer/templates/config/motion.conf b/installer/templates/config/motion.conf
index d8a2347..b2595e5 100644
--- a/installer/templates/config/motion.conf
+++ b/installer/templates/config/motion.conf
@@ -162,7 +162,7 @@ webcontrol_parms 0
stream_port 8081
# Restrict stream connections to the localhost.
-stream_localhost on
+stream_localhost off
##############################################################
# Camera config files - One for each camera.
diff --git a/installer/templates/config/traps.toml b/installer/templates/config/traps.toml
index 7c8cff3..1a4b556 100644
--- a/installer/templates/config/traps.toml
+++ b/installer/templates/config/traps.toml
@@ -70,9 +70,9 @@ subscriptions = [
"PluginTerminateEvent",
]
{% endif %}
-{% if deploy_oracle %}
+{% if deploy_ckn %}
[[plugins.external]]
-plugin_name = "ext_oracle_monitor_plugin"
+plugin_name = "ext_ckn_plugin"
id = "6e153711-9823-4ee6-b608-58e2e801db51"
external_port = 6011
subscriptions = [
@@ -95,5 +95,6 @@ subscriptions = [
"ImageDeletedEvent",
"ImageReceivedEvent",
"PluginTerminateEvent",
+ "PluginTerminatingEvent",
]
{% endif %}
diff --git a/installer/templates/docker-compose.yml b/installer/templates/docker-compose.yml
index e232558..a305b69 100644
--- a/installer/templates/docker-compose.yml
+++ b/installer/templates/docker-compose.yml
@@ -91,9 +91,9 @@ services:
# NOTE: this is NOT the shared images directory!!
- {{ source_image_dir }}:/example_images:ro
{% endif %}
- {% if deploy_oracle %}
- # mount the directory shared with the oracle plugin
- - {{ host_output_dir }}/{{ oracle_plugin_output_dir }}:/output
+ {% if deploy_ckn %}
+ # mount the directory shared with the CKN plugin
+ - {{ host_output_dir }}/{{ ckn_plugin_output_dir }}:/output
{% endif %}
{% endif %}
@@ -140,10 +140,16 @@ services:
- IMAGE_DETECTING_PLUGIN_PORT=6002
- MIN_SECONDS_BETWEEN_IMAGES={{ min_seconds_between_images }}
- IMAGE_DETECTING_LOG_LEVEL={{ image_detecting_log_level }}
- {% if motion_video_type == 'device' or motion_video_type == 'file' %}
+ {% if fake_stream %}
- MODE=simulation
- TRAPS_VIDEO_INFO_PATH=/video_info/video_info.yaml
{% endif %}
+ {% if motion_video_device %}
+ - DEVICE={{ motion_video_device }}
+ {% endif %}
+ {% if image_detecting_monitor_power %}
+ - MONITOR_POWER=true
+ {% endif %}
volumes:
- {{ host_config_dir }}/motion.conf:/etc/motion/motion.conf:ro
{% if motion_video_type == 'file' %}
@@ -157,6 +163,8 @@ services:
{% endif %}
depends_on:
- engine
+ ports:
+ - "8081:8081"
{% endif %}
imageScoringPlugin:
@@ -319,18 +327,27 @@ services:
{% endif %}
-{% if deploy_oracle %}
- oraclePlugin:
- container_name: oracle_monitor
- image: {{ oracle_plugin_image }}:{{ ct_version }}
+{% if deploy_ckn %}
+ cknPlugin:
+ container_name: ckn_plugin
+ image: {{ ckn_plugin_image }}:{{ ct_version }}
{% if run_containers_as_user %}
user: "{{ uid }}:{{ gid }}"
{% endif %}
environment:
- - ORACLE_PLUGIN_PORT=6011
- - TRAPS_ORACLE_OUTPUT_PATH=/output
+ - CKN_PLUGIN_PORT=6011
+ - TRAPS_CKN_OUTPUT_PATH=/output
- DEVICE_ID={{ device_id }}
- MODEL_ID={{ model_id }}
+ - CKN_KAFKA_BROKER={{ ckn_kafka_broker_address }}:{{ ckn_kafka_broker_port }}
+ - CKN_KAFKA_TOPIC=oracle-events
+ - CAMERA_TRAPS_DEVICE_ID={{ device_id }}
+ - EXPERIMENT_ID={{ experiment_id }}
+ - USER_ID={{ user_id }}
+ - ENABLE_POWER_MONITORING={{ ckn_enable_power_monitoring }}
+ - POWER_SUMMARY_FILE=/power_logs/power_summary_report.json
+ - POWER_SUMMARY_TOPIC=cameratraps-power-summary
+ - TRAPS_VIDEO_INFO_PATH=/video_info/video_info.yaml
{% if deploy_video_generating %}
- MODE=video
{% else %}
@@ -353,7 +370,9 @@ services:
volumes:
- {{ host_config_dir }}/traps.toml:/traps.toml:ro
- - {{ host_output_dir }}/{{ oracle_plugin_output_dir }}:/output
+ - {{ host_output_dir }}/{{ ckn_plugin_output_dir }}:/output
+ - {{ host_output_dir }}/{{ power_output_dir }}:/power_logs
+ - {{ host_output_dir }}/{{ video_output_dir }}:/video_info
{% endif %}
{% if deploy_reporter %}
@@ -368,6 +387,10 @@ services:
- TRAPS_DETECTION_REPORTER_OUTPUT_PATH=/output
- TRAPS_DETECTED_EVENTS_FILE={{ detected_events_file }}
- TRAPS_DETECTION_FILE=/traps-detection.toml
+ - TRAPS_VIDEO_INFO_FILE=/video_info/video_info.yaml
+ {% if detection_reporter_monitor_power %}
+ - MONITOR_POWER=true
+ {% endif %}
networks:
- cameratraps
{% if use_host_pid %}
@@ -387,37 +410,10 @@ services:
- {{ host_config_dir }}/traps.toml:/traps.toml:ro
- {{ host_config_dir }}/traps-detection.toml:/traps-detection.toml:ro
- {{ host_output_dir }}/{{ detection_reporter_plugin_output_dir }}:/output
+ - {{ host_output_dir }}/{{ video_output_dir }}:/video_info
{% endif %}
-{% if deploy_ckn %}
- cknDaemon:
- container_name: ckn_daemon
- image: iud2i/ckn-daemon-cameratraps:{{ ckn_daemon_tag }}
- environment:
- - ORACLE_CSV_PATH=/logs/image_mapping_final.json
- - CKN_LOG_FILE=/logs/ckn.log
- - CKN_KAFKA_BROKER={{ ckn_kafka_broker_address }}:{{ ckn_kafka_broker_port }}
- - CKN_KAFKA_TOPIC=oracle-events
- - CAMERA_TRAPS_DEVICE_ID={{ device_id }}
- - EXPERIMENT_ID={{ experiment_id }}
- - USER_ID={{ user_id }}
- - POWER_SUMMARY_FILE=/power_logs/power_summary_report.json
- - POWER_SUMMARY_TOPIC=cameratraps-power-summary
- - ENABLE_POWER_MONITORING={{ ckn_enable_power_monitoring }}
- - SYSTEM_MEASUREMENTS_FILE=/power_logs/system_measurements.csv
- networks:
- - cameratraps
-
- pid: host
-
- depends_on:
- - oraclePlugin
-
- volumes:
- - {{ host_output_dir }}/{{ oracle_plugin_output_dir }}:/logs
- - {{ host_output_dir }}/{{ power_output_dir }}:/power_logs
-{% endif %}
{% if inference_server %}
MDServer:
diff --git a/releases/0.3.3/docker-compose.yml b/releases/0.3.3/docker-compose.yml
new file mode 100644
index 0000000..dfb90a3
--- /dev/null
+++ b/releases/0.3.3/docker-compose.yml
@@ -0,0 +1,149 @@
+version: "3.0"
+
+networks:
+ cameratraps:
+ driver: bridge
+
+services:
+ # the name `engine` is important here; in general, the sevice name is addressable by other conatiners on the same
+ # docker network. The default "hostname" used by the python plugin library (pyevents) is "engine"
+ engine:
+ container_name: engine
+ image: tapis/camera_traps_engine:0.3.3
+ networks:
+ - cameratraps
+ environment:
+ - TRAPS_CONFIG_FILE=/traps.toml
+ volumes:
+ # mount the traps.toml in the current working directory.
+ - ./config/traps.toml:/traps.toml:ro
+ # mount the image output directory from the host to the directory specified in traps.toml
+ # Docker compose hijacks $HOME so we use a workaround. If the source directory doesn't
+ # exist it will be created with root ownership.
+ - icicle:/root/camera-traps/images # Using `icicle` volume for images
+# - ./images_output_dir:/root/camera-traps/images
+ # mount the log4rs configuration file over the baked into the image. Comment out
+ # this mount if you want to use the image's default logging configuration.
+ - ../../resources/log4rs.yml:/resources/log4rs.yml
+
+ imageGeneratingPlugin:
+ container_name: image_generating
+ image: tapis/image_generating_plugin_py:0.3.3
+ networks:
+ - cameratraps
+ pid: host
+ depends_on:
+ - engine
+ environment:
+ - IMAGE_GENERATING_PLUGIN_PORT=6000
+ - POWER_MEASURING_PLUGIN_PORT=6010
+ - TRAPS_POWER_LOG_PATH=/logs
+ - TRAPS_TEST_POWER_FUNCTION=0
+ - MONITOR_POWER=true
+ volumes:
+ # mount the traps.toml in the current working directory.
+ - ./config/traps.toml:/traps.toml:ro
+ # mount the example images directory; this is the source of the images used for
+ # generating NewImage events. The path `/example_images` is the default path
+ # where the plugin looks for images. This can be changed by providing a different
+ # configuration file.
+ # NOTE: this is NOT the shared images directory!!
+ - ../common/example_images:/example_images:ro
+ # The following is optional but can be used to mount a different configuration file
+ # into the Image Generating plugin. Note that if doing so, values in the configuration
+ # file must match configurations provided in other parts of this file.
+ - ./config/image_gen_config.json:/input.json
+
+ imageScoringPlugin:
+ container_name: image_scoring
+ image: tapis/image_scoring_plugin_py_3.8:0.3.3
+ networks:
+ - cameratraps
+ pid: host
+ # Uncomment the following line to use GPUs with the Image Scoring pluging. Note that
+ # the NVIDIA drivers must be installed, the NVIDIA container toolkit installed, and
+ # Docker must be configured to use it. See the README for details.
+ #
+ # UNCOMMENT --------------
+ # deploy:
+ # resources:
+ # reservations:
+ # devices:
+ # - driver: nvidia
+ # count: all # Use 'all' or specify the number of GPUs you want to use
+ # capabilities: [ gpu ]
+ # --------------------------
+ depends_on:
+ - engine
+ - imageGeneratingPlugin
+ environment:
+ - IMAGE_SCORING_PLUGIN_PORT=6001
+ # The IMAGE_PATH variable needs to match what is specified in the vouume mount below for the
+ # container path portion of the shared images directory.
+ - IMAGE_PATH=/input_images
+ # The IMAGE_PATH_PREFIX is optional and needs to agree with the image_file_prefix
+ # variable in the traps.toml file; when not being used, comment the line.
+ # - IMAGE_FILE_PREFIX=
+ - POWER_MEASURING_PLUGIN_PORT=6010
+ - TRAPS_POWER_LOG_PATH=/logs
+ - TRAPS_TEST_POWER_FUNCTION=0
+ - MONITOR_POWER=true
+ volumes:
+ # mount the traps.toml in the current working directory.
+ # NOTE -- not currently used (TODO)
+ - ./config/traps.toml:/traps.toml:ro
+ # mount the shared images directory from the host to the container directory specified in the
+ # IMAGE_PATH environment variable, above.
+# - ./images_output_dir:/input_images
+ - icicle:/input_images # Using `icicle` volume for images
+
+ powerMonitorPlugin:
+ # USAGE Example:
+ # In your plugins, import the necessary:
+ # E.g. Image_scoring_plugin.py:
+ #
+ # from ctevents.ctevents import send_power_measure_fb_event
+ # import os
+ #
+ # When trying to measure power:
+ # my_pids = [os.getpid()] # any pids you want to measure
+ # monitor_type = [0] # 0 for CPU & GPU, 1 for CPU, 2 for GPU (TODO: 3 for DRAM)
+ # monitor_duration = 10 # seconds
+ # send_power_measure_fb_event(socket, my_pids, monitor_type, monitor_duration)
+ #
+ container_name: power_monitor
+ image: tapis/power_measuring_plugin_py:0.3.3
+ networks:
+ - cameratraps
+ pid: host
+ # Uncomment the following line to use GPUs with the Power Monitor pluging. Note that
+ # the NVIDIA drivers must be installed, the NVIDIA container toolkit installed, and
+ # Docker must be configured to use it. See the README for details.
+ # UNCOMMENT --------------
+ # deploy:
+ # resources:
+ # reservations:
+ # devices:
+ # - driver: nvidia
+ # count: all # Use 'all' or specify the number of GPUs you want to use
+ # capabilities: [ gpu ]
+ # ------------------------
+ depends_on:
+ - engine
+ - imageGeneratingPlugin
+ - imageScoringPlugin
+ environment:
+ - POWER_MEASURING_PLUGIN_PORT=6010
+ # Note: this LOG_PATH is the path in the conatiner, so must matcht the RHS of the mount below.
+ - TRAPS_POWER_LOG_PATH=/logs
+ - TRAPS_TEST_POWER_FUNCTION=1
+ # 1 for debug; Set to 0 when actually use it to measure other plugin's power
+ volumes:
+ # mount the traps.toml in the current working directory.
+ - ./config/traps.toml:/traps.toml:ro
+# - ./power_output_dir:/logs # Path for CPU and GPU power logs
+ - icicle:/logs # Using `icicle` volume for images
+
+volumes:
+ icicle:
+ external: true
diff --git a/resources/events.fbs b/resources/events.fbs
index 58b487e..c34dc34 100644
--- a/resources/events.fbs
+++ b/resources/events.fbs
@@ -32,10 +32,19 @@ table ImageReceivedEvent {
image_format:string;
}
+// Represents a bounding box in the format (xc, yc, w, h)
+table BoundingBox {
+ x_center:float;
+ y_center:float;
+ width:float;
+ height:float;
+}
+
// Represents the probability that an image has a specific label.
table ImageLabelScore {
label:string;
probability:float;
+ bbox:BoundingBox;
}
// Event indicating an image's scores.
diff --git a/src/events.rs b/src/events.rs
index 0ab8f09..92f7bb9 100644
--- a/src/events.rs
+++ b/src/events.rs
@@ -392,6 +392,7 @@ impl ImageReceivedEvent {
pub struct ImageLabelScore {
label: String,
probability: f32,
+ bbox: Option,
}
impl ImageLabelScore {
@@ -400,6 +401,7 @@ impl ImageLabelScore {
ImageLabelScore {
label,
probability,
+ bbox: None,
}
}
@@ -413,14 +415,35 @@ impl ImageLabelScore {
// Get the label's probability value.
let probability = ev.probability();
+ // Parse optional bbox if present.
+ let bbox = match ev.bbox() {
+ Some(bb) => Some(BoundingBox {
+ x_center: bb.x_center(),
+ y_center: bb.y_center(),
+ width: bb.width(),
+ height: bb.height(),
+ }),
+ None => None,
+ };
+
// Return the object.
Result::Ok(ImageLabelScore {
label: String::from(label),
probability,
+ bbox,
})
}
}
+// Simple bounding-box type used by application-level events.
+#[derive(Serialize, Clone)]
+pub struct BoundingBox {
+ pub x_center: f32,
+ pub y_center: f32,
+ pub width: f32,
+ pub height: f32,
+}
+
// ------------------------------
// ------ ImageScoredEvent
// ------------------------------
@@ -465,12 +488,26 @@ impl Event for ImageScoredEvent {
// Assign the string fields.
let label = Some(fbuf.create_string(&score.label));
+ let bbox = match &score.bbox {
+ Some(b) => {
+ let bb_args = gen_events::BoundingBoxArgs {
+ x_center: b.x_center,
+ y_center: b.y_center,
+ width: b.width,
+ height: b.height,
+ };
+ Some(gen_events::BoundingBox::create(&mut fbuf, &bb_args))
+ }
+ None => None,
+ };
+
// Create each generated score object
let im_score = gen_events::ImageLabelScore::create(
&mut fbuf,
&gen_events::ImageLabelScoreArgs {
label,
probability: score.probability,
+ bbox: bbox,
},
);
// Add the current generated score object to the list.
@@ -618,10 +655,22 @@ impl ImageScoredEvent {
// Extract the probability.
let probability = gen_score.probability();
+ // Parse optional bbox from the generated score
+ let bbox = match gen_score.bbox() {
+ Some(bb) => Some(BoundingBox {
+ x_center: bb.x_center(),
+ y_center: bb.y_center(),
+ width: bb.width(),
+ height: bb.height(),
+ }),
+ None => None,
+ };
+
// Create the event imagelabelscore and add it to the vector.
let new_score = ImageLabelScore {
label: label.to_string(),
probability,
+ bbox,
};
scores.push(new_score);
}
diff --git a/src/events_generated.rs b/src/events_generated.rs
index c1f6ed4..a09c578 100644
--- a/src/events_generated.rs
+++ b/src/events_generated.rs
@@ -496,6 +496,142 @@ impl core::fmt::Debug for ImageReceivedEvent<'_> {
ds.finish()
}
}
+pub enum BoundingBoxOffset {}
+#[derive(Copy, Clone, PartialEq)]
+
+pub struct BoundingBox<'a> {
+ pub _tab: flatbuffers::Table<'a>,
+}
+
+impl<'a> flatbuffers::Follow<'a> for BoundingBox<'a> {
+ type Inner = BoundingBox<'a>;
+ #[inline]
+ fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
+ Self { _tab: flatbuffers::Table { buf, loc } }
+ }
+}
+
+impl<'a> BoundingBox<'a> {
+ pub const VT_X_CENTER: flatbuffers::VOffsetT = 4;
+ pub const VT_Y_CENTER: flatbuffers::VOffsetT = 6;
+ pub const VT_WIDTH: flatbuffers::VOffsetT = 8;
+ pub const VT_HEIGHT: flatbuffers::VOffsetT = 10;
+
+ #[inline]
+ pub fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
+ BoundingBox { _tab: table }
+ }
+ #[allow(unused_mut)]
+ pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr>(
+ _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr>,
+ args: &'args BoundingBoxArgs
+ ) -> flatbuffers::WIPOffset> {
+ let mut builder = BoundingBoxBuilder::new(_fbb);
+ builder.add_height(args.height);
+ builder.add_width(args.width);
+ builder.add_y_center(args.y_center);
+ builder.add_x_center(args.x_center);
+ builder.finish()
+ }
+
+
+ #[inline]
+ pub fn x_center(&self) -> f32 {
+ self._tab.get::(BoundingBox::VT_X_CENTER, Some(0.0)).unwrap()
+ }
+ #[inline]
+ pub fn y_center(&self) -> f32 {
+ self._tab.get::(BoundingBox::VT_Y_CENTER, Some(0.0)).unwrap()
+ }
+ #[inline]
+ pub fn width(&self) -> f32 {
+ self._tab.get::(BoundingBox::VT_WIDTH, Some(0.0)).unwrap()
+ }
+ #[inline]
+ pub fn height(&self) -> f32 {
+ self._tab.get::(BoundingBox::VT_HEIGHT, Some(0.0)).unwrap()
+ }
+}
+
+impl flatbuffers::Verifiable for BoundingBox<'_> {
+ #[inline]
+ fn run_verifier(
+ v: &mut flatbuffers::Verifier, pos: usize
+ ) -> Result<(), flatbuffers::InvalidFlatbuffer> {
+ use self::flatbuffers::Verifiable;
+ v.visit_table(pos)?
+ .visit_field::("x_center", Self::VT_X_CENTER, false)?
+ .visit_field::("y_center", Self::VT_Y_CENTER, false)?
+ .visit_field::("width", Self::VT_WIDTH, false)?
+ .visit_field::("height", Self::VT_HEIGHT, false)?
+ .finish();
+ Ok(())
+ }
+}
+pub struct BoundingBoxArgs {
+ pub x_center: f32,
+ pub y_center: f32,
+ pub width: f32,
+ pub height: f32,
+}
+impl<'a> Default for BoundingBoxArgs {
+ #[inline]
+ fn default() -> Self {
+ BoundingBoxArgs {
+ x_center: 0.0,
+ y_center: 0.0,
+ width: 0.0,
+ height: 0.0,
+ }
+ }
+}
+
+pub struct BoundingBoxBuilder<'a: 'b, 'b> {
+ fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a>,
+ start_: flatbuffers::WIPOffset,
+}
+impl<'a: 'b, 'b> BoundingBoxBuilder<'a, 'b> {
+ #[inline]
+ pub fn add_x_center(&mut self, x_center: f32) {
+ self.fbb_.push_slot::(BoundingBox::VT_X_CENTER, x_center, 0.0);
+ }
+ #[inline]
+ pub fn add_y_center(&mut self, y_center: f32) {
+ self.fbb_.push_slot::(BoundingBox::VT_Y_CENTER, y_center, 0.0);
+ }
+ #[inline]
+ pub fn add_width(&mut self, width: f32) {
+ self.fbb_.push_slot::(BoundingBox::VT_WIDTH, width, 0.0);
+ }
+ #[inline]
+ pub fn add_height(&mut self, height: f32) {
+ self.fbb_.push_slot::(BoundingBox::VT_HEIGHT, height, 0.0);
+ }
+ #[inline]
+ pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a>) -> BoundingBoxBuilder<'a, 'b> {
+ let start = _fbb.start_table();
+ BoundingBoxBuilder {
+ fbb_: _fbb,
+ start_: start,
+ }
+ }
+ #[inline]
+ pub fn finish(self) -> flatbuffers::WIPOffset> {
+ let o = self.fbb_.end_table(self.start_);
+ flatbuffers::WIPOffset::new(o.value())
+ }
+}
+
+impl core::fmt::Debug for BoundingBox<'_> {
+ fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+ let mut ds = f.debug_struct("BoundingBox");
+ ds.field("x_center", &self.x_center());
+ ds.field("y_center", &self.y_center());
+ ds.field("width", &self.width());
+ ds.field("height", &self.height());
+ ds.finish()
+ }
+}
pub enum ImageLabelScoreOffset {}
#[derive(Copy, Clone, PartialEq)]
@@ -514,6 +650,7 @@ impl<'a> flatbuffers::Follow<'a> for ImageLabelScore<'a> {
impl<'a> ImageLabelScore<'a> {
pub const VT_LABEL: flatbuffers::VOffsetT = 4;
pub const VT_PROBABILITY: flatbuffers::VOffsetT = 6;
+ pub const VT_BBOX: flatbuffers::VOffsetT = 8;
#[inline]
pub fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
@@ -525,6 +662,7 @@ impl<'a> ImageLabelScore<'a> {
args: &'args ImageLabelScoreArgs<'args>
) -> flatbuffers::WIPOffset> {
let mut builder = ImageLabelScoreBuilder::new(_fbb);
+ if let Some(x) = args.bbox { builder.add_bbox(x); }
builder.add_probability(args.probability);
if let Some(x) = args.label { builder.add_label(x); }
builder.finish()
@@ -539,6 +677,10 @@ impl<'a> ImageLabelScore<'a> {
pub fn probability(&self) -> f32 {
self._tab.get::(ImageLabelScore::VT_PROBABILITY, Some(0.0)).unwrap()
}
+ #[inline]
+ pub fn bbox(&self) -> Option> {
+ self._tab.get::>(ImageLabelScore::VT_BBOX, None)
+ }
}
impl flatbuffers::Verifiable for ImageLabelScore<'_> {
@@ -550,6 +692,7 @@ impl flatbuffers::Verifiable for ImageLabelScore<'_> {
v.visit_table(pos)?
.visit_field::>("label", Self::VT_LABEL, false)?
.visit_field::("probability", Self::VT_PROBABILITY, false)?
+ .visit_field::>("bbox", Self::VT_BBOX, false)?
.finish();
Ok(())
}
@@ -557,6 +700,7 @@ impl flatbuffers::Verifiable for ImageLabelScore<'_> {
pub struct ImageLabelScoreArgs<'a> {
pub label: Option>,
pub probability: f32,
+ pub bbox: Option>>,
}
impl<'a> Default for ImageLabelScoreArgs<'a> {
#[inline]
@@ -564,6 +708,7 @@ impl<'a> Default for ImageLabelScoreArgs<'a> {
ImageLabelScoreArgs {
label: None,
probability: 0.0,
+ bbox: None,
}
}
}
@@ -582,6 +727,10 @@ impl<'a: 'b, 'b> ImageLabelScoreBuilder<'a, 'b> {
self.fbb_.push_slot::(ImageLabelScore::VT_PROBABILITY, probability, 0.0);
}
#[inline]
+ pub fn add_bbox(&mut self, bbox: flatbuffers::WIPOffset>) {
+ self.fbb_.push_slot_always::>(ImageLabelScore::VT_BBOX, bbox);
+ }
+ #[inline]
pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a>) -> ImageLabelScoreBuilder<'a, 'b> {
let start = _fbb.start_table();
ImageLabelScoreBuilder {
@@ -601,6 +750,7 @@ impl core::fmt::Debug for ImageLabelScore<'_> {
let mut ds = f.debug_struct("ImageLabelScore");
ds.field("label", &self.label());
ds.field("probability", &self.probability());
+ ds.field("bbox", &self.bbox());
ds.finish()
}
}
diff --git a/src/python/ctevents/ctevents.py b/src/python/ctevents/ctevents.py
index 157f14c..8c1c7c4 100644
--- a/src/python/ctevents/ctevents.py
+++ b/src/python/ctevents/ctevents.py
@@ -4,7 +4,7 @@
from tokenize import String
import flatbuffers
from ctevents.gen_events import NewImageEvent, ImageReceivedEvent, ImageScoredEvent, ImageStoredEvent, ImageDeletedEvent, ImageLabelScore, PluginStartedEvent, PluginTerminateEvent, PluginTerminatingEvent, MonitorPowerStartEvent, MonitorPowerStopEvent, MonitorType
-from ctevents.gen_events import Event
+from ctevents.gen_events import Event, BoundingBox
from ctevents.gen_events.EventType import EventType
# zmq and socket helper lib
@@ -88,7 +88,7 @@ def _generate_new_image_fb_event(uuid: String, format: String, image: bytearray)
# '''
for i in reversed(range(len(image))):
builder.PrependByte(image[i])
- image_fb = builder.EndVector()
+ image_fb = builder.EndVector(len(image))
# ----- start the new image event, add the individual fields, then call End() -----
# every time you want o create a Table, you need to follow this pattern:
@@ -178,19 +178,36 @@ def _generate_image_scored_fb_event(image_uuid, image_format, scores: "list(dict
for score in scores:
label_fb = builder.CreateString(score['label'])
prob_fb = score['probability']
- scores_fb.append({'label': label_fb, 'prob': prob_fb})
+ bbox_list = score.get('bbox')
+ if bbox_list:
+ if isinstance(bbox_list, (list, tuple)) and len(bbox_list) == 4:
+ BoundingBox.Start(builder)
+ BoundingBox.AddXCenter(builder, bbox_list[0])
+ BoundingBox.AddYCenter(builder, bbox_list[1])
+ BoundingBox.AddWidth(builder, bbox_list[2])
+ BoundingBox.AddHeight(builder, bbox_list[3])
+ bbox = BoundingBox.End(builder)
+ scores_fb.append({'label': label_fb, 'prob': prob_fb, 'bbox': bbox})
+ else:
+ print(f'Got invalid bounding box for {image_uuid}: {bbox}, ignoring bbox')
+ scores_fb.append({'label': label_fb, 'prob': prob_fb})
+ else:
+ scores_fb.append({'label': label_fb, 'prob': prob_fb})
image_label_scores = []
for score in scores_fb:
ImageLabelScore.ImageLabelScoreStart(builder)
ImageLabelScore.AddLabel(builder, score['label'])
ImageLabelScore.AddProbability(builder, score['prob'])
+ bbox = score.get('bbox')
+ if bbox:
+ ImageLabelScore.AddBbox(builder, bbox)
image_label_score = ImageLabelScore.ImageLabelScoreEnd(builder)
image_label_scores.append(image_label_score)
ImageScoredEvent.ImageScoredEventStartScoresVector(builder, len(image_label_scores))
for s in reversed(image_label_scores):
builder.PrependUOffsetTRelative(s)
- scores_fb_vector = builder.EndVector()
+ scores_fb_vector = builder.EndVector(len(image_label_scores))
ImageScoredEvent.Start(builder)
ImageScoredEvent.AddScores(builder, scores_fb_vector)
ImageScoredEvent.AddEventCreateTs(builder, ts_fb)
@@ -460,13 +477,13 @@ def _generate_monitor_power_start_event(pids: list, monitor_types: list, monitor
MonitorPowerStartEvent.MonitorPowerStartEventStartPidsVector(builder, len(pids))
for pid in reversed(pids):
builder.PrependInt32(pid)
- pids_fb = builder.EndVector()
+ pids_fb = builder.EndVector(len(pids))
# Start adding MonitorTypes
MonitorPowerStartEvent.MonitorPowerStartEventStartMonitorTypesVector(builder, len(monitor_types))
for monitor_type in reversed(monitor_types):
builder.PrependInt8(monitor_type)
- monitor_types_fb = builder.EndVector()
+ monitor_types_fb = builder.EndVector(len(monitor_types))
# Start building the MonitorPowerStartEvent
MonitorPowerStartEvent.MonitorPowerStartEventStart(builder)
@@ -508,6 +525,11 @@ def _bytes_to_event(b: bytearray):
returns the raw Flatbuffers event object associated with it.
"""
try:
+ if type(b) == bytes:
+ b = bytearray(b)
+ first_two = bytes(b[0:len(EVENT_TYPE_BYTE_PREFIX['NEW_IMAGE'])])
+ if first_two in EVENT_TYPE_BYTE_PREFIX.values():
+ b = _remove_event_prefix(b)
event = Event.Event.GetRootAs(b, 0)
return event
except Exception as e:
diff --git a/src/python/ctevents/gen_events/BoundingBox.py b/src/python/ctevents/gen_events/BoundingBox.py
new file mode 100644
index 0000000..3c22574
--- /dev/null
+++ b/src/python/ctevents/gen_events/BoundingBox.py
@@ -0,0 +1,72 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace: gen_events
+
+import flatbuffers
+from flatbuffers.compat import import_numpy
+np = import_numpy()
+
+class BoundingBox(object):
+ __slots__ = ['_tab']
+
+ @classmethod
+ def GetRootAs(cls, buf, offset=0):
+ n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
+ x = BoundingBox()
+ x.Init(buf, n + offset)
+ return x
+
+ @classmethod
+ def GetRootAsBoundingBox(cls, buf, offset=0):
+ """This method is deprecated. Please switch to GetRootAs."""
+ return cls.GetRootAs(buf, offset)
+ # BoundingBox
+ def Init(self, buf, pos):
+ self._tab = flatbuffers.table.Table(buf, pos)
+
+ # BoundingBox
+ def XCenter(self):
+ o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(4))
+ if o != 0:
+ return self._tab.Get(flatbuffers.number_types.Float32Flags, o + self._tab.Pos)
+ return 0.0
+
+ # BoundingBox
+ def YCenter(self):
+ o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
+ if o != 0:
+ return self._tab.Get(flatbuffers.number_types.Float32Flags, o + self._tab.Pos)
+ return 0.0
+
+ # BoundingBox
+ def Width(self):
+ o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(8))
+ if o != 0:
+ return self._tab.Get(flatbuffers.number_types.Float32Flags, o + self._tab.Pos)
+ return 0.0
+
+ # BoundingBox
+ def Height(self):
+ o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(10))
+ if o != 0:
+ return self._tab.Get(flatbuffers.number_types.Float32Flags, o + self._tab.Pos)
+ return 0.0
+
+def BoundingBoxStart(builder): builder.StartObject(4)
+def Start(builder):
+ return BoundingBoxStart(builder)
+def BoundingBoxAddXCenter(builder, xCenter): builder.PrependFloat32Slot(0, xCenter, 0.0)
+def AddXCenter(builder, xCenter):
+ return BoundingBoxAddXCenter(builder, xCenter)
+def BoundingBoxAddYCenter(builder, yCenter): builder.PrependFloat32Slot(1, yCenter, 0.0)
+def AddYCenter(builder, yCenter):
+ return BoundingBoxAddYCenter(builder, yCenter)
+def BoundingBoxAddWidth(builder, width): builder.PrependFloat32Slot(2, width, 0.0)
+def AddWidth(builder, width):
+ return BoundingBoxAddWidth(builder, width)
+def BoundingBoxAddHeight(builder, height): builder.PrependFloat32Slot(3, height, 0.0)
+def AddHeight(builder, height):
+ return BoundingBoxAddHeight(builder, height)
+def BoundingBoxEnd(builder): return builder.EndObject()
+def End(builder):
+ return BoundingBoxEnd(builder)
\ No newline at end of file
diff --git a/src/python/ctevents/gen_events/ImageLabelScore.py b/src/python/ctevents/gen_events/ImageLabelScore.py
index 8e1d6d8..1cbc6cb 100644
--- a/src/python/ctevents/gen_events/ImageLabelScore.py
+++ b/src/python/ctevents/gen_events/ImageLabelScore.py
@@ -38,7 +38,18 @@ def Probability(self):
return self._tab.Get(flatbuffers.number_types.Float32Flags, o + self._tab.Pos)
return 0.0
-def ImageLabelScoreStart(builder): builder.StartObject(2)
+ # ImageLabelScore
+ def Bbox(self):
+ o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(8))
+ if o != 0:
+ x = self._tab.Indirect(o + self._tab.Pos)
+ from gen_events.BoundingBox import BoundingBox
+ obj = BoundingBox()
+ obj.Init(self._tab.Bytes, x)
+ return obj
+ return None
+
+def ImageLabelScoreStart(builder): builder.StartObject(3)
def Start(builder):
return ImageLabelScoreStart(builder)
def ImageLabelScoreAddLabel(builder, label): builder.PrependUOffsetTRelativeSlot(0, flatbuffers.number_types.UOffsetTFlags.py_type(label), 0)
@@ -47,6 +58,9 @@ def AddLabel(builder, label):
def ImageLabelScoreAddProbability(builder, probability): builder.PrependFloat32Slot(1, probability, 0.0)
def AddProbability(builder, probability):
return ImageLabelScoreAddProbability(builder, probability)
+def ImageLabelScoreAddBbox(builder, bbox): builder.PrependUOffsetTRelativeSlot(2, flatbuffers.number_types.UOffsetTFlags.py_type(bbox), 0)
+def AddBbox(builder, bbox):
+ return ImageLabelScoreAddBbox(builder, bbox)
def ImageLabelScoreEnd(builder): return builder.EndObject()
def End(builder):
return ImageLabelScoreEnd(builder)
\ No newline at end of file
diff --git a/src/python/ctevents/gen_events/MonitorPowerStartEvent.py b/src/python/ctevents/gen_events/MonitorPowerStartEvent.py
index 8a78486..ee1738d 100644
--- a/src/python/ctevents/gen_events/MonitorPowerStartEvent.py
+++ b/src/python/ctevents/gen_events/MonitorPowerStartEvent.py
@@ -3,17 +3,23 @@
# namespace: gen_events
import flatbuffers
+from flatbuffers.compat import import_numpy
+np = import_numpy()
class MonitorPowerStartEvent(object):
__slots__ = ['_tab']
@classmethod
- def GetRootAsMonitorPowerStartEvent(cls, buf, offset):
+ def GetRootAs(cls, buf, offset=0):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = MonitorPowerStartEvent()
x.Init(buf, n + offset)
return x
+ @classmethod
+ def GetRootAsMonitorPowerStartEvent(cls, buf, offset=0):
+ """This method is deprecated. Please switch to GetRootAs."""
+ return cls.GetRootAs(buf, offset)
# MonitorPowerStartEvent
def Init(self, buf, pos):
self._tab = flatbuffers.table.Table(buf, pos)
@@ -47,6 +53,11 @@ def PidsLength(self):
return self._tab.VectorLen(o)
return 0
+ # MonitorPowerStartEvent
+ def PidsIsNone(self):
+ o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
+ return o == 0
+
# MonitorPowerStartEvent
def MonitorTypes(self, j):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(8))
@@ -69,6 +80,11 @@ def MonitorTypesLength(self):
return self._tab.VectorLen(o)
return 0
+ # MonitorPowerStartEvent
+ def MonitorTypesIsNone(self):
+ o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(8))
+ return o == 0
+
# MonitorPowerStartEvent
def MonitorStartTs(self):
o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(10))
@@ -84,11 +100,29 @@ def MonitorSeconds(self):
return 0
def MonitorPowerStartEventStart(builder): builder.StartObject(5)
+def Start(builder):
+ return MonitorPowerStartEventStart(builder)
def MonitorPowerStartEventAddEventCreateTs(builder, eventCreateTs): builder.PrependUOffsetTRelativeSlot(0, flatbuffers.number_types.UOffsetTFlags.py_type(eventCreateTs), 0)
+def AddEventCreateTs(builder, eventCreateTs):
+ return MonitorPowerStartEventAddEventCreateTs(builder, eventCreateTs)
def MonitorPowerStartEventAddPids(builder, pids): builder.PrependUOffsetTRelativeSlot(1, flatbuffers.number_types.UOffsetTFlags.py_type(pids), 0)
+def AddPids(builder, pids):
+ return MonitorPowerStartEventAddPids(builder, pids)
def MonitorPowerStartEventStartPidsVector(builder, numElems): return builder.StartVector(4, numElems, 4)
+def StartPidsVector(builder, numElems):
+ return MonitorPowerStartEventStartPidsVector(builder, numElems)
def MonitorPowerStartEventAddMonitorTypes(builder, monitorTypes): builder.PrependUOffsetTRelativeSlot(2, flatbuffers.number_types.UOffsetTFlags.py_type(monitorTypes), 0)
+def AddMonitorTypes(builder, monitorTypes):
+ return MonitorPowerStartEventAddMonitorTypes(builder, monitorTypes)
def MonitorPowerStartEventStartMonitorTypesVector(builder, numElems): return builder.StartVector(1, numElems, 1)
+def StartMonitorTypesVector(builder, numElems):
+ return MonitorPowerStartEventStartMonitorTypesVector(builder, numElems)
def MonitorPowerStartEventAddMonitorStartTs(builder, monitorStartTs): builder.PrependUOffsetTRelativeSlot(3, flatbuffers.number_types.UOffsetTFlags.py_type(monitorStartTs), 0)
+def AddMonitorStartTs(builder, monitorStartTs):
+ return MonitorPowerStartEventAddMonitorStartTs(builder, monitorStartTs)
def MonitorPowerStartEventAddMonitorSeconds(builder, monitorSeconds): builder.PrependUint32Slot(4, monitorSeconds, 0)
+def AddMonitorSeconds(builder, monitorSeconds):
+ return MonitorPowerStartEventAddMonitorSeconds(builder, monitorSeconds)
def MonitorPowerStartEventEnd(builder): return builder.EndObject()
+def End(builder):
+ return MonitorPowerStartEventEnd(builder)
\ No newline at end of file
diff --git a/src/python/ctevents/gen_events/MonitorPowerStopEvent.py b/src/python/ctevents/gen_events/MonitorPowerStopEvent.py
index cafd9bc..cd025cc 100644
--- a/src/python/ctevents/gen_events/MonitorPowerStopEvent.py
+++ b/src/python/ctevents/gen_events/MonitorPowerStopEvent.py
@@ -3,17 +3,23 @@
# namespace: gen_events
import flatbuffers
+from flatbuffers.compat import import_numpy
+np = import_numpy()
class MonitorPowerStopEvent(object):
__slots__ = ['_tab']
@classmethod
- def GetRootAsMonitorPowerStopEvent(cls, buf, offset):
+ def GetRootAs(cls, buf, offset=0):
n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
x = MonitorPowerStopEvent()
x.Init(buf, n + offset)
return x
+ @classmethod
+ def GetRootAsMonitorPowerStopEvent(cls, buf, offset=0):
+ """This method is deprecated. Please switch to GetRootAs."""
+ return cls.GetRootAs(buf, offset)
# MonitorPowerStopEvent
def Init(self, buf, pos):
self._tab = flatbuffers.table.Table(buf, pos)
@@ -47,8 +53,23 @@ def PidsLength(self):
return self._tab.VectorLen(o)
return 0
+ # MonitorPowerStopEvent
+ def PidsIsNone(self):
+ o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
+ return o == 0
+
def MonitorPowerStopEventStart(builder): builder.StartObject(2)
+def Start(builder):
+ return MonitorPowerStopEventStart(builder)
def MonitorPowerStopEventAddEventCreateTs(builder, eventCreateTs): builder.PrependUOffsetTRelativeSlot(0, flatbuffers.number_types.UOffsetTFlags.py_type(eventCreateTs), 0)
+def AddEventCreateTs(builder, eventCreateTs):
+ return MonitorPowerStopEventAddEventCreateTs(builder, eventCreateTs)
def MonitorPowerStopEventAddPids(builder, pids): builder.PrependUOffsetTRelativeSlot(1, flatbuffers.number_types.UOffsetTFlags.py_type(pids), 0)
+def AddPids(builder, pids):
+ return MonitorPowerStopEventAddPids(builder, pids)
def MonitorPowerStopEventStartPidsVector(builder, numElems): return builder.StartVector(4, numElems, 4)
+def StartPidsVector(builder, numElems):
+ return MonitorPowerStopEventStartPidsVector(builder, numElems)
def MonitorPowerStopEventEnd(builder): return builder.EndObject()
+def End(builder):
+ return MonitorPowerStopEventEnd(builder)
\ No newline at end of file
diff --git a/src/python/ctevents/gen_events/MonitorType.py b/src/python/ctevents/gen_events/MonitorType.py
index 05cee13..7c63372 100644
--- a/src/python/ctevents/gen_events/MonitorType.py
+++ b/src/python/ctevents/gen_events/MonitorType.py
@@ -3,8 +3,7 @@
# namespace: gen_events
class MonitorType(object):
- ALL = 0
- CPU = 1
- GPU = 2
- DRAM = 3
-
+ ALL = 1
+ CPU = 2
+ GPU = 3
+ DRAM = 4
diff --git a/src/python/test_ctevents.py b/src/python/test_ctevents.py
index 212b320..dd64ae0 100644
--- a/src/python/test_ctevents.py
+++ b/src/python/test_ctevents.py
@@ -1,6 +1,7 @@
import datetime
import uuid
import sys
+import os
from ctevents.ctevents import _bytes_to_event, _event_to_typed_event
from ctevents.ctevents import _generate_new_image_fb_event, _generate_image_received_fb_event, _generate_image_scored_fb_event, _generate_delete_image_fb_event, _generate_start_plugin_fb_event, _generate_terminating_plugin_fb_event, _generate_terminate_plugin_fb_event, _generate_store_image_fb_event
from ctevents.ctevents import _generate_new_image_fb_with_prefix, _generate_image_received_fb_with_prefix, _generate_store_image_fb_with_prefix, _generate_terminating_plugin_fb_with_prefix, _generate_delete_image_fb_with_prefix, _generate_image_scored_fb_with_prefix, _generate_start_plugin_fb_with_prefix, _generate_terminate_plugin_fb_with_prefix
@@ -94,7 +95,7 @@ def test_image_received_event_fb():
# check the fields; each should match the previous test data we generated
assert image_received_event.ImageUuid() == uuid_str.encode('utf-8')
assert image_received_event.ImageFormat() == format_str.encode('utf-8')
- now = datetime.datetime.utcnow().isoformat()
+ now = datetime.datetime.now(datetime.UTC).isoformat()
# assert times have the same year --
assert now[:19] == image_received_event.EventCreateTs().decode('utf-8')[:19]
@@ -328,7 +329,7 @@ def test_image_scored_event_fb():
assert "8f5f3962-d301-4e96-9994-3bd63c472ce8" == image_scored_event.ImageUuid().decode('utf-8')
assert image_scored_event.ImageFormat() == image_format.encode('utf-8')
- now = datetime.datetime.utcnow().isoformat()
+ now = datetime.datetime.now(datetime.UTC).isoformat()
# assert times have the same year --
assert now[:4] == image_scored_event.EventCreateTs().decode('utf-8')[:4]
@@ -363,6 +364,36 @@ def test_image_scored_event_with_prefix():
# check that prefix is the right thing
assert image_scored_fb[0:2] == EVENT_TYPE_BYTE_PREFIX['IMAGE_SCORED']
+def test_image_scored_event_with_bbox():
+ """
+ A basic test function to check that serializing and deserializing image scored event with bounding boxes
+ """
+ scores = [
+ {"image_uuid": "8f5f3962-d301-4e96-9994-3bd63c472ce8", "label": "lab", "probability": 0.95, "bbox": [0.5, 0.5, 0.4, 0.4]},
+ {"image_uuid": "8f5f3962-d301-4e96-9994-3bd63c472ce8", "label": "golden_retriever", "probability": 0.05, "bbox": [0.6, 0.6, 0.3, 0.5]},
+ {"image_uuid": "8f5f3962-d301-4e96-9994-3bd63c472ce8", "label": "pug", "probability": 0.012, "bbox": [0.4, 0.4, 0.2, 0.2]}
+ ]
+ image_uuid = "8f5f3962-d301-4e96-9994-3bd63c472ce8"
+ image_format = 'jpg'
+
+ # make a test image scored event flattbuffer
+ image_scored_fb = _generate_image_scored_fb_with_prefix(image_uuid, image_format, scores)
+
+ # check that prefix is the right thing
+ assert image_scored_fb[0:2] == EVENT_TYPE_BYTE_PREFIX['IMAGE_SCORED']
+ e = _bytes_to_event(image_scored_fb)
+
+ # convert the root event object to a typed event (of type new image)
+ image_scored_event = _event_to_typed_event(e)
+
+ for i in range(image_scored_event.ScoresLength()):
+ bbox = image_scored_event.Scores(i).Bbox()
+ assert bbox is not None
+ assert abs(bbox.XCenter() - scores[i]['bbox'][0]) < 0.01
+ assert abs(bbox.YCenter() - scores[i]['bbox'][1]) < 0.01
+ assert abs(bbox.Width() - scores[i]['bbox'][2]) < 0.01
+ assert abs(bbox.Height() - scores[i]['bbox'][3]) < 0.01
+
if __name__ == "__main__":
test_new_image_event_fb()
@@ -371,6 +402,7 @@ def test_image_scored_event_with_prefix():
test_image_received_event_with_prefix()
test_image_scored_event_fb()
test_image_scored_event_with_prefix()
+ test_image_scored_event_with_bbox()
test_image_stored_event_with_prefix()
test_image_stored_event_fb()
test_delete_image_event_fb()
@@ -380,4 +412,4 @@ def test_image_scored_event_with_prefix():
test_terminating_plugin_event_fb()
test_terminating_plugin_event_with_prefix()
test_terminate_plugin_event_fb()
- test_terminate_plugin_event_with_prefix()
\ No newline at end of file
+ test_terminate_plugin_event_with_prefix()