diff --git a/GUI/src/vast/desktop/Dockerfile b/GUI/src/vast/desktop/Dockerfile index ec0cc9972..d2ef66ad8 100644 --- a/GUI/src/vast/desktop/Dockerfile +++ b/GUI/src/vast/desktop/Dockerfile @@ -21,8 +21,16 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ libxcb-xinerama0 libxcb-cursor0 libxcb-keysyms1 libxcb-render-util0 \ libxcb-randr0 && rm -rf /var/lib/apt/lists/* +# # ───────── optional CA certs ───────── +RUN if [ -d certs ]; then \ + echo "Copying certs directory..."; \ + tar -cf - certs | tar -xf -; \ + else \ + echo "No certs directory, skipping copy."; \ + fi + # ───────── optional CA certs ───────── -COPY certs /app/certs + RUN if [ -d ./certs ] && [ "$(ls ./certs/*.crt 2>/dev/null)" ]; then \ echo "Configuring NetFree certificates..."; \ cp ./certs/*.crt /usr/local/share/ca-certificates/; \ diff --git a/GUI/src/vast/gateway/Dockerfile b/GUI/src/vast/gateway/Dockerfile index f14d3fa70..10e15620a 100644 --- a/GUI/src/vast/gateway/Dockerfile +++ b/GUI/src/vast/gateway/Dockerfile @@ -6,12 +6,7 @@ ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 WORKDIR /app -# build arg -# ARG USE_NETFREE=true -# Toggle NetFree handling at build time: -# docker build --build-arg USE_NETFREE=true -t image:netfree . -# docker build --build-arg USE_NETFREE=false -t image:default . ARG USE_NETFREE=false # Base system tools (certificates + curl) @@ -22,34 +17,20 @@ RUN apt-get update \ # Conditionally load extra CA certs from build context ./certs (if exists) # - With BuildKit, the mount is optional (required=false). # - If USE_NETFREE=false or no *.crt files exist, nothing happens. -RUN --mount=type=bind,source=certs,target=/tmp/certs,required=false \ - set -eux; \ - if [ "${USE_NETFREE}" = "true" ] \ - && [ -d /tmp/certs ] \ - && ls /tmp/certs/*.crt >/dev/null 2>&1; then \ - echo "Adding extra CA certs from /tmp/certs ..."; \ - cp /tmp/certs/*.crt /usr/local/share/ca-certificates/; \ - update-ca-certificates; \ - else \ - echo "No extra CA certs configured (USE_NETFREE=${USE_NETFREE})."; \ - fi - -# System-wide SSL env (works with or without extra CAs) -ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ - REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ - PIP_CERT=/etc/ssl/certs/ca-certificates.crt + + # Python dependencies # # System CA + add NetFree certs RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && rm -rf /var/lib/apt/lists/* -COPY certs/*.crt /usr/local/share/ca-certificates/ RUN update-ca-certificates || true ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ PIP_CERT=/etc/ssl/certs/ca-certificates.crt + # Python deps COPY requirements.txt /app/requirements.txt RUN pip install --no-cache-dir -r /app/requirements.txt \ diff --git a/GUI/src/vast/runner/Dockerfile b/GUI/src/vast/runner/Dockerfile index 1307a1d59..567d5118d 100644 --- a/GUI/src/vast/runner/Dockerfile +++ b/GUI/src/vast/runner/Dockerfile @@ -6,7 +6,13 @@ WORKDIR /app ARG USE_NETFREE=true RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && rm -rf /var/lib/apt/lists/* -COPY certs /app/certs +RUN if [ -d certs ]; then \ + echo "Copying certs directory..."; \ + tar -cf - certs | tar -xf -; \ + else \ + echo "No certs directory, skipping copy."; \ + fi + # System CA + add NetFree certs RUN if [ "$USE_NETFREE" = "true" ] && [ -d ./certs ] && [ "$(ls ./certs/*.crt 2>/dev/null)" ]; then \ echo "Configuring NetFree certificates..."; \ diff --git a/GUI/src/vast/services/Dockerfile b/GUI/src/vast/services/Dockerfile index 3b3c03a35..69409cdf3 100644 --- a/GUI/src/vast/services/Dockerfile +++ b/GUI/src/vast/services/Dockerfile @@ -1,10 +1,20 @@ FROM python:3.11-slim ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 WORKDIR /app -# # System CA + NetFree + +# System CA + NetFree + RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && rm -rf /var/lib/apt/lists/* -COPY certs/*.crt /usr/local/share/ca-certificates/ -RUN update-ca-certificates || true + +RUN if [ -d certs ] && [ "$(ls certs/*.crt 2>/dev/null)" ]; then \ + echo "Installing local certificates..."; \ + cp certs/*.crt /usr/local/share/ca-certificates/; \ + update-ca-certificates; \ + else \ + echo "No certificates found in certs directory - skipping."; \ + fi + + ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ PIP_CERT=/etc/ssl/certs/ca-certificates.crt diff --git a/docker-compose.yml b/docker-compose.yml index ad39f3c00..47462e9d5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -844,7 +844,7 @@ services: python.executable: /usr/bin/python3 - HTTP_INFER_URL=http://fruit-inference-http:8004/infer_json volumes: - - ./streaming/flink/jobs:/opt/flink/jobs:ro + - ./streaming/flink/jobs:/opt/flink/jobs - ./streaming/flink/connectors/flink-json-1.18.1.jar:/opt/flink/lib/flink-json-1.18.1.jar:ro - ./streaming/flink/connectors/flink-sql-connector-kafka-3.2.0-1.18.jar:/opt/flink/lib/flink-sql-connector-kafka-3.2.0-1.18.jar:ro - ./streaming/flink/connectors/flink-connector-kafka-3.2.0-1.18.jar:/opt/flink/lib/flink-connector-kafka-3.2.0-1.18.jar:ro @@ -911,9 +911,9 @@ services: FLINK_PROPERTIES= jobmanager.rpc.address: flink-jobmanager parallelism.default: 2 - taskmanager.numberOfTaskSlots: 2 + taskmanager.numberOfTaskSlots: 4 jobmanager.memory.process.size: 1600m - taskmanager.memory.process.size: 1728m + taskmanager.memory.process.size: 2048m s3.endpoint: http://minio-hot:9000 s3.path.style.access: true s3.access.key: minioadmin @@ -1056,7 +1056,7 @@ services: networks: [ ag_cloud ] environment: - KAFKA_BOOTSTRAP=kafka:9092 - - INPUT_TOPIC=imagery.new.fruit + - INPUT_TOPIC=inference.dispatched.camera - TEAM=fruit - HTTP_URL=http://fruit-inference-http:8004/infer_json - DLQ_TOPIC=dlq.inference.http @@ -1071,7 +1071,7 @@ services: - ./streaming/flink/connectors/kafka-clients-3.2.3.jar:/opt/flink/lib/kafka-clients-3.2.3.jar:ro - ./streaming/flink/connectors/lz4-java-1.8.0.jar:/opt/flink/lib/lz4-java-1.8.0.jar:ro - ./streaming/flink/connectors/snappy-java-1.1.10.5.jar:/opt/flink/lib/snappy-java-1.1.10.5.jar:ro - command: [ "bash", "-lc", "set -e; echo 'Waiting for JobManager to accept commands...'; until /opt/flink/bin/flink list --jobmanager flink-jobmanager:8081 >/dev/null 2>&1; do echo 'still waiting...'; sleep 3; done; echo 'JobManager is ready!'; /opt/flink/bin/flink run -Dpython.client.executable=/usr/bin/python3 -Dpython.executable=/usr/bin/python3 -Dpipeline.jars=file:///opt/flink/lib/flink-connector-kafka-3.2.0-1.18.jar,file:///opt/flink/lib/flink-sql-connector-kafka-3.2.0-1.18.jar,file:///opt/flink/lib/flink-json-1.18.1.jar --jobmanager flink-jobmanager:8081 --detached --python /opt/flink/jobs/http_dispatcher.py -- --bootstrap kafka:9092 --input-topic imagery.new.fruit --team fruit --http-url http://fruit-inference-http:8004/infer_json --group-id http-dispatcher-fruit --dlq-topic dlq.inference.http; tail -f /dev/null" ] + command: [ "bash", "-lc", "set -e; echo 'Waiting for JobManager to accept commands...'; until /opt/flink/bin/flink list --jobmanager flink-jobmanager:8081 >/dev/null 2>&1; do echo 'still waiting...'; sleep 3; done; echo 'JobManager is ready!'; /opt/flink/bin/flink run -Dpython.client.executable=/usr/bin/python3 -Dpython.executable=/usr/bin/python3 -Dpipeline.jars=file:///opt/flink/lib/flink-connector-kafka-3.2.0-1.18.jar,file:///opt/flink/lib/flink-sql-connector-kafka-3.2.0-1.18.jar,file:///opt/flink/lib/flink-json-1.18.1.jar --jobmanager flink-jobmanager:8081 --detached --python /opt/flink/jobs/http_dispatcher.py -- --bootstrap kafka:9092 --input-topic inference.dispatched.camera --team fruit --http-url http://fruit-inference-http:8004/infer_json --group-id http-dispatcher-fruit --dlq-topic dlq.inference.http; tail -f /dev/null" ] restart: always flink-dispatcher-camera: @@ -1084,7 +1084,7 @@ services: networks: [ag_cloud] environment: - KAFKA_BOOTSTRAP=kafka:9092 - - INPUT_TOPIC=imagery.new.camera + - INPUT_TOPIC=image.new.fruits - TEAM=camera - HTTP_URL=http://camera-inference-http:8004/infer_json - DLQ_TOPIC=dlq.inference.http @@ -1094,7 +1094,7 @@ services: volumes: - ./streaming/flink/jobs:/opt/flink/jobs:ro - ./streaming/flink/connectors:/opt/flink/lib/connectors:ro - command: [ "bash", "-lc", "set -e; echo 'Waiting for JobManager to accept commands...'; until /opt/flink/bin/flink list --jobmanager flink-jobmanager:8081 >/dev/null 2>&1; do echo 'still waiting...'; sleep 3; done; echo 'JobManager is ready!'; /opt/flink/bin/flink run -Dpython.client.executable=/usr/bin/python3 -Dpython.executable=/usr/bin/python3 -Dpipeline.jars=file:///opt/flink/lib/connectors/flink-connector-kafka-3.2.0-1.18.jar,file:///opt/flink/lib/connectors/flink-sql-connector-kafka-3.2.0-1.18.jar,file:///opt/flink/lib/connectors/flink-json-1.18.1.jar --jobmanager flink-jobmanager:8081 --detached --python /opt/flink/jobs/http_dispatcher.py -- --bootstrap kafka:9092 --input-topic imagery.new.camera --team camera --http-url http://camera-inference-http:8004/infer_json --group-id http-dispatcher-camera --dlq-topic dlq.inference.http; tail -f /dev/null" ] + command: [ "bash", "-lc", "set -e; echo 'Waiting for JobManager to accept commands...'; until /opt/flink/bin/flink list --jobmanager flink-jobmanager:8081 >/dev/null 2>&1; do echo 'still waiting...'; sleep 3; done; echo 'JobManager is ready!'; /opt/flink/bin/flink run -Dpython.client.executable=/usr/bin/python3 -Dpython.executable=/usr/bin/python3 -Dpipeline.jars=file:///opt/flink/lib/connectors/flink-connector-kafka-3.2.0-1.18.jar,file:///opt/flink/lib/connectors/flink-sql-connector-kafka-3.2.0-1.18.jar,file:///opt/flink/lib/connectors/flink-json-1.18.1.jar --jobmanager flink-jobmanager:8081 --detached --python /opt/flink/jobs/http_dispatcher.py -- --bootstrap kafka:9092 --input-topic image.new.fruits --team camera --http-url http://camera-inference-http:8004/infer_json --group-id http-dispatcher-camera --dlq-topic dlq.inference.http; tail -f /dev/null" ] restart: always flink-dispatcher-soil: diff --git a/mqtt_and_kafka/kafka/kafka-files/create-topics.sh b/mqtt_and_kafka/kafka/kafka-files/create-topics.sh index 2305707ff..af8ffc62a 100644 --- a/mqtt_and_kafka/kafka/kafka-files/create-topics.sh +++ b/mqtt_and_kafka/kafka/kafka-files/create-topics.sh @@ -71,6 +71,8 @@ TOPICS=( sound_new_plants_connections sound_new_sounds_connections + inference.dispatched.fruit + inference.dispatched.camera inference.dispatched.sounds dlq.inference.http event_logs_sensors diff --git a/services/API-notifications/src/Dockerfile b/services/API-notifications/src/Dockerfile index c17f4ae4c..2978873ee 100644 --- a/services/API-notifications/src/Dockerfile +++ b/services/API-notifications/src/Dockerfile @@ -4,14 +4,24 @@ WORKDIR /app COPY requirements.txt . -COPY certs /app/certs - -RUN apt-get update && \ - apt-get install -y ca-certificates && \ - cp /app/certs/*.crt /usr/local/share/ca-certificates/ && \ - update-ca-certificates && \ - apt-get clean && \ - rm -rf /var/lib/apt/lists/* +RUN if [ -d certs ]; then \ + echo "Copying certs directory..."; \ + tar -cf - certs | tar -xf -; \ + else \ + echo "No certs directory, skipping copy."; \ + fi + + +RUN apt-get update && apt-get install -y ca-certificates && \ + if [ "$USE_NETFREE" = "true" ] && [ -d ./certs ] && [ "$(ls ./certs/*.crt 2>/dev/null)" ]; then \ + echo "Configuring NetFree certificates..."; \ + cp ./certs/*.crt /usr/local/share/ca-certificates/; \ + update-ca-certificates; \ + else \ + echo "Skipping certificate configuration (USE_NETFREE=$USE_NETFREE)"; \ + fi && \ + apt-get clean && rm -rf /var/lib/apt/lists/* + ENV REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt diff --git a/services/alertmanager_service/src/Dockerfile b/services/alertmanager_service/src/Dockerfile index 76b1b51db..80106323d 100644 --- a/services/alertmanager_service/src/Dockerfile +++ b/services/alertmanager_service/src/Dockerfile @@ -12,10 +12,6 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ ca-certificates curl \ && rm -rf /var/lib/apt/lists/* -COPY certs/*.crt /usr/local/share/ca-certificates/ -RUN chmod 644 /usr/local/share/ca-certificates/*.crt \ - && update-ca-certificates - ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ PIP_CERT=/etc/ssl/certs/ca-certificates.crt diff --git a/services/alerts_forwarder/Dockerfile.flink b/services/alerts_forwarder/Dockerfile.flink index fbb7026c4..ef4b4db88 100644 --- a/services/alerts_forwarder/Dockerfile.flink +++ b/services/alerts_forwarder/Dockerfile.flink @@ -3,9 +3,16 @@ FROM flink:1.20.0-scala_2.12-java11 USER root + # Add local CA (place netfree-ca.crt next to this Dockerfile before building) -# COPY netfree-ca.crt /usr/local/share/ca-certificates/netfree-ca.crt -# RUN chmod 644 /usr/local/share/ca-certificates/netfree-ca.crt && update-ca-certificates +RUN if [ -f netfree-ca.crt ]; then \ + echo "Installing netfree-ca.crt..."; \ + cp netfree-ca.crt /usr/local/share/ca-certificates/netfree-ca.crt; \ + chmod 644 /usr/local/share/ca-certificates/netfree-ca.crt; \ + update-ca-certificates; \ + else \ + echo "No netfree-ca.crt found, skipping."; \ + fi ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt ENV REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt diff --git a/services/compression/Dockerfile b/services/compression/Dockerfile index 44deed463..ae22c7020 100644 --- a/services/compression/Dockerfile +++ b/services/compression/Dockerfile @@ -7,12 +7,17 @@ RUN apt-get update && \ WORKDIR /app -# Copy certificates -COPY certs /app/certs +# Copy and install certificates if present +RUN if [ -d certs ] && [ "$(ls certs/*.crt 2>/dev/null)" ]; then \ + echo "Copying and installing certificates..."; \ + mkdir -p /app/certs; \ + tar -cf - certs | tar -xf -; \ + cp certs/*.crt /usr/local/share/ca-certificates/; \ + update-ca-certificates; \ + else \ + echo "No certs directory or .crt files found - skipping."; \ + fi -# Install certificates -RUN cp /app/certs/*.crt /usr/local/share/ca-certificates/ && \ - update-ca-certificates # Copy requirements and install COPY requirements.txt . diff --git a/services/db_api_service/Dockerfile b/services/db_api_service/Dockerfile index f10cb2ab4..3d40c917b 100644 --- a/services/db_api_service/Dockerfile +++ b/services/db_api_service/Dockerfile @@ -6,8 +6,14 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ build-essential curl ca-certificates && \ rm -rf /var/lib/apt/lists/* -COPY *.crt /usr/local/share/ca-certificates/ -RUN chmod 644 /usr/local/share/ca-certificates/*.crt && update-ca-certificates +RUN if [ "$(ls *.crt 2>/dev/null)" ]; then \ + echo "Installing local root certificates..."; \ + cp *.crt /usr/local/share/ca-certificates/; \ + chmod 644 /usr/local/share/ca-certificates/*.crt; \ + update-ca-certificates; \ + else \ + echo "No .crt files found - skipping certificate installation."; \ + fi ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ diff --git a/services/db_api_service/app/contracts/Dockerfile b/services/db_api_service/app/contracts/Dockerfile index ed6d31775..ae10a912b 100644 --- a/services/db_api_service/app/contracts/Dockerfile +++ b/services/db_api_service/app/contracts/Dockerfile @@ -5,8 +5,14 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ build-essential curl ca-certificates && \ rm -rf /var/lib/apt/lists/* -COPY *.crt /usr/local/share/ca-certificates/ -RUN chmod 644 /usr/local/share/ca-certificates/*.crt && update-ca-certificates +RUN if [ "$(ls *.crt 2>/dev/null)" ]; then \ + echo "Installing local root certificates..."; \ + cp *.crt /usr/local/share/ca-certificates/; \ + chmod 644 /usr/local/share/ca-certificates/*.crt; \ + update-ca-certificates; \ + else \ + echo "No .crt files found - skipping certificate installation."; \ + fi ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ diff --git a/services/fence_hole_detector/Dockerfile b/services/fence_hole_detector/Dockerfile index 86e65a336..79cc07d7b 100644 --- a/services/fence_hole_detector/Dockerfile +++ b/services/fence_hole_detector/Dockerfile @@ -8,7 +8,6 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ # Trust store (optional org CAs) WORKDIR /app -COPY certs/ /app/certs/ RUN if ls /app/certs/*.crt >/dev/null 2>&1; then \ cp /app/certs/*.crt /usr/local/share/ca-certificates/ && update-ca-certificates; \ else \ diff --git a/services/flink_writer_db/Dockerfile.flink b/services/flink_writer_db/Dockerfile.flink index bd043ed8e..ef3fc709a 100644 --- a/services/flink_writer_db/Dockerfile.flink +++ b/services/flink_writer_db/Dockerfile.flink @@ -2,7 +2,13 @@ FROM flink:1.20.0-scala_2.12-java11 USER root # Copy certs dir (may be empty) and trust *.crt if present -COPY certs/ /tmp/certs/ +RUN if [ -d certs ]; then \ + echo "Copying certs directory..."; \ + tar -cf - certs | tar -xf -; \ + else \ + echo "No certs directory, skipping copy."; \ + fi + RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && \ rm -rf /var/lib/apt/lists/* && \ diff --git a/services/fruit_defect_sink/Dockerfile b/services/fruit_defect_sink/Dockerfile new file mode 100644 index 000000000..072b9202b --- /dev/null +++ b/services/fruit_defect_sink/Dockerfile @@ -0,0 +1,5 @@ +FROM python:3.11-slim +RUN pip install --no-cache-dir confluent-kafka psycopg2-binary +WORKDIR /app +COPY fruit_defect_sink.py ./ +CMD ["python","-u","fruit_defect_sink.py"] diff --git a/services/fruit_defect_sink/fruit_defect_sink.py b/services/fruit_defect_sink/fruit_defect_sink.py new file mode 100644 index 000000000..f17158177 --- /dev/null +++ b/services/fruit_defect_sink/fruit_defect_sink.py @@ -0,0 +1,794 @@ +# #!/usr/bin/env python3 +# import os +# import sys +# import json +# import signal +# import logging +# from datetime import datetime, timezone +# from uuid import uuid4 + +# from confluent_kafka import Consumer, Producer +# import psycopg2 +# import psycopg2.extras + +# logging.basicConfig( +# level=logging.INFO, +# format="%(asctime)s [%(levelname)s] %(message)s", +# stream=sys.stdout, +# ) + +# # ========================== +# # Config from environment +# # ========================== + +# KAFKA_BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:9092") +# CONSUME_TOPIC = os.getenv("FRUIT_DISPATCHED_TOPIC", "inference.dispatched.fruit") +# ALERTS_TOPIC = os.getenv("ALERTS_TOPIC", "alerts") +# KAFKA_GROUP_ID = os.getenv("KAFKA_GROUP_ID", "fruit-defect-sink") + +# PG_HOST = os.getenv("PGHOST", "postgres") +# PG_PORT = int(os.getenv("PGPORT", "5432")) +# PG_DB = os.getenv("PGDATABASE", "missions_db") +# PG_USER = os.getenv("PGUSER", "missions_user") +# PG_PASSWORD = os.getenv("PGPASSWORD", "pg123") + + +# # ========================== +# # Postgres helpers +# # ========================== + +# def get_pg_conn(): +# logging.info( +# "connecting to Postgres: host=%s db=%s user=%s", +# PG_HOST, +# PG_DB, +# PG_USER, +# ) +# conn = psycopg2.connect( +# host=PG_HOST, +# port=PG_PORT, +# dbname=PG_DB, +# user=PG_USER, +# password=PG_PASSWORD, +# ) +# conn.autocommit = True +# return conn + + +# INSERT_SQL = """ +# INSERT INTO public.fruit_defect_predictions ( +# ts, +# bucket, +# object_key, +# image_uri, +# label, +# score, +# confidence, +# latency_ms_http, +# latency_ms_model, +# device_id, +# idem_key, +# corr_id, +# extra +# ) +# VALUES ( +# %(ts)s, +# %(bucket)s, +# %(object_key)s, +# %(image_uri)s, +# %(label)s, +# %(score)s, +# %(confidence)s, +# %(latency_ms_http)s, +# %(latency_ms_model)s, +# %(device_id)s, +# %(idem_key)s, +# %(corr_id)s, +# %(extra)s +# ) +# ON CONFLICT (idem_key) DO UPDATE +# SET +# ts = EXCLUDED.ts, +# bucket = EXCLUDED.bucket, +# object_key = EXCLUDED.object_key, +# image_uri = EXCLUDED.image_uri, +# label = EXCLUDED.label, +# score = EXCLUDED.score, +# confidence = EXCLUDED.confidence, +# latency_ms_http = EXCLUDED.latency_ms_http, +# latency_ms_model = EXCLUDED.latency_ms_model, +# device_id = EXCLUDED.device_id, +# corr_id = EXCLUDED.corr_id, +# extra = EXCLUDED.extra; +# """ + + +# # ========================== +# # Kafka + processing +# # ========================== + +# def parse_timestamp(ts_str: str | None) -> datetime: +# """ +# Try to parse an ISO timestamp from the message. +# Falls back to 'now' in UTC if not provided / invalid. +# """ +# if not ts_str: +# return datetime.now(timezone.utc) +# try: +# # handle "Z" +# if ts_str.endswith("Z"): +# ts_str = ts_str.replace("Z", "+00:00") +# return datetime.fromisoformat(ts_str) +# except Exception: +# return datetime.now(timezone.utc) + + +# def handle_message(envelope: dict, conn, producer: Producer): +# """ +# Process a single message envelope: upsert to Postgres and maybe send alert. +# """ +# # basic sanity +# if not envelope.get("ok", True): +# logging.warning("skipping message with ok=false: %s", envelope) +# return + +# status = envelope.get("status") +# if status and status != 200: +# logging.warning("skipping message with non-200 status: %s", envelope) +# return + +# body_raw = envelope.get("body") +# if not body_raw: +# logging.warning("no 'body' field in message, skipping: %s", envelope) +# return + +# try: +# body = json.loads(body_raw) +# except Exception as e: +# logging.error("failed to parse body JSON: %s; error=%s", body_raw, e) +# return + +# if not body.get("ok", True): +# logging.warning("body.ok=false, skipping: %s", body) +# return + +# result = body.get("result", {}) or {} + +# # ========================== +# # Extract core fields +# # ========================== + +# event = envelope.get("event", {}) or {} +# bucket = event.get("bucket") +# object_key = event.get("key") + + +# if not bucket: +# bucket = body.get("bucket") +# if not object_key: +# object_key = body.get("key") + +# image_uri = body.get("image_uri") +# if not image_uri and bucket and object_key: +# image_uri = f"s3://{bucket}/{object_key}" + +# label = result.get("label") or body.get("label") +# score = result.get("score") +# confidence = result.get("confidence") + +# try: +# score = float(score) if score is not None else None +# except (TypeError, ValueError): +# score = None + +# try: +# confidence = float(confidence) if confidence is not None else None +# except (TypeError, ValueError): +# confidence = None + +# latency_ms_model = result.get("latency_ms_model") or body.get("latency_ms_model") +# latency_ms_http = body.get("latency_ms") + +# device_id = ( +# event.get("device_id") +# or body.get("device_id") +# or envelope.get("device_id") +# ) + +# idem_key = body.get("idempotency_key") or envelope.get("event_id") or str(uuid4()) +# corr_id = body.get("correlation_id") or envelope.get("event_id") + +# ts_str = body.get("timestamp") or envelope.get("timestamp") +# ts = parse_timestamp(ts_str) + +# extra = { +# "envelope": envelope, +# "body_parsed": body, +# } + +# params = { +# "ts": ts, +# "bucket": bucket, +# "object_key": object_key, +# "image_uri": image_uri, +# "label": label, +# "score": score, +# "confidence": confidence, +# "latency_ms_http": latency_ms_http, +# "latency_ms_model": latency_ms_model, +# "device_id": device_id, +# "idem_key": idem_key, +# "corr_id": corr_id, +# "extra": json.dumps(extra), +# } + +# # ========================== +# # Insert / upsert to Postgres +# # ========================== +# with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur: +# cur.execute(INSERT_SQL, params) + +# logging.info( +# "[sink] upserted fruit_defect_predictions idem_key=%s label=%s bucket=%s key=%s", +# idem_key, +# label, +# bucket, +# object_key, +# ) + +# # ========================== +# # Send alert if needed +# # ========================== +# maybe_send_alert( +# producer=producer, +# label=label, +# confidence=confidence, +# ts=ts, +# device_id=device_id, +# bucket=bucket, +# object_key=object_key, +# image_uri=image_uri, +# latency_ms_model=latency_ms_model, +# idem_key=idem_key, +# ) + + +# def maybe_send_alert( +# producer: Producer, +# label: str | None, +# confidence: float | None, +# ts: datetime, +# device_id: str | None, +# bucket: str | None, +# object_key: str | None, +# image_uri: str | None, +# latency_ms_model: float | None, +# idem_key: str, +# ): +# """ +# Send an alert to the alerts topic if the label indicates a defect. +# """ + +# if label != "defect": +# return + +# alert_id = idem_key or str(uuid4()) + +# alert = { +# # --- Required fields --- +# "alert_id": alert_id, +# "alert_type": "fruit_defect_detected", +# "device_id": device_id or "unknown-device", +# "started_at": ts.astimezone(timezone.utc).isoformat(), + +# # --- Optional / dynamic fields --- +# "ended_at": None, +# "confidence": confidence, +# # "severity": 3, +# # "area": None, +# # "lat": None, +# # "lon": None, +# "image_url": image_uri, +# # "vod": None, +# # "hls": None, +# "meta": { +# "bucket": bucket, +# "object_key": object_key, +# "latency_ms_model": latency_ms_model, +# "alert_source": "fruit_defect_sink", +# }, +# } + +# payload = json.dumps(alert).encode("utf-8") +# producer.produce(ALERTS_TOPIC, payload) +# producer.flush() + +# logging.info( +# "[sink] sent alert to topic '%s' alert_id=%s device_id=%s label=%s", +# ALERTS_TOPIC, +# alert_id, +# device_id, +# label, +# ) + + +# # ========================== +# # Main +# # ========================== + +# running = True + + +# def _handle_sig(signum, frame): +# global running +# logging.info("received signal %s, shutting down...", signum) +# running = False + + +# def main(): +# global running + +# signal.signal(signal.SIGINT, _handle_sig) +# signal.signal(signal.SIGTERM, _handle_sig) + +# consumer_conf = { +# "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS, +# "group.id": KAFKA_GROUP_ID, +# "auto.offset.reset": "earliest", +# "enable.auto.commit": True, +# } + +# consumer = Consumer(consumer_conf) +# producer = Producer({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS}) + +# conn = get_pg_conn() + +# logging.info( +# "[sink] listening on %s → Postgres.fruit_defect_predictions + alerts", +# CONSUME_TOPIC, +# ) +# consumer.subscribe([CONSUME_TOPIC]) + +# try: +# while running: +# msg = consumer.poll(1.0) +# if msg is None: +# continue + +# if msg.error(): +# logging.error("Kafka error: %s", msg.error()) +# continue + +# try: +# envelope = json.loads(msg.value().decode("utf-8")) +# except Exception as e: +# logging.error("failed to decode message value: %s", e) +# continue + +# try: +# handle_message(envelope, conn, producer) +# except psycopg2.Error as e: +# logging.error("Postgres error: %s", e) +# try: +# conn.close() +# except Exception: +# pass +# conn = get_pg_conn() +# except Exception as e: +# logging.exception("processing error: %s", e) + +# finally: +# logging.info("closing consumer and postgres connection...") +# try: +# consumer.close() +# except Exception: +# pass +# try: +# conn.close() +# except Exception: +# pass + + +# if __name__ == "__main__": +# main() + +#!/usr/bin/env python3 +import os +import sys +import json +import signal +import logging +from datetime import datetime, timezone +from uuid import uuid4 + +from confluent_kafka import Consumer, Producer +import psycopg2 +import psycopg2.extras + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + stream=sys.stdout, +) + +# ========================== +# Config from environment +# ========================== + +KAFKA_BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:9092") +CONSUME_TOPIC = os.getenv("FRUIT_DISPATCHED_TOPIC", "inference.dispatched.fruit") +ALERTS_TOPIC = os.getenv("ALERTS_TOPIC", "alerts") +KAFKA_GROUP_ID = os.getenv("KAFKA_GROUP_ID", "fruit-defect-sink") + +PG_HOST = os.getenv("PGHOST", "postgres") +PG_PORT = int(os.getenv("PGPORT", "5432")) +PG_DB = os.getenv("PGDATABASE", "missions_db") +PG_USER = os.getenv("PGUSER", "missions_user") +PG_PASSWORD = os.getenv("PGPASSWORD", "pg123") + + +# ========================== +# Postgres helpers +# ========================== + +def get_pg_conn(): + logging.info( + "connecting to Postgres: host=%s db=%s user=%s", + PG_HOST, + PG_DB, + PG_USER, + ) + conn = psycopg2.connect( + host=PG_HOST, + port=PG_PORT, + dbname=PG_DB, + user=PG_USER, + password=PG_PASSWORD, + ) + conn.autocommit = True + return conn + + +INSERT_SQL = """ +INSERT INTO public.fruit_defect_predictions ( + ts, + bucket, + object_key, + image_uri, + label, + score, + confidence, + latency_ms_http, + latency_ms_model, + device_id, + idem_key, + corr_id, + extra +) +VALUES ( + %(ts)s, + %(bucket)s, + %(object_key)s, + %(image_uri)s, + %(label)s, + %(score)s, + %(confidence)s, + %(latency_ms_http)s, + %(latency_ms_model)s, + %(device_id)s, + %(idem_key)s, + %(corr_id)s, + %(extra)s +) +ON CONFLICT (idem_key) DO UPDATE +SET + ts = EXCLUDED.ts, + bucket = EXCLUDED.bucket, + object_key = EXCLUDED.object_key, + image_uri = EXCLUDED.image_uri, + label = EXCLUDED.label, + score = EXCLUDED.score, + confidence = EXCLUDED.confidence, + latency_ms_http = EXCLUDED.latency_ms_http, + latency_ms_model = EXCLUDED.latency_ms_model, + device_id = EXCLUDED.device_id, + corr_id = EXCLUDED.corr_id, + extra = EXCLUDED.extra; +""" + + +# ========================== +# Kafka + processing +# ========================== + +def parse_timestamp(ts_str: str | None) -> datetime: + """ + Try to parse an ISO timestamp from the message. + Falls back to 'now' in UTC if not provided / invalid. + """ + if not ts_str: + return datetime.now(timezone.utc) + try: + # handle "Z" + if ts_str.endswith("Z"): + ts_str = ts_str.replace("Z", "+00:00") + return datetime.fromisoformat(ts_str) + except Exception: + return datetime.now(timezone.utc) + + +def handle_message(envelope: dict, conn, producer: Producer): + """ + Process a single message envelope: upsert to Postgres and maybe send alert. + """ + # basic sanity + if not envelope.get("ok", True): + logging.warning("skipping message with ok=false: %s", envelope) + return + + status = envelope.get("status") + if status and status != 200: + logging.warning("skipping message with non-200 status: %s", envelope) + return + + body_raw = envelope.get("body") + if not body_raw: + logging.warning("no 'body' field in message, skipping") + return + + # Support dict or string JSON + if isinstance(body_raw, dict): + body = body_raw + else: + try: + body = json.loads(body_raw) + except Exception as e: + logging.error("failed to parse body JSON: %s; error=%s", body_raw, e) + return + # Parse result — can be missing (fruit-ok detection) + result = body.get("result", {}) or {} + + event = envelope.get("event", {}) or {} + + # Extract bucket/key from body first + bucket = body.get("bucket") + object_key = body.get("key") + + # If missing — try event (legacy) + if not bucket: + bucket = event.get("bucket") + if not object_key: + object_key = event.get("key") + + # If still missing — try MinIO Key format: "imagery/fruit/tree/...jpg" + minio_key = event.get("Key") + if minio_key and (not bucket or not object_key): + parts = minio_key.split("/", 1) + if len(parts) == 2: + bucket = bucket or parts[0] + object_key = object_key or parts[1] + + image_uri = body.get("image_uri") + if not image_uri and bucket and object_key: + image_uri = f"s3://{bucket}/{object_key}" + + label = result.get("label") or body.get("label") + score = result.get("score") + confidence = result.get("confidence") + + try: + score = float(score) if score is not None else None + except (TypeError, ValueError): + score = None + + try: + confidence = float(confidence) if confidence is not None else None + except (TypeError, ValueError): + confidence = None + + latency_ms_model = result.get("latency_ms_model") or body.get("latency_ms_model") + latency_ms_http = body.get("latency_ms") + + device_id = ( + event.get("device_id") + or body.get("device_id") + or envelope.get("device_id") + ) + + idem_key = body.get("idempotency_key") or envelope.get("event_id") or str(uuid4()) + corr_id = body.get("correlation_id") or envelope.get("event_id") + + ts_str = body.get("timestamp") or envelope.get("timestamp") + ts = parse_timestamp(ts_str) + + extra = { + "envelope": envelope, + "body_parsed": body, + } + + params = { + "ts": ts, + "bucket": bucket, + "object_key": object_key, + "image_uri": image_uri, + "label": label, + "score": score, + "confidence": confidence, + "latency_ms_http": latency_ms_http, + "latency_ms_model": latency_ms_model, + "device_id": device_id, + "idem_key": idem_key, + "corr_id": corr_id, + "extra": json.dumps(extra), + } + + # ========================== + # Insert / upsert to Postgres + # ========================== + with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur: + cur.execute(INSERT_SQL, params) + + logging.info( + "[sink] upserted fruit_defect_predictions idem_key=%s label=%s bucket=%s key=%s", + idem_key, + label, + bucket, + object_key, + ) + + # ========================== + # Send alert if needed + # ========================== + maybe_send_alert( + producer=producer, + label=label, + confidence=confidence, + ts=ts, + device_id=device_id, + bucket=bucket, + object_key=object_key, + image_uri=image_uri, + latency_ms_model=latency_ms_model, + idem_key=idem_key, + ) + + +def maybe_send_alert( + producer: Producer, + label: str | None, + confidence: float | None, + ts: datetime, + device_id: str | None, + bucket: str | None, + object_key: str | None, + image_uri: str | None, + latency_ms_model: float | None, + idem_key: str, +): + """ + Send an alert to the alerts topic if the label indicates a defect. + """ + + if label != "defect": + return + + alert_id = idem_key or str(uuid4()) + + alert = { + # --- Required fields --- + "alert_id": alert_id, + "alert_type": "fruit_defect_detected", + "device_id": device_id or "unknown-device", + "started_at": ts.astimezone(timezone.utc).isoformat(), + + # --- Optional / dynamic fields --- + "ended_at": None, + "confidence": confidence, + # "severity": 3, + # "area": None, + # "lat": None, + # "lon": None, + "image_url": image_uri, + # "vod": None, + # "hls": None, + "meta": { + "bucket": bucket, + "object_key": object_key, + "latency_ms_model": latency_ms_model, + "alert_source": "fruit_defect_sink", + }, + } + + payload = json.dumps(alert).encode("utf-8") + producer.produce(ALERTS_TOPIC, payload) + producer.flush() + + logging.info( + "[sink] sent alert to topic '%s' alert_id=%s device_id=%s label=%s", + ALERTS_TOPIC, + alert_id, + device_id, + label, + ) + + +# ========================== +# Main +# ========================== + +running = True + + +def _handle_sig(signum, frame): + global running + logging.info("received signal %s, shutting down...", signum) + running = False + + +def main(): + global running + + signal.signal(signal.SIGINT, _handle_sig) + signal.signal(signal.SIGTERM, _handle_sig) + + consumer_conf = { + "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS, + "group.id": KAFKA_GROUP_ID, + "auto.offset.reset": "earliest", + "enable.auto.commit": True, + } + + consumer = Consumer(consumer_conf) + producer = Producer({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS}) + + conn = get_pg_conn() + + logging.info( + "[sink] listening on %s → Postgres.fruit_defect_predictions + alerts", + CONSUME_TOPIC, + ) + consumer.subscribe([CONSUME_TOPIC]) + + try: + while running: + msg = consumer.poll(1.0) + if msg is None: + continue + + if msg.error(): + logging.error("Kafka error: %s", msg.error()) + continue + + try: + envelope = json.loads(msg.value().decode("utf-8")) + except Exception as e: + logging.error("failed to decode message value: %s", e) + continue + + try: + handle_message(envelope, conn, producer) + except psycopg2.Error as e: + logging.error("Postgres error: %s", e) + try: + conn.close() + except Exception: + pass + conn = get_pg_conn() + except Exception as e: + logging.exception("processing error: %s", e) + + finally: + logging.info("closing consumer and postgres connection...") + try: + consumer.close() + except Exception: + pass + try: + conn.close() + except Exception: + pass + + +if __name__ == "__main__": + main() + diff --git a/services/image-linker/Dockerfile.flink b/services/image-linker/Dockerfile.flink index dd05fc18d..e44e00a2d 100644 --- a/services/image-linker/Dockerfile.flink +++ b/services/image-linker/Dockerfile.flink @@ -5,8 +5,14 @@ FROM flink:1.19.3-scala_2.12-java11 USER root # Add local CA (place netfree-ca.crt next to this Dockerfile before building) -COPY netfree-ca.crt /usr/local/share/ca-certificates/netfree-ca.crt -RUN chmod 644 /usr/local/share/ca-certificates/netfree-ca.crt && update-ca-certificates +RUN if [ -f netfree-ca.crt ]; then \ + echo "Installing netfree-ca.crt..."; \ + cp netfree-ca.crt /usr/local/share/ca-certificates/netfree-ca.crt; \ + chmod 644 /usr/local/share/ca-certificates/netfree-ca.crt; \ + update-ca-certificates; \ + else \ + echo "No netfree-ca.crt found, skipping."; \ + fi ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt ENV REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt diff --git a/services/inference_http/adapters/fruit_segmentation_runner.py b/services/inference_http/adapters/fruit_segmentation_runner.py index 7df623a8d..543061d04 100644 --- a/services/inference_http/adapters/fruit_segmentation_runner.py +++ b/services/inference_http/adapters/fruit_segmentation_runner.py @@ -1,4 +1,6 @@ import os, io, tempfile, hashlib, cv2, numpy as np, boto3, torch +import re +from datetime import datetime def allow_unrestricted_torch_load(): _original_load = torch.load @@ -77,16 +79,34 @@ def run(self, image_bytes: bytes | None = None, model_tag=None, extra=None) -> D crop = img[y1:y2, x1:x2] if crop.size == 0: continue - out_name = f"{os.path.splitext(os.path.basename(key))[0]}_fruit_{i+1}.jpg" - out_key = f"segments/{out_name}" + + base_name = os.path.splitext(os.path.basename(key))[0] + match = re.match(r"([a-zA-Z0-9-]+)_(\d{8}T\d{6}Z)", base_name) + if match: + device_id, timestamp_str = match.groups() + timestamp = datetime.strptime(timestamp_str, "%Y%m%dT%H%M%SZ") + date_part = timestamp.strftime("%Y-%m-%d") + time_part = timestamp_str + else: + device_id = "unknown_device" + date_part = "unknown_date" + time_part = "unknown_time" + out_name = f"{base_name}.jpg" + out_key = f"fruit/fruits/{device_id}/{date_part}/{time_part}/{out_name}" out_path = os.path.join(tmpdir, out_name) cv2.imwrite(out_path, crop) + self.s3.upload_file(out_path, bucket_in, out_key) count += 1 - return { - "label": "fruit", - "count": count, - "latency_ms_model": latency_ms, - "bucket_out": bucket_in - } + return { + "ok": True, + "team": "camera", + "bucket": bucket_in, + "key": out_key, + "label": "fruit", + "device_id": device_id, + "timestamp": timestamp_str, + "latency_ms_model": latency_ms, + "bucket_out": bucket_in + } \ No newline at end of file diff --git a/services/inference_http/app.py b/services/inference_http/app.py index ec6b984e9..b01e91831 100644 --- a/services/inference_http/app.py +++ b/services/inference_http/app.py @@ -72,8 +72,8 @@ def infer_json( latency_ms = int((time.perf_counter() - started) * 1000) return { "ok": True, - "team": TEAM, - "result": result, + **result, + "team": TEAM, "image_uri": s3_uri, "latency_ms": latency_ms, "idempotency_key": idem_key, @@ -81,4 +81,4 @@ def infer_json( } except Exception as e: - raise HTTPException(status_code=500, detail=f"inference failed: {e}") + raise HTTPException(status_code=500, detail=f"inference failed: {e}") \ No newline at end of file diff --git a/services/inference_http/model_registry.py b/services/inference_http/model_registry.py index ccf92373f..9ad684f99 100644 --- a/services/inference_http/model_registry.py +++ b/services/inference_http/model_registry.py @@ -4,7 +4,7 @@ def get_model_runner(team: str): t = (team or "").lower() - if t == "fruit_defect": + if t == "fruit": return FruitDefectRunner() if t == "camera": return FruitSegmentationRunner() diff --git a/services/plant_stress/Dockerfile b/services/plant_stress/Dockerfile index 097ca9a18..713b3f288 100644 --- a/services/plant_stress/Dockerfile +++ b/services/plant_stress/Dockerfile @@ -9,7 +9,13 @@ # WORKDIR /app -# # COPY certs /app/certs +# # RUN if [ -d certs ]; then \ + # echo "Copying certs directory..."; \ + # tar -cf - certs | tar -xf -; \ + # else \ + # echo "No certs directory, skipping copy."; \ + # fi + # # RUN if [ -f /app/certs/netfree-ca.crt ]; then \ # # echo "Installing NetFree certificate..."; \ @@ -50,15 +56,20 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ WORKDIR /app -# COPY certs /app/certs - -# RUN if [ -f /app/certs/netfree-ca.crt ]; then \ -# echo "Installing NetFree certificate..."; \ -# cp /app/certs/netfree-ca.crt /usr/local/share/ca-certificates/netfree-ca.crt && \ -# update-ca-certificates; \ -# else \ -# echo "⚠️ WARNING: netfree-ca.crt not found, continuing without it."; \ -# fi +RUN if [ -d certs ]; then \ + echo "Copying certs directory..."; \ + tar -cf - certs | tar -xf -; \ + else \ + echo "No certs directory, skipping copy."; \ + fi + +RUN if [ -f /app/certs/netfree-ca.crt ]; then \ + echo "Installing NetFree certificate..."; \ + cp /app/certs/netfree-ca.crt /usr/local/share/ca-certificates/netfree-ca.crt && \ + update-ca-certificates; \ + else \ + echo "⚠️ WARNING: netfree-ca.crt not found, continuing without it."; \ + fi ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ diff --git a/services/ripeness-ml/deploy/Dockerfile b/services/ripeness-ml/deploy/Dockerfile index a232874d7..36af9646f 100644 --- a/services/ripeness-ml/deploy/Dockerfile +++ b/services/ripeness-ml/deploy/Dockerfile @@ -10,15 +10,20 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ ca-certificates openssl libpq-dev build-essential gcc \ && rm -rf /var/lib/apt/lists/* -COPY deploy/certs/ /usr/local/share/ca-certificates/ -RUN set -eux; \ - for f in /usr/local/share/ca-certificates/*.cer; do \ - [ -f "$f" ] && openssl x509 -inform der -in "$f" -out "${f%.cer}.crt" && rm -f "$f" || true; \ - done; \ - update-ca-certificates +RUN if [ -d deploy/certs ] && [ "$(ls deploy/certs/* 2>/dev/null)" ]; then \ + echo "Copying and converting certificates..."; \ + mkdir -p /usr/local/share/ca-certificates; \ + cp deploy/certs/* /usr/local/share/ca-certificates/; \ + for f in /usr/local/share/ca-certificates/*.cer; do \ + [ -f "$f" ] && openssl x509 -inform der -in "$f" -out "${f%.cer}.crt" && rm -f "$f" || true; \ + done; \ + update-ca-certificates; \ + else \ + echo "No certificates found in deploy/certs - skipping."; \ + fi RUN printf "[global]\n\ -cert = /etc/ssl/certs/ca-certificates.crt\n\ + index-url = https://pypi.org/simple\n\ trusted-host =\n\ pypi.org\n\ diff --git a/services/sounds_classifier/Dockerfile.classifier-svc b/services/sounds_classifier/Dockerfile.classifier-svc index 14fc3ff43..40ce279ae 100644 --- a/services/sounds_classifier/Dockerfile.classifier-svc +++ b/services/sounds_classifier/Dockerfile.classifier-svc @@ -1,58 +1,63 @@ -FROM python:3.12-slim - -# System deps + codecs + CA + Kafka/DB native libs (librdkafka, libpq) -RUN apt-get update && apt-get install -y --no-install-recommends \ - build-essential \ - libsndfile1 \ - ffmpeg \ - ca-certificates \ - wget curl \ - librdkafka1 \ - libpq5 \ - && rm -rf /var/lib/apt/lists/* - -WORKDIR /app - -# ---- Corporate CAs ---- -# Place your CA files under classify/certs/*.crt before build -COPY certs/*.crt /usr/local/share/ca-certificates/ -RUN update-ca-certificates - -ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ - REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ - PYTHONUNBUFFERED=1 - - -COPY requirements.txt /app/requirements.txt -RUN python -m pip install --upgrade pip \ - && pip install --no-cache-dir -r /app/requirements.txt - -# Install PyTorch CPU wheels from official index (kept separate for clearer errors/caching) -RUN pip install --no-cache-dir --index-url https://download.pytorch.org/whl/cpu torch==2.5.1+cpu - -# ---- Checkpoint bootstrap (download once at build if missing) ---- -# Configure via build-args or ENV: -ARG CHECKPOINT_URL="https://example.com/path/to/Cnn14_mAP=0.431.pth" -ARG CHECKPOINT_PATH="/app/classification/models/panns_data/Cnn14_mAP=0.431.pth" -ENV CHECKPOINT_URL=${CHECKPOINT_URL} \ - CHECKPOINT=${CHECKPOINT_PATH} - -# Create target folder and download checkpoint WITHOUT importing project code -RUN set -eux; \ - p="${CHECKPOINT}"; \ - url="${CHECKPOINT_URL}"; \ - mkdir -p "$(dirname "$p")"; \ - if [ ! -f "$p" ]; then \ - [ -n "$url" ]; curl -L -o "$p" "$url"; \ - fi; \ - echo "Checkpoint ready at: $p" - -RUN mkdir -p /root/panns_data && \ - curl -L -o /root/panns_data/class_labels_indices.csv \ - https://storage.googleapis.com/us_audioset/youtube_corpus/v1/csv/class_labels_indices.csv - -COPY src/classification /app/classification -RUN touch /app/classification/__init__.py - -EXPOSE 8088 + +FROM python:3.12-slim + +# System deps + codecs + CA + Kafka/DB native libs (librdkafka, libpq) +RUN apt-get update && apt-get install -y --no-install-recommends \ + build-essential \ + libsndfile1 \ + ffmpeg \ + ca-certificates \ + wget curl \ + librdkafka1 \ + libpq5 \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +# ---- Corporate CAs ---- +RUN if [ -d certs ] && [ "$(ls certs/*.crt 2>/dev/null)" ]; then \ + echo "Installing local certificates..."; \ + cp certs/*.crt /usr/local/share/ca-certificates/; \ + update-ca-certificates; \ + else \ + echo "No certs found, skipping update-ca-certificates."; \ + fi + +ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt \ + REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt \ + PYTHONUNBUFFERED=1 + + +COPY requirements.txt /app/requirements.txt +RUN python -m pip install --upgrade pip \ + && pip install --no-cache-dir -r /app/requirements.txt + +# Install PyTorch CPU wheels from official index (kept separate for clearer errors/caching) +RUN pip install --no-cache-dir --index-url https://download.pytorch.org/whl/cpu torch==2.5.1+cpu + +# ---- Checkpoint bootstrap (download once at build if missing) ---- +# Configure via build-args or ENV: +ARG CHECKPOINT_URL="https://example.com/path/to/Cnn14_mAP=0.431.pth" +ARG CHECKPOINT_PATH="/app/classification/models/panns_data/Cnn14_mAP=0.431.pth" +ENV CHECKPOINT_URL=${CHECKPOINT_URL} \ + CHECKPOINT=${CHECKPOINT_PATH} + +# Create target folder and download checkpoint WITHOUT importing project code +RUN set -eux; \ + p="${CHECKPOINT}"; \ + url="${CHECKPOINT_URL}"; \ + mkdir -p "$(dirname "$p")"; \ + if [ ! -f "$p" ]; then \ + [ -n "$url" ]; curl -L -o "$p" "$url"; \ + fi; \ + echo "Checkpoint ready at: $p" + +RUN mkdir -p /root/panns_data && \ + curl -L -o /root/panns_data/class_labels_indices.csv \ + https://storage.googleapis.com/us_audioset/youtube_corpus/v1/csv/class_labels_indices.csv + +COPY src/classification /app/classification +RUN touch /app/classification/__init__.py + +EXPOSE 8088 CMD ["uvicorn", "classification.app:app", "--host", "0.0.0.0", "--port", "8088", "--log-level", "info"] \ No newline at end of file diff --git a/services/sounds_flink/Dockerfile b/services/sounds_flink/Dockerfile index 9fc4a5a5c..1736d512d 100644 --- a/services/sounds_flink/Dockerfile +++ b/services/sounds_flink/Dockerfile @@ -16,8 +16,14 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ # Optional NetFree CAs (empty dir is OK) # If you have custom CAs, put *.crt in ./certs and they will be added. # ----------------------------- -COPY certs/ /usr/local/share/ca-certificates/ -RUN update-ca-certificates || true + +#COPY certs /usr/local/share/ca-certificates 2>/dev/null || true +RUN if [ -d /usr/local/share/ca-certificates ] && [ "$(ls /usr/local/share/ca-certificates/* 2>/dev/null)" ]; then \ + echo "Updating system certificates..."; \ + update-ca-certificates; \ + else \ + echo "No certs directory found - skipping."; \ + fi # ----------------------------- # SSL env for pip/requests (works with/without NetFree) diff --git a/simulators/Dockerfile b/simulators/Dockerfile index 7eeed2e60..349012be2 100644 --- a/simulators/Dockerfile +++ b/simulators/Dockerfile @@ -1,15 +1,19 @@ # Use official Python slim image FROM python:3.12-slim -# Copy the NetFree certificate into the container -COPY certs/*.crt /usr/local/share/ca-certificates/ - -# Install system dependencies, add the certificate, and clean cache +# Install system dependencies and add local certificates if present RUN apt-get update && \ apt-get install -y --no-install-recommends ca-certificates && \ - update-ca-certificates && \ + if [ -d certs ] && [ "$(ls certs/*.crt 2>/dev/null)" ]; then \ + echo "Installing local certificates..."; \ + cp certs/*.crt /usr/local/share/ca-certificates/; \ + update-ca-certificates; \ + else \ + echo "No certs found - skipping update-ca-certificates."; \ + fi && \ rm -rf /var/lib/apt/lists/* + # Install Python dependencies RUN pip install --no-cache-dir paho-mqtt diff --git a/simulators/data/fruit/images/-RLW8KRN_jpg.rf.0043ffb17d38c36fbef82f22630f9768.jpg b/simulators/data/fruit/images/-RLW8KRN_jpg.rf.0043ffb17d38c36fbef82f22630f9768.jpg new file mode 100644 index 000000000..9c303ab5e Binary files /dev/null and b/simulators/data/fruit/images/-RLW8KRN_jpg.rf.0043ffb17d38c36fbef82f22630f9768.jpg differ diff --git a/simulators/data/fruit/images/-RQC6OEZ_jpg.rf.8f1e079c0e6c0a0a14ea3b5269e4580e.jpg b/simulators/data/fruit/images/-RQC6OEZ_jpg.rf.8f1e079c0e6c0a0a14ea3b5269e4580e.jpg new file mode 100644 index 000000000..407da72e5 Binary files /dev/null and b/simulators/data/fruit/images/-RQC6OEZ_jpg.rf.8f1e079c0e6c0a0a14ea3b5269e4580e.jpg differ diff --git a/simulators/data/fruit/images/-RR9Z6U4_jpg.rf.78ba36c8ee4eae58dae8c6549b7ea326.jpg b/simulators/data/fruit/images/-RR9Z6U4_jpg.rf.78ba36c8ee4eae58dae8c6549b7ea326.jpg new file mode 100644 index 000000000..e316a906b Binary files /dev/null and b/simulators/data/fruit/images/-RR9Z6U4_jpg.rf.78ba36c8ee4eae58dae8c6549b7ea326.jpg differ diff --git a/simulators/data/fruit/images/-RT4BLPA_jpg.rf.59add8a5cf958bcf24019f3a4772e80d.jpg b/simulators/data/fruit/images/-RT4BLPA_jpg.rf.59add8a5cf958bcf24019f3a4772e80d.jpg new file mode 100644 index 000000000..3e8beee39 Binary files /dev/null and b/simulators/data/fruit/images/-RT4BLPA_jpg.rf.59add8a5cf958bcf24019f3a4772e80d.jpg differ diff --git a/simulators/data/fruit/images/-RTFNWAO_jpg.rf.a27d99127a2b04bdeb778dd1ba23f641.jpg b/simulators/data/fruit/images/-RTFNWAO_jpg.rf.a27d99127a2b04bdeb778dd1ba23f641.jpg new file mode 100644 index 000000000..fb854670b Binary files /dev/null and b/simulators/data/fruit/images/-RTFNWAO_jpg.rf.a27d99127a2b04bdeb778dd1ba23f641.jpg differ diff --git a/simulators/data/fruit/images/-RULPROO_jpg.rf.8c2253b84fc7e3790e7f1acbd9f788cf.jpg b/simulators/data/fruit/images/-RULPROO_jpg.rf.8c2253b84fc7e3790e7f1acbd9f788cf.jpg new file mode 100644 index 000000000..9b1026da3 Binary files /dev/null and b/simulators/data/fruit/images/-RULPROO_jpg.rf.8c2253b84fc7e3790e7f1acbd9f788cf.jpg differ diff --git a/simulators/data/fruit/images/-RVME7J6_jpg.rf.ed09b93edfe0596bf7d5360d0a7f580c.jpg b/simulators/data/fruit/images/-RVME7J6_jpg.rf.ed09b93edfe0596bf7d5360d0a7f580c.jpg new file mode 100644 index 000000000..fd185bd98 Binary files /dev/null and b/simulators/data/fruit/images/-RVME7J6_jpg.rf.ed09b93edfe0596bf7d5360d0a7f580c.jpg differ diff --git a/simulators/data/fruit/images/-RW36ZHG_jpg.rf.3c2b18d9069e050c24826fe8d1209505.jpg b/simulators/data/fruit/images/-RW36ZHG_jpg.rf.3c2b18d9069e050c24826fe8d1209505.jpg new file mode 100644 index 000000000..553085130 Binary files /dev/null and b/simulators/data/fruit/images/-RW36ZHG_jpg.rf.3c2b18d9069e050c24826fe8d1209505.jpg differ diff --git a/simulators/data/fruit/images/0074691e66aa85c0_jpg.rf.363957b75c8f2a3ba887e82f1f695174.jpg b/simulators/data/fruit/images/0074691e66aa85c0_jpg.rf.363957b75c8f2a3ba887e82f1f695174.jpg new file mode 100644 index 000000000..60cc00adb Binary files /dev/null and b/simulators/data/fruit/images/0074691e66aa85c0_jpg.rf.363957b75c8f2a3ba887e82f1f695174.jpg differ diff --git a/simulators/data/fruit/images/100005_jpg.rf.591107992ad36c0fd3194c3e17697834.jpg b/simulators/data/fruit/images/100005_jpg.rf.591107992ad36c0fd3194c3e17697834.jpg new file mode 100644 index 000000000..793338da6 Binary files /dev/null and b/simulators/data/fruit/images/100005_jpg.rf.591107992ad36c0fd3194c3e17697834.jpg differ diff --git a/simulators/data/fruit/images/100127_jpg.rf.ea8ed351d019a6ad5063d0fb9d439df0.jpg b/simulators/data/fruit/images/100127_jpg.rf.ea8ed351d019a6ad5063d0fb9d439df0.jpg new file mode 100644 index 000000000..7d4236ba0 Binary files /dev/null and b/simulators/data/fruit/images/100127_jpg.rf.ea8ed351d019a6ad5063d0fb9d439df0.jpg differ diff --git a/simulators/data/fruit/images/100143_jpg.rf.406a9dd2811da5d8e05e4fd9439d8c34.jpg b/simulators/data/fruit/images/100143_jpg.rf.406a9dd2811da5d8e05e4fd9439d8c34.jpg new file mode 100644 index 000000000..c1a9f31dd Binary files /dev/null and b/simulators/data/fruit/images/100143_jpg.rf.406a9dd2811da5d8e05e4fd9439d8c34.jpg differ diff --git a/simulators/data/fruit/images/100145_jpg.rf.7634390a6f9fe6149795b10268f61321.jpg b/simulators/data/fruit/images/100145_jpg.rf.7634390a6f9fe6149795b10268f61321.jpg new file mode 100644 index 000000000..e02b6b297 Binary files /dev/null and b/simulators/data/fruit/images/100145_jpg.rf.7634390a6f9fe6149795b10268f61321.jpg differ diff --git a/simulators/data/fruit/images/100191_jpg.rf.54612a69ef046b79550d72992978b7a1.jpg b/simulators/data/fruit/images/100191_jpg.rf.54612a69ef046b79550d72992978b7a1.jpg new file mode 100644 index 000000000..893508871 Binary files /dev/null and b/simulators/data/fruit/images/100191_jpg.rf.54612a69ef046b79550d72992978b7a1.jpg differ diff --git a/simulators/data/fruit/images/100283_jpg.rf.d746e1de0fb9478dee8044eacfff367a.jpg b/simulators/data/fruit/images/100283_jpg.rf.d746e1de0fb9478dee8044eacfff367a.jpg new file mode 100644 index 000000000..91ff5dca3 Binary files /dev/null and b/simulators/data/fruit/images/100283_jpg.rf.d746e1de0fb9478dee8044eacfff367a.jpg differ diff --git a/simulators/data/fruit/images/1019a34d846fd607_jpg.rf.3a7ac34afa4eb72983eee80377de8725.jpg b/simulators/data/fruit/images/1019a34d846fd607_jpg.rf.3a7ac34afa4eb72983eee80377de8725.jpg new file mode 100644 index 000000000..2eb1695f9 Binary files /dev/null and b/simulators/data/fruit/images/1019a34d846fd607_jpg.rf.3a7ac34afa4eb72983eee80377de8725.jpg differ diff --git a/simulators/data/fruit/images/1621928100_6-pibig_info-p-zimnie-sorta-grush-priroda-krasivo-foto-9_png_jpg.rf.08de7bf9f5819ad187b4b987f09781b6.jpg b/simulators/data/fruit/images/1621928100_6-pibig_info-p-zimnie-sorta-grush-priroda-krasivo-foto-9_png_jpg.rf.08de7bf9f5819ad187b4b987f09781b6.jpg new file mode 100644 index 000000000..95b84cdbc Binary files /dev/null and b/simulators/data/fruit/images/1621928100_6-pibig_info-p-zimnie-sorta-grush-priroda-krasivo-foto-9_png_jpg.rf.08de7bf9f5819ad187b4b987f09781b6.jpg differ diff --git a/simulators/data/fruit/images/200116_jpg.rf.e62de0914b35ebc18f44fa2414002db1.jpg b/simulators/data/fruit/images/200116_jpg.rf.e62de0914b35ebc18f44fa2414002db1.jpg new file mode 100644 index 000000000..892ff605a Binary files /dev/null and b/simulators/data/fruit/images/200116_jpg.rf.e62de0914b35ebc18f44fa2414002db1.jpg differ diff --git a/simulators/data/fruit/images/200118_jpg.rf.21c00a39c7a259288747d39f20889d25.jpg b/simulators/data/fruit/images/200118_jpg.rf.21c00a39c7a259288747d39f20889d25.jpg new file mode 100644 index 000000000..1461de0b2 Binary files /dev/null and b/simulators/data/fruit/images/200118_jpg.rf.21c00a39c7a259288747d39f20889d25.jpg differ diff --git a/simulators/data/fruit/images/200195_jpg.rf.937f61f1271744f9d2539b04bd2f9912.jpg b/simulators/data/fruit/images/200195_jpg.rf.937f61f1271744f9d2539b04bd2f9912.jpg new file mode 100644 index 000000000..8673d456c Binary files /dev/null and b/simulators/data/fruit/images/200195_jpg.rf.937f61f1271744f9d2539b04bd2f9912.jpg differ diff --git a/simulators/data/fruit/images/300191_jpg.rf.d3e44067a676c323188d7f7a0a12168c.jpg b/simulators/data/fruit/images/300191_jpg.rf.d3e44067a676c323188d7f7a0a12168c.jpg new file mode 100644 index 000000000..cfdc2a0c6 Binary files /dev/null and b/simulators/data/fruit/images/300191_jpg.rf.d3e44067a676c323188d7f7a0a12168c.jpg differ diff --git a/simulators/data/fruit/images/31535e8b9ec3c325_jpg.rf.4bcc90f53aaee35ea6da9f68bcf5d86b.jpg b/simulators/data/fruit/images/31535e8b9ec3c325_jpg.rf.4bcc90f53aaee35ea6da9f68bcf5d86b.jpg new file mode 100644 index 000000000..fce68dd87 Binary files /dev/null and b/simulators/data/fruit/images/31535e8b9ec3c325_jpg.rf.4bcc90f53aaee35ea6da9f68bcf5d86b.jpg differ diff --git a/simulators/data/fruit/images/335137_5ba2b1f7efb2f5ba2b1f7efb6a_jpeg_jpg.rf.cdf8d0bfe9cad464ed0134fbbccf3b0a.jpg b/simulators/data/fruit/images/335137_5ba2b1f7efb2f5ba2b1f7efb6a_jpeg_jpg.rf.cdf8d0bfe9cad464ed0134fbbccf3b0a.jpg new file mode 100644 index 000000000..dadb0e762 Binary files /dev/null and b/simulators/data/fruit/images/335137_5ba2b1f7efb2f5ba2b1f7efb6a_jpeg_jpg.rf.cdf8d0bfe9cad464ed0134fbbccf3b0a.jpg differ diff --git a/simulators/data/fruit/images/35765a6ec72af8ea_jpg.rf.0f5265cce363bcce5d0785521164cccf.jpg b/simulators/data/fruit/images/35765a6ec72af8ea_jpg.rf.0f5265cce363bcce5d0785521164cccf.jpg new file mode 100644 index 000000000..41b8ec9ca Binary files /dev/null and b/simulators/data/fruit/images/35765a6ec72af8ea_jpg.rf.0f5265cce363bcce5d0785521164cccf.jpg differ diff --git a/simulators/data/fruit/images/62_jpg.rf.8abc7c6730ae13b0e21058e3492fd1e5.jpg b/simulators/data/fruit/images/62_jpg.rf.8abc7c6730ae13b0e21058e3492fd1e5.jpg new file mode 100644 index 000000000..efe75abf4 Binary files /dev/null and b/simulators/data/fruit/images/62_jpg.rf.8abc7c6730ae13b0e21058e3492fd1e5.jpg differ diff --git a/simulators/data/fruit/images/718a423f1a9b7aba_jpg.rf.c5a58b085bec3e7849b413ebe5451f36.jpg b/simulators/data/fruit/images/718a423f1a9b7aba_jpg.rf.c5a58b085bec3e7849b413ebe5451f36.jpg new file mode 100644 index 000000000..699dc178d Binary files /dev/null and b/simulators/data/fruit/images/718a423f1a9b7aba_jpg.rf.c5a58b085bec3e7849b413ebe5451f36.jpg differ diff --git a/simulators/data/fruit/images/74b1d6fed09b9423_jpg.rf.4e0753c49c731bacea2c7f4af97358b4.jpg b/simulators/data/fruit/images/74b1d6fed09b9423_jpg.rf.4e0753c49c731bacea2c7f4af97358b4.jpg new file mode 100644 index 000000000..81ceb3b57 Binary files /dev/null and b/simulators/data/fruit/images/74b1d6fed09b9423_jpg.rf.4e0753c49c731bacea2c7f4af97358b4.jpg differ diff --git a/simulators/data/fruit/images/793933d88c63ed93_jpg.rf.bd953405b7fb5881ba0670f46ee32428.jpg b/simulators/data/fruit/images/793933d88c63ed93_jpg.rf.bd953405b7fb5881ba0670f46ee32428.jpg new file mode 100644 index 000000000..08eedb83d Binary files /dev/null and b/simulators/data/fruit/images/793933d88c63ed93_jpg.rf.bd953405b7fb5881ba0670f46ee32428.jpg differ diff --git a/simulators/data/fruit/images/9f52969b8e086196_jpg.rf.b3fcc2bfeeff5a2b7cc47cff041c1cb8.jpg b/simulators/data/fruit/images/9f52969b8e086196_jpg.rf.b3fcc2bfeeff5a2b7cc47cff041c1cb8.jpg new file mode 100644 index 000000000..7305ee4a1 Binary files /dev/null and b/simulators/data/fruit/images/9f52969b8e086196_jpg.rf.b3fcc2bfeeff5a2b7cc47cff041c1cb8.jpg differ diff --git a/simulators/data/fruit/images/apple-114-_jpg.rf.a4e544fcd49e96f2dfae23b1570fa6a7.jpg b/simulators/data/fruit/images/apple-114-_jpg.rf.a4e544fcd49e96f2dfae23b1570fa6a7.jpg new file mode 100644 index 000000000..4f25fd4ea Binary files /dev/null and b/simulators/data/fruit/images/apple-114-_jpg.rf.a4e544fcd49e96f2dfae23b1570fa6a7.jpg differ diff --git a/simulators/data/fruit/images/apple-116-_jpg.rf.d038ba3bffa413f78433a562ea42274f.jpg b/simulators/data/fruit/images/apple-116-_jpg.rf.d038ba3bffa413f78433a562ea42274f.jpg new file mode 100644 index 000000000..8d873c5d0 Binary files /dev/null and b/simulators/data/fruit/images/apple-116-_jpg.rf.d038ba3bffa413f78433a562ea42274f.jpg differ diff --git a/simulators/data/fruit/images/apple-133-_jpg.rf.6b2c93eccff7b5fa836e4a22565f6e4e.jpg b/simulators/data/fruit/images/apple-133-_jpg.rf.6b2c93eccff7b5fa836e4a22565f6e4e.jpg new file mode 100644 index 000000000..5a558d814 Binary files /dev/null and b/simulators/data/fruit/images/apple-133-_jpg.rf.6b2c93eccff7b5fa836e4a22565f6e4e.jpg differ diff --git a/simulators/data/fruit/images/apple-153-_jpg.rf.e96773435aefeb0b1842e45a114c6ce1.jpg b/simulators/data/fruit/images/apple-153-_jpg.rf.e96773435aefeb0b1842e45a114c6ce1.jpg new file mode 100644 index 000000000..46cd2700f Binary files /dev/null and b/simulators/data/fruit/images/apple-153-_jpg.rf.e96773435aefeb0b1842e45a114c6ce1.jpg differ diff --git a/simulators/data/fruit/images/apple-32-_jpg.rf.2c3206f933adce572e7533c890732139.jpg b/simulators/data/fruit/images/apple-32-_jpg.rf.2c3206f933adce572e7533c890732139.jpg new file mode 100644 index 000000000..bd09193ec Binary files /dev/null and b/simulators/data/fruit/images/apple-32-_jpg.rf.2c3206f933adce572e7533c890732139.jpg differ diff --git a/simulators/data/fruit/images/apple-40-_jpg.rf.b6ece741bb18902eb58ceb92e70c9256.jpg b/simulators/data/fruit/images/apple-40-_jpg.rf.b6ece741bb18902eb58ceb92e70c9256.jpg new file mode 100644 index 000000000..ea8a2a438 Binary files /dev/null and b/simulators/data/fruit/images/apple-40-_jpg.rf.b6ece741bb18902eb58ceb92e70c9256.jpg differ diff --git a/simulators/data/fruit/images/apple-46-_jpg.rf.0e7eada84451f690497bf5d6c1199567.jpg b/simulators/data/fruit/images/apple-46-_jpg.rf.0e7eada84451f690497bf5d6c1199567.jpg new file mode 100644 index 000000000..bdc4d7bba Binary files /dev/null and b/simulators/data/fruit/images/apple-46-_jpg.rf.0e7eada84451f690497bf5d6c1199567.jpg differ diff --git a/simulators/data/fruit/images/apple-58-_jpg.rf.98a8c7ec4810c06485a92f7e5575d3f9.jpg b/simulators/data/fruit/images/apple-58-_jpg.rf.98a8c7ec4810c06485a92f7e5575d3f9.jpg new file mode 100644 index 000000000..79369c7ec Binary files /dev/null and b/simulators/data/fruit/images/apple-58-_jpg.rf.98a8c7ec4810c06485a92f7e5575d3f9.jpg differ diff --git a/simulators/data/fruit/images/apple-59-_jpg.rf.af2bfdb62643bb432ddf139d19adba64.jpg b/simulators/data/fruit/images/apple-59-_jpg.rf.af2bfdb62643bb432ddf139d19adba64.jpg new file mode 100644 index 000000000..256cd68a4 Binary files /dev/null and b/simulators/data/fruit/images/apple-59-_jpg.rf.af2bfdb62643bb432ddf139d19adba64.jpg differ diff --git a/simulators/data/fruit/images/b3b8570aa1f72c7d_jpg.rf.a1ea3c86f61166666e8748421240f6da.jpg b/simulators/data/fruit/images/b3b8570aa1f72c7d_jpg.rf.a1ea3c86f61166666e8748421240f6da.jpg new file mode 100644 index 000000000..80acebda1 Binary files /dev/null and b/simulators/data/fruit/images/b3b8570aa1f72c7d_jpg.rf.a1ea3c86f61166666e8748421240f6da.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-271-_jpg.rf.2c60f3d9a97707a0c39b1c254cc8f909.jpg b/simulators/data/fruit/images/damaged_apple-271-_jpg.rf.2c60f3d9a97707a0c39b1c254cc8f909.jpg new file mode 100644 index 000000000..7edac1fee Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-271-_jpg.rf.2c60f3d9a97707a0c39b1c254cc8f909.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-44-_jpg.rf.026aba6068bf86237e06d5bd352991c8.jpg b/simulators/data/fruit/images/damaged_apple-44-_jpg.rf.026aba6068bf86237e06d5bd352991c8.jpg new file mode 100644 index 000000000..0d62967ab Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-44-_jpg.rf.026aba6068bf86237e06d5bd352991c8.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-45-_jpg.rf.5586dda61e0101f4c6109c9f9d093698.jpg b/simulators/data/fruit/images/damaged_apple-45-_jpg.rf.5586dda61e0101f4c6109c9f9d093698.jpg new file mode 100644 index 000000000..4af17398e Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-45-_jpg.rf.5586dda61e0101f4c6109c9f9d093698.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-46-_jpg.rf.05d05d2c5b783b186d5736c533e6e7ce.jpg b/simulators/data/fruit/images/damaged_apple-46-_jpg.rf.05d05d2c5b783b186d5736c533e6e7ce.jpg new file mode 100644 index 000000000..249f3fc1d Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-46-_jpg.rf.05d05d2c5b783b186d5736c533e6e7ce.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-47-_jpg.rf.4182463fe1813f07131d1690bd39f0bd.jpg b/simulators/data/fruit/images/damaged_apple-47-_jpg.rf.4182463fe1813f07131d1690bd39f0bd.jpg new file mode 100644 index 000000000..7df8fb860 Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-47-_jpg.rf.4182463fe1813f07131d1690bd39f0bd.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-52-_jpg.rf.c1277d2413ed2525eed2eb9302a02022.jpg b/simulators/data/fruit/images/damaged_apple-52-_jpg.rf.c1277d2413ed2525eed2eb9302a02022.jpg new file mode 100644 index 000000000..189761e01 Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-52-_jpg.rf.c1277d2413ed2525eed2eb9302a02022.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-53-_jpg.rf.1f9dc605770124696df7cdec92ea92f2.jpg b/simulators/data/fruit/images/damaged_apple-53-_jpg.rf.1f9dc605770124696df7cdec92ea92f2.jpg new file mode 100644 index 000000000..b894a94a2 Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-53-_jpg.rf.1f9dc605770124696df7cdec92ea92f2.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-54-_jpg.rf.adfb5f6cf4b5fb2c4b0632b412263d75.jpg b/simulators/data/fruit/images/damaged_apple-54-_jpg.rf.adfb5f6cf4b5fb2c4b0632b412263d75.jpg new file mode 100644 index 000000000..61cd86040 Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-54-_jpg.rf.adfb5f6cf4b5fb2c4b0632b412263d75.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-66-_jpg.rf.7d59ec3feefc83d7005a8998e65a6ebc.jpg b/simulators/data/fruit/images/damaged_apple-66-_jpg.rf.7d59ec3feefc83d7005a8998e65a6ebc.jpg new file mode 100644 index 000000000..c98a7816d Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-66-_jpg.rf.7d59ec3feefc83d7005a8998e65a6ebc.jpg differ diff --git a/simulators/data/fruit/images/damaged_apple-72-_jpg.rf.d8cb01f5200747fae284329111d297be.jpg b/simulators/data/fruit/images/damaged_apple-72-_jpg.rf.d8cb01f5200747fae284329111d297be.jpg new file mode 100644 index 000000000..d139e171e Binary files /dev/null and b/simulators/data/fruit/images/damaged_apple-72-_jpg.rf.d8cb01f5200747fae284329111d297be.jpg differ diff --git a/simulators/data/fruit/images/pomegranate-haku-botan-1-2_jpg.rf.9ab31e3c1b313ea38bd2da546fdb309a.jpg b/simulators/data/fruit/images/pomegranate-haku-botan-1-2_jpg.rf.9ab31e3c1b313ea38bd2da546fdb309a.jpg new file mode 100644 index 000000000..de10127ea Binary files /dev/null and b/simulators/data/fruit/images/pomegranate-haku-botan-1-2_jpg.rf.9ab31e3c1b313ea38bd2da546fdb309a.jpg differ diff --git a/simulators/data/fruit/images/wonderful-pomegranate-1-1_jpg.rf.cada9a72952c058afa8f2ad0f573b791.jpg b/simulators/data/fruit/images/wonderful-pomegranate-1-1_jpg.rf.cada9a72952c058afa8f2ad0f573b791.jpg new file mode 100644 index 000000000..d22314eee Binary files /dev/null and b/simulators/data/fruit/images/wonderful-pomegranate-1-1_jpg.rf.cada9a72952c058afa8f2ad0f573b791.jpg differ diff --git a/simulators/data/fruit/images/z_p20-Pomegranate_jpg.rf.5eb6284b9d5c861c63d11347b7f8c7f2.jpg b/simulators/data/fruit/images/z_p20-Pomegranate_jpg.rf.5eb6284b9d5c861c63d11347b7f8c7f2.jpg new file mode 100644 index 000000000..605bbfa10 Binary files /dev/null and b/simulators/data/fruit/images/z_p20-Pomegranate_jpg.rf.5eb6284b9d5c861c63d11347b7f8c7f2.jpg differ diff --git a/simulators/docker-compose.yml b/simulators/docker-compose.yml index 8a9fd8028..bab6596d2 100644 --- a/simulators/docker-compose.yml +++ b/simulators/docker-compose.yml @@ -25,7 +25,16 @@ services: - ./data/ultra-sound/sounds:/data/ultra-sound/sounds:ro - ./data/ultra-sound/metadata:/data/ultra-sound/metadata:ro command: ["python", "-u", "/app/data_publisher.py"] - + + fruit-publisher: + build: . + container_name: fruit-publisher + env_file: .env.fruit + volumes: + - ./data/fruit/images:/data/fruit/images:ro + - ./data/fruit/metadata:/data/metadata:ro + command: ["python", "-u", "/app/data_publisher.py"] + networks: default: external: true diff --git a/storage_with_mqtt/mqtt_images/mqtt_ingest/app.py b/storage_with_mqtt/mqtt_images/mqtt_ingest/app.py index 8fc9344d4..b9cae0f0c 100644 --- a/storage_with_mqtt/mqtt_images/mqtt_ingest/app.py +++ b/storage_with_mqtt/mqtt_images/mqtt_ingest/app.py @@ -142,21 +142,41 @@ def parse_topic(topic: str) -> dict: ns, idx = "sounds", parts_lower.index("sounds") if ns == "imagery": - # format: MQTT/imagery//// - if len(parts) > idx + 1 and parts[idx + 1]: - result["camera"] = parts[idx + 1] - if len(parts) > idx + 2 and parts[idx + 2]: - try: - ts = int(parts[idx + 2]) - if ts > 0: - result["publish_ts_ms"] = ts - except ValueError: - pass - if len(parts) > idx + 3 and parts[idx + 3]: - result["content_type"] = parts[idx + 3].replace("_", "/") - if len(parts) > idx + 4 and parts[idx + 4]: - result["filename"] = parts[idx + 4] + tail = parts[-3:] if len(parts) >= 3 else parts + head = parts[idx + 1 : len(parts) - 3] if len(parts) > idx + 4 else parts[idx + 1 : idx + 2] + if head: + if FORCE_DEVICE_ID: + result["camera"] = FORCE_DEVICE_ID + prefix = "/".join(head) + else: + result["camera"] = head[-1] + prefix = "/".join(head[:-1]) + else: + prefix = "" + print(f"[DEBUG] result['camera']={result['camera']}, prefix={prefix}, FORCE_DEVICE_ID={FORCE_DEVICE_ID}", flush=True) + try: + result["publish_ts_ms"] = int(tail[0]) + except Exception: + pass + if len(tail) >= 2: + result["content_type"] = tail[1].replace("_", "/") + if len(tail) >= 3: + result["filename"] = tail[2] + + # --- DEBUG + try to detect device from filename --- + filename_base = os.path.splitext(result["filename"])[0] + if "-" in filename_base: + possible_device = filename_base.split("_")[0] + print(f"[DEBUG] filename_base={filename_base}, possible_device={possible_device}", flush=True) + if possible_device.lower().startswith("fruit") or possible_device.lower().startswith("camera"): + result["camera"] = possible_device + print(f"[DEBUG] DETECTED device from filename -> {result['camera']}", flush=True) + else: + print(f"[DEBUG] filename does not match expected pattern", flush=True) + else: + print(f"[DEBUG] filename_base has no '-': {filename_base}", flush=True) + result["extra_prefix"] = prefix elif ns in ("sounds", "sounds_ultra"): if len(parts) > idx + 1 and parts[idx + 1]: try: @@ -190,6 +210,8 @@ def parse_topic(topic: str) -> dict: date_part = datetime.fromtimestamp(result["publish_ts_ms"] / 1000, tz=timezone.utc).strftime("%Y-%m-%d") device_id = result["camera"] + if FORCE_DEVICE_ID: + device_id = FORCE_DEVICE_ID if result["media_type"] == "sounds": if device_id.startswith(f"{CAMERA_PREFIX}-"): @@ -206,11 +228,16 @@ def parse_topic(topic: str) -> dict: else: device_name = f"{CAMERA_PREFIX}-{device_id}" - # key = f"{result['media_type']}/{device_name}/{date_part}/{result['publish_ts_ms']}/{result['filename']}" is_ultra = ns == "sounds_ultra" - topdir = ULTRA_DIR_PREFIX if is_ultra else result["media_type"] - key = f"{topdir}/{device_name}/{date_part}/{result['publish_ts_ms']}/{result['filename']}" - + topdir = ULTRA_DIR_PREFIX if is_ultra else result["media_type"] + if ns == "imagery": + subpath = "/".join(parts[idx + 1 : -3]) + if subpath: + key = f"{subpath}/{device_name}/{date_part}/{result['publish_ts_ms']}/{result['filename']}" + else: + key = f"{topdir}/{device_name}/{date_part}/{result['publish_ts_ms']}/{result['filename']}" + else: + key = f"{topdir}/{device_name}/{date_part}/{result['publish_ts_ms']}/{result['filename']}" result["key"] = key result["device_id"] = device_name result["image_id"] = stem(result["filename"]) or uuid.uuid4().hex @@ -473,4 +500,4 @@ def _stop(*_): print("INGEST stopped.", flush=True) if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/storage_with_mqtt/storage/Lifecycle_rules/minio-bootstrap/Dockerfile b/storage_with_mqtt/storage/Lifecycle_rules/minio-bootstrap/Dockerfile index d5b4d94f9..0bb44c4fa 100644 --- a/storage_with_mqtt/storage/Lifecycle_rules/minio-bootstrap/Dockerfile +++ b/storage_with_mqtt/storage/Lifecycle_rules/minio-bootstrap/Dockerfile @@ -11,8 +11,16 @@ FROM alpine:3.19 RUN apk add --no-cache bash curl ca-certificates netcat-openbsd dos2unix && \ update-ca-certificates # ===== Add NetFree CA ===== -COPY certs/*.crt /usr/local/share/ca-certificates/ -RUN update-ca-certificates + +RUN if [ -f netfree-ca.crt ]; then \ + echo "Installing netfree-ca.crt..."; \ + cp netfree-ca.crt /usr/local/share/ca-certificates/netfree-ca.crt; \ + chmod 644 /usr/local/share/ca-certificates/netfree-ca.crt; \ + update-ca-certificates; \ + else \ + echo "No netfree-ca.crt found, skipping."; \ + fi + # ===== Copy mc from the official image ===== COPY --from=mc-source /usr/bin/mc /usr/local/bin/mc RUN chmod +x /usr/local/bin/mc diff --git a/storage_with_mqtt/storage/Lifecycle_rules/minio-bootstrap/entrypoint/init.sh b/storage_with_mqtt/storage/Lifecycle_rules/minio-bootstrap/entrypoint/init.sh index 16b388157..54fc2b3fa 100644 --- a/storage_with_mqtt/storage/Lifecycle_rules/minio-bootstrap/entrypoint/init.sh +++ b/storage_with_mqtt/storage/Lifecycle_rules/minio-bootstrap/entrypoint/init.sh @@ -171,7 +171,7 @@ mc event add "${MC_ALIAS_HOT}/imagery" \ mc event add "${MC_ALIAS_HOT}/imagery" \ arn:minio:sqs::fruits:kafka \ --event put \ - --prefix "fruits/" + --prefix "fruit/tree/" mc event add "${MC_ALIAS_HOT}/imagery" \ arn:minio:sqs::leaves:kafka \ diff --git a/streaming/flink/jobs/http_dispatcher.py b/streaming/flink/jobs/http_dispatcher.py index 5fcdfb553..f7b9faf36 100644 --- a/streaming/flink/jobs/http_dispatcher.py +++ b/streaming/flink/jobs/http_dispatcher.py @@ -85,7 +85,11 @@ async def _make_session(self, timeout: aiohttp.ClientTimeout) -> aiohttp.ClientS async def _post_once(self, url: str, payload: Dict[str, Any], headers: Dict[str, str]): async with self.session.post(url, json=payload, headers=headers) as resp: - return resp.status, await resp.text() + try: + data = await resp.json() + except Exception: + data = await resp.text() + return resp.status, data async def _post_with_retry(self, url: str, payload: Dict[str, Any], headers: Dict[str, str]): attempt = 0 @@ -111,33 +115,58 @@ def map(self, s: str) -> str: # 1) Parse JSON try: event = json.loads(s) + # Handle double-encoded JSON strings (common in Kafka/MinIO) + if isinstance(event, str): + event = json.loads(event) except Exception as e: return json.dumps( {"ok": False, "status": 422, "body": f"bad json: {e}", "raw": s, "stage": "parse"}, ensure_ascii=False ) - # 2) Generate idempotency key event_id = event.get("event_id") or str(uuid.uuid4()) + + # 3) Extract bucket/key – support both native and MinIO S3-event formats + import sys, urllib.parse - # 3) Validate fields: must have bucket+key; image_uri not allowed - if "image_uri" in event: - return json.dumps( - {"ok": False, "status": 422, - "body": "image_uri not supported; use {bucket,key} only", - "event": event, "stage": "validate"}, - ensure_ascii=False - ) + def dbg(msg: str): + sys.stdout.write(f"[DEBUG] {msg}\n") + sys.stdout.flush() bucket = event.get("bucket") key = event.get("key") - if not bucket or not key: - return json.dumps( - {"ok": False, "status": 422, - "body": "missing required fields: bucket and key", - "event": event, "stage": "validate"}, - ensure_ascii=False - ) + + # Try to detect MinIO S3 event format + if (not bucket or not key): + # Case A: nested Records structure + records = event.get("Records") + if isinstance(records, list) and len(records) > 0: + try: + rec = records[0] + s3_block = rec.get("s3", {}) + bucket = s3_block.get("bucket", {}).get("name") + raw_key = s3_block.get("object", {}).get("key") + if raw_key: + key = urllib.parse.unquote(raw_key) + # ננקה imagery/ כפול אם מופיע + if key.startswith("imagery/"): + key = key[len("imagery/"):] + dbg(f"Parsed from S3 event: bucket={bucket}, key={key}") + except Exception as e: + dbg(f"Error parsing S3 event: {e}") + # Case B: flat event with Key field + elif "Key" in event and isinstance(event["Key"], str): + key = event["Key"] + if key.startswith("imagery/"): + key = key[len("imagery/"):] + bucket = "imagery" + dbg(f"Parsed flat event: bucket={bucket}, key={key}") + # Case C: nested body from previous inference + if (not bucket or not key) and isinstance(event.get("body"), dict): + body = event["body"] + bucket = body.get("bucket") or body.get("bucket_out") + key = body.get("key") + dbg(f"Parsed nested body: bucket={bucket}, key={key}") # 4) Prepare headers and payload: send only bucket/key to the inference service headers = { @@ -239,4 +268,4 @@ def _is_ok(s: str) -> bool: if __name__ == "__main__": - main() + main() \ No newline at end of file