Copilot commented on code in PR #2258:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2258#discussion_r3968143901
##########
minifi_rust/extensions/minifi_tensor/src/processors/draw_bounding_box.rs:
##########
@@ -0,0 +1,165 @@
+use crate::utils::bounding_box::{BoundingBox, BoundingBoxes};
+use image::{ImageFormat, Rgb, load_from_memory};
+use minifi_native::macros::ComponentIdentifier;
+use minifi_native::{
+ FlowFileTransform, GetAttribute, GetControllerService, GetId, GetProperty,
InputStream, Logger,
+ MinifiError, OutputAttribute, ProcessError, ProcessorDefinition,
ProcessorInputRequirement,
+ Property, PropertyConstraints, PropertyType, Relationship, RouteErrorExt,
Schedule,
+ TransformedFlowFile,
+};
+use minifi_native::{PropertyDefinition, PropertySchema, property_definitions};
+use std::io::Cursor;
+
+pub(crate) const SUCCESS: Relationship = Relationship {
+ name: "success",
+ description: "Flowfiles are routed here after drawing the bounding boxes",
+};
+
+pub(crate) const FAILURE: Relationship = Relationship {
+ name: "failure",
+ description: "Invalid FlowFiles are routed here",
+};
+
+pub(crate) const BOUNDING_BOXES: Property<BoundingBoxes> = Property::new(
+ "Bounding boxes",
+ "JSON array of bounding boxes to draw onto the image (fields class_id,
confidence, x_min, \
+ y_min, x_max, y_max; coordinates normalised to [0,1] against the image).
Typically the \
+ attribute produced by an upstream DetectObject or FilterBoundingBoxes
processor.",
+)
+.with_default("${enrichment.value}");
Review Comment:
This property defaults to an Expression Language value
(`${enrichment.value}`) and the Behave tests also set it via
`${detected_objects}`, but the property does not declare
`.supports_expression_language()`. As a result, the literal string `${...}` is
likely passed to JSON parsing and will fail at runtime. Mark `BOUNDING_BOXES`
as supporting expression language (and ensure the evaluation happens before
`PropertyType::parse`).
##########
minifi_rust/extensions/minifi_tensor/src/low_level_processors/filter_bounding_boxes.rs:
##########
@@ -0,0 +1,491 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod filter_bounding_boxes_def;
+
+use crate::utils::bounding_box::BoundingBox;
+use crate::utils::dimensions::Dimensions;
+use crate::utils::score_activation::ScoreActivation;
+use crate::utils::tensor_helpers::{deserialize_tensors, tensor_as_f32};
+use filter_bounding_boxes_def::SUCCESS;
+pub(crate) use filter_bounding_boxes_def::{
+ BACKGROUND_CLASS_INDEX, BOX_FORMAT, BOX_OUTPUT_INDEX, CLASS_OUTPUT_INDEX,
CONFIDENCE_THRESHOLD,
+ IOU_THRESHOLD, OUTPUT_ATTRIBUTE_NAME, SCORE_ACTIVATION, SCORE_OUTPUT_INDEX,
+};
+use minifi_native::macros::{ComponentIdentifier, PropertyType};
+use minifi_native::{
+ Content, FlowFileTransform, GetAttribute, GetId, GetProperty, InputStream,
Logger, MinifiError,
+ ProcessError, RouteErrorExt, Schedule, TransformedFlowFile, debug,
+};
+use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
+use tract::Tensor;
+
+#[derive(
+ Debug, Clone, Copy, PartialEq, Display, EnumString, VariantNames,
IntoStaticStr, PropertyType,
+)]
+#[strum(serialize_all = "PascalCase", const_into_str)]
+pub(crate) enum BoxFormat {
+ /// `[x_min, y_min, x_max, y_max]` — SSD, MobileNet-SSD, most PyTorch
models.
+ Xyxy,
+ /// `[y_min, x_min, y_max, x_max]` — TensorFlow Object Detection API.
+ Yxyx,
+ /// `[cx, cy, w, h]` — YOLOv3/5/8 raw output (center + size).
+ Cxcywh,
+}
+
+/// Convert the four floats at `box_floats[offset..offset+4]` into a canonical
+/// `(x_min, y_min, x_max, y_max)` tuple, regardless of the source layout.
+fn decode_box(box_floats: &[f32], offset: usize, format: BoxFormat) -> (f32,
f32, f32, f32) {
+ let a = box_floats[offset];
+ let b = box_floats[offset + 1];
+ let c = box_floats[offset + 2];
+ let d = box_floats[offset + 3];
+ match format {
+ BoxFormat::Xyxy => (a, b, c, d),
+ BoxFormat::Yxyx => (b, a, d, c),
+ BoxFormat::Cxcywh => {
+ let (cx, cy, w, h) = (a, b, c, d);
+ (cx - w / 2.0, cy - h / 2.0, cx + w / 2.0, cy + h / 2.0)
+ }
+ }
+}
+
+struct ScoredClass {
+ class_id: usize,
+ confidence: f32,
+}
+
+fn score_box(
+ logits: &[f32],
+ activation: ScoreActivation,
+ background_class_index: Option<usize>,
+) -> ScoredClass {
+ let num_classes = logits.len();
+
+ let best_valid = logits
+ .iter()
+ .enumerate()
+ .filter(|&(_, &logit)| logit.is_finite())
+ .filter(|&(id, _)| match background_class_index {
+ Some(bg_idx) => !(num_classes > 1 && id == bg_idx),
+ None => true,
+ })
+ .max_by(|a, b| a.1.total_cmp(b.1));
+
+ let (class_id, &best_logit) = match best_valid {
+ Some(val) => val,
+ None => {
+ return ScoredClass {
+ class_id: 0,
+ confidence: f32::NEG_INFINITY,
+ };
+ }
+ };
+
+ let confidence = match activation {
+ ScoreActivation::Softmax => {
+ let max_logit = logits
+ .iter()
+ .copied()
+ .filter(|l| l.is_finite())
+ .reduce(f32::max)
+ .unwrap_or(f32::NEG_INFINITY);
+ let sum_exp: f32 = logits
+ .iter()
+ .filter(|l| l.is_finite())
+ .map(|&l| (l - max_logit).exp())
+ .sum();
+
+ (best_logit - max_logit).exp() / sum_exp
+ }
+ ScoreActivation::Sigmoid => 1.0 / (1.0 + (-best_logit).exp()),
+ ScoreActivation::None => best_logit,
+ };
+
+ ScoredClass {
+ class_id,
+ confidence,
+ }
+}
+
+/// Turn a single per-box score into a confidence for the "separate class-id
+/// tensor" path.
+/// Sigmoid maps a raw logit to a probability;
+/// Softmax has no meaning over a single scalar, thus pass-through.
+/// None passes the score through.
+fn activate_scalar(score: f32, activation: ScoreActivation) -> f32 {
+ match activation {
+ ScoreActivation::Sigmoid => 1.0 / (1.0 + (-score).exp()),
+ ScoreActivation::Softmax | ScoreActivation::None => score,
+ }
+}
+
+#[derive(ComponentIdentifier)]
+pub(crate) struct FilterBoundingBoxes {
+ confidence_threshold: f32,
+ iou_threshold: f32,
+ score_output_index: usize,
+ box_output_index: usize,
+ box_format: BoxFormat,
+ score_activation: ScoreActivation,
+ background_class_index: Option<usize>,
+ class_output_index: Option<usize>,
+}
+
+impl Schedule for FilterBoundingBoxes {
+ fn schedule<Ctx: GetProperty, L: Logger>(
+ context: &Ctx,
+ _logger: &L,
+ ) -> Result<Self, MinifiError> {
+ let confidence_threshold =
context.get_property(&CONFIDENCE_THRESHOLD)?;
+ let iou_threshold = context.get_property(&IOU_THRESHOLD)?;
+ let score_output_index = context.get_property(&SCORE_OUTPUT_INDEX)?;
+ let box_output_index = context.get_property(&BOX_OUTPUT_INDEX)?;
+ let box_format = context.get_property(&BOX_FORMAT)?;
+ let score_activation = context.get_property(&SCORE_ACTIVATION)?;
+ let background_class_index =
context.get_property(&BACKGROUND_CLASS_INDEX)?;
+ let class_output_index = context.get_property(&CLASS_OUTPUT_INDEX)?;
+
+ Ok(Self {
+ confidence_threshold,
+ iou_threshold,
+ score_output_index,
+ box_output_index,
+ box_format,
+ score_activation,
+ background_class_index,
+ class_output_index,
+ })
+ }
+}
+
+impl FilterBoundingBoxes {
+ pub(crate) fn filter<'a, Context: GetProperty, LoggerImpl: Logger>(
+ &self,
+ context: &Context,
+ logger: &LoggerImpl,
+ tensors: Vec<Tensor>,
+ orig_dim: Dimensions,
+ target_dim: Dimensions,
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
+ let score_floats =
+ tensor_as_f32(&tensors,
self.score_output_index).route_err_to_failure()?;
+ let box_floats = tensor_as_f32(&tensors,
self.box_output_index).route_err_to_failure()?;
+
+ let scale = (target_dim.width / orig_dim.width).min(target_dim.height
/ orig_dim.height);
+ let pad_x = (target_dim.width - (orig_dim.width * scale)) / 2.0;
+ let pad_y = (target_dim.height - (orig_dim.height * scale)) / 2.0;
Review Comment:
The coordinate un-letterboxing logic always assumes `scale =
min(target/orig)` plus symmetric padding, which matches letterbox resize but is
incorrect for `Stretch` (different `scale_x`/`scale_y`, zero padding). Since
`ImageToTensor`/`DetectObject` support both resize modes, `FilterBoundingBoxes`
should either (a) take/derive the resize mode and compute inverse mapping
accordingly, or (b) have `ImageToTensor` emit explicit resize metadata (scale +
pad) as attributes for downstream consumers. Otherwise detections will be
mis-mapped whenever the input was stretched.
##########
minifi_rust/extensions/minifi_tensor/src/low_level_processors/filter_bounding_boxes.rs:
##########
@@ -0,0 +1,491 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod filter_bounding_boxes_def;
+
+use crate::utils::bounding_box::BoundingBox;
+use crate::utils::dimensions::Dimensions;
+use crate::utils::score_activation::ScoreActivation;
+use crate::utils::tensor_helpers::{deserialize_tensors, tensor_as_f32};
+use filter_bounding_boxes_def::SUCCESS;
+pub(crate) use filter_bounding_boxes_def::{
+ BACKGROUND_CLASS_INDEX, BOX_FORMAT, BOX_OUTPUT_INDEX, CLASS_OUTPUT_INDEX,
CONFIDENCE_THRESHOLD,
+ IOU_THRESHOLD, OUTPUT_ATTRIBUTE_NAME, SCORE_ACTIVATION, SCORE_OUTPUT_INDEX,
+};
+use minifi_native::macros::{ComponentIdentifier, PropertyType};
+use minifi_native::{
+ Content, FlowFileTransform, GetAttribute, GetId, GetProperty, InputStream,
Logger, MinifiError,
+ ProcessError, RouteErrorExt, Schedule, TransformedFlowFile, debug,
+};
+use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
+use tract::Tensor;
+
+#[derive(
+ Debug, Clone, Copy, PartialEq, Display, EnumString, VariantNames,
IntoStaticStr, PropertyType,
+)]
+#[strum(serialize_all = "PascalCase", const_into_str)]
+pub(crate) enum BoxFormat {
+ /// `[x_min, y_min, x_max, y_max]` — SSD, MobileNet-SSD, most PyTorch
models.
+ Xyxy,
+ /// `[y_min, x_min, y_max, x_max]` — TensorFlow Object Detection API.
+ Yxyx,
+ /// `[cx, cy, w, h]` — YOLOv3/5/8 raw output (center + size).
+ Cxcywh,
+}
+
+/// Convert the four floats at `box_floats[offset..offset+4]` into a canonical
+/// `(x_min, y_min, x_max, y_max)` tuple, regardless of the source layout.
+fn decode_box(box_floats: &[f32], offset: usize, format: BoxFormat) -> (f32,
f32, f32, f32) {
+ let a = box_floats[offset];
+ let b = box_floats[offset + 1];
+ let c = box_floats[offset + 2];
+ let d = box_floats[offset + 3];
+ match format {
+ BoxFormat::Xyxy => (a, b, c, d),
+ BoxFormat::Yxyx => (b, a, d, c),
+ BoxFormat::Cxcywh => {
+ let (cx, cy, w, h) = (a, b, c, d);
+ (cx - w / 2.0, cy - h / 2.0, cx + w / 2.0, cy + h / 2.0)
+ }
+ }
+}
+
+struct ScoredClass {
+ class_id: usize,
+ confidence: f32,
+}
+
+fn score_box(
+ logits: &[f32],
+ activation: ScoreActivation,
+ background_class_index: Option<usize>,
+) -> ScoredClass {
+ let num_classes = logits.len();
+
+ let best_valid = logits
+ .iter()
+ .enumerate()
+ .filter(|&(_, &logit)| logit.is_finite())
+ .filter(|&(id, _)| match background_class_index {
+ Some(bg_idx) => !(num_classes > 1 && id == bg_idx),
+ None => true,
+ })
+ .max_by(|a, b| a.1.total_cmp(b.1));
+
+ let (class_id, &best_logit) = match best_valid {
+ Some(val) => val,
+ None => {
+ return ScoredClass {
+ class_id: 0,
+ confidence: f32::NEG_INFINITY,
+ };
+ }
+ };
+
+ let confidence = match activation {
+ ScoreActivation::Softmax => {
+ let max_logit = logits
+ .iter()
+ .copied()
+ .filter(|l| l.is_finite())
+ .reduce(f32::max)
+ .unwrap_or(f32::NEG_INFINITY);
+ let sum_exp: f32 = logits
+ .iter()
+ .filter(|l| l.is_finite())
+ .map(|&l| (l - max_logit).exp())
+ .sum();
+
+ (best_logit - max_logit).exp() / sum_exp
+ }
+ ScoreActivation::Sigmoid => 1.0 / (1.0 + (-best_logit).exp()),
+ ScoreActivation::None => best_logit,
+ };
+
+ ScoredClass {
+ class_id,
+ confidence,
+ }
+}
+
+/// Turn a single per-box score into a confidence for the "separate class-id
+/// tensor" path.
+/// Sigmoid maps a raw logit to a probability;
+/// Softmax has no meaning over a single scalar, thus pass-through.
+/// None passes the score through.
+fn activate_scalar(score: f32, activation: ScoreActivation) -> f32 {
+ match activation {
+ ScoreActivation::Sigmoid => 1.0 / (1.0 + (-score).exp()),
+ ScoreActivation::Softmax | ScoreActivation::None => score,
+ }
+}
+
+#[derive(ComponentIdentifier)]
+pub(crate) struct FilterBoundingBoxes {
+ confidence_threshold: f32,
+ iou_threshold: f32,
+ score_output_index: usize,
+ box_output_index: usize,
+ box_format: BoxFormat,
+ score_activation: ScoreActivation,
+ background_class_index: Option<usize>,
+ class_output_index: Option<usize>,
+}
+
+impl Schedule for FilterBoundingBoxes {
+ fn schedule<Ctx: GetProperty, L: Logger>(
+ context: &Ctx,
+ _logger: &L,
+ ) -> Result<Self, MinifiError> {
+ let confidence_threshold =
context.get_property(&CONFIDENCE_THRESHOLD)?;
+ let iou_threshold = context.get_property(&IOU_THRESHOLD)?;
+ let score_output_index = context.get_property(&SCORE_OUTPUT_INDEX)?;
+ let box_output_index = context.get_property(&BOX_OUTPUT_INDEX)?;
+ let box_format = context.get_property(&BOX_FORMAT)?;
+ let score_activation = context.get_property(&SCORE_ACTIVATION)?;
+ let background_class_index =
context.get_property(&BACKGROUND_CLASS_INDEX)?;
+ let class_output_index = context.get_property(&CLASS_OUTPUT_INDEX)?;
+
+ Ok(Self {
+ confidence_threshold,
+ iou_threshold,
+ score_output_index,
+ box_output_index,
+ box_format,
+ score_activation,
+ background_class_index,
+ class_output_index,
+ })
+ }
+}
+
+impl FilterBoundingBoxes {
+ pub(crate) fn filter<'a, Context: GetProperty, LoggerImpl: Logger>(
+ &self,
+ context: &Context,
+ logger: &LoggerImpl,
+ tensors: Vec<Tensor>,
+ orig_dim: Dimensions,
+ target_dim: Dimensions,
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
+ let score_floats =
+ tensor_as_f32(&tensors,
self.score_output_index).route_err_to_failure()?;
+ let box_floats = tensor_as_f32(&tensors,
self.box_output_index).route_err_to_failure()?;
+
+ let scale = (target_dim.width / orig_dim.width).min(target_dim.height
/ orig_dim.height);
+ let pad_x = (target_dim.width - (orig_dim.width * scale)) / 2.0;
+ let pad_y = (target_dim.height - (orig_dim.height * scale)) / 2.0;
+
+ if !box_floats.len().is_multiple_of(4) {
+ return Err(MinifiError::custom(
+ "Box tensor byte length is not a multiple of 16 (4 f32 per
box)",
+ )
+ .into());
+ }
+ let num_boxes = box_floats.len() / 4;
+ if num_boxes == 0 {
+ debug!(logger, "No boxes to filter; emitting empty array");
+ return Ok(TransformedFlowFile::new(&SUCCESS, None)
+ .with_content(b"[]".to_vec().into())
+ .with_attribute("object.count", "0")
+ .with_attribute("mime.type", "application/json"));
+ }
+
+ let make_box = |i: usize, class_id: usize, confidence: f32| ->
BoundingBox {
+ let (raw_x_min, raw_y_min, raw_x_max, raw_y_max) =
+ decode_box(&box_floats, i * 4, self.box_format);
+ let true_x_min = (((raw_x_min * target_dim.width) - pad_x) /
scale) / orig_dim.width;
+ let true_y_min = (((raw_y_min * target_dim.height) - pad_y) /
scale) / orig_dim.height;
+ let true_x_max = (((raw_x_max * target_dim.width) - pad_x) /
scale) / orig_dim.width;
+ let true_y_max = (((raw_y_max * target_dim.height) - pad_y) /
scale) / orig_dim.height;
Review Comment:
The coordinate un-letterboxing logic always assumes `scale =
min(target/orig)` plus symmetric padding, which matches letterbox resize but is
incorrect for `Stretch` (different `scale_x`/`scale_y`, zero padding). Since
`ImageToTensor`/`DetectObject` support both resize modes, `FilterBoundingBoxes`
should either (a) take/derive the resize mode and compute inverse mapping
accordingly, or (b) have `ImageToTensor` emit explicit resize metadata (scale +
pad) as attributes for downstream consumers. Otherwise detections will be
mis-mapped whenever the input was stretched.
##########
minifi_rust/extensions/minifi_tensor/features/environment.py:
##########
@@ -0,0 +1,122 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import hashlib
+import os
+import shutil
+import ssl
+import urllib.request
+
+import certifi
+
+from minifi_behave.core.hooks import (
+ add_extension_to_minifi_container,
+ common_after_scenario,
+ common_before_scenario,
+)
+
+
+_SSL_CONTEXT = ssl.create_default_context(cafile=certifi.where())
+
+
+class RemoteAsset:
+ def __init__(self, url: str, sha256: str):
+ self.url = url
+ self.sha256 = sha256
+
+ def acquire(self, cache_dir: str, filename: str) -> str:
+ dest = os.path.join(cache_dir, filename)
+ if os.path.exists(dest) and self._verify(dest):
+ return dest
+ os.makedirs(cache_dir, exist_ok=True)
+ tmp = dest + ".part"
+ print(f"[minifi_tensor tests] fetching {filename} from {self.url}")
Review Comment:
`urlopen()` is called without a timeout, so a stalled network connection can
hang the Behave test job indefinitely. Please pass a reasonable timeout (and
consider surfacing a clearer error message on timeout) to keep CI behavior
predictable.
##########
minifi_rust/extensions/minifi_tensor/src/low_level_processors/classify_output.rs:
##########
@@ -0,0 +1,426 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::utils::score_activation::ScoreActivation;
+use crate::utils::tensor_helpers::{deserialize_tensors, tensor_as_f32,
tensor_shape};
+use classify_output_def::SUCCESS;
+pub(crate) use classify_output_def::{
+ CONFIDENCE_THRESHOLD, LABEL_INDEX_OFFSET, LABELS_FILE_PATH,
OUTPUT_ATTRIBUTE_NAME,
+ SCORE_ACTIVATION, SCORE_OUTPUT_INDEX, TOP_K,
+};
+use minifi_native::macros::ComponentIdentifier;
+use minifi_native::{
+ Content, FlowFileTransform, GetAttribute, GetId, GetProperty, InputStream,
Logger, MinifiError,
+ ProcessError, RouteErrorExt, Schedule, TransformedFlowFile,
+};
+use serde::Serialize;
+use std::path::Path;
+use tract::Tensor;
+
+mod classify_output_def;
+
+#[derive(Serialize, Clone, Debug, PartialEq)]
+struct Prediction {
+ class_id: usize,
+ confidence: f32,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ class_name: Option<String>,
+}
+
+fn load_labels(path: &Path) -> Result<Vec<String>, MinifiError> {
+ let content = std::fs::read_to_string(path).map_err(|e| {
+ MinifiError::custom(format!("Failed to read labels file '{:?}': {}",
path, e))
+ })?;
+ Ok(content
+ .lines()
+ .map(|line| line.trim_end().to_string())
+ .collect())
+}
+
+fn top_k(mut scored: Vec<(usize, f32)>, k: usize) -> Vec<(usize, f32)> {
+ scored.sort_by(|&(ai, a), &(bi, b)| b.total_cmp(&a).then(ai.cmp(&bi)));
+ scored.truncate(k);
+ scored
+}
+
+#[derive(ComponentIdentifier)]
+pub(crate) struct ClassifyOutput {
+ top_k: usize,
+ score_output_index: usize,
+ score_activation: ScoreActivation,
+ confidence_threshold: f32,
+ labels: Vec<String>,
+ label_index_offset: usize,
+}
+
+impl Schedule for ClassifyOutput {
+ fn schedule<Ctx: GetProperty, L: Logger>(
+ context: &Ctx,
+ _logger: &L,
+ ) -> Result<Self, MinifiError>
+ where
+ Self: Sized,
+ {
+ let top_k = context.get_property(&TOP_K)?;
+ if top_k == 0 {
+ return Err(MinifiError::validation("Top K must be >= 1"));
+ }
+ let score_output_index = context.get_property(&SCORE_OUTPUT_INDEX)?;
+ let score_activation = context.get_property(&SCORE_ACTIVATION)?;
+ let confidence_threshold =
context.get_property(&CONFIDENCE_THRESHOLD)?;
+
+ let labels = match context.get_property(&LABELS_FILE_PATH)? {
+ Some(path) => load_labels(&path)?,
+ _ => Vec::new(),
+ };
+ let label_index_offset = context.get_property(&LABEL_INDEX_OFFSET)?;
+
+ Ok(Self {
+ top_k,
+ score_output_index,
+ score_activation,
+ confidence_threshold,
+ labels,
+ label_index_offset,
+ })
+ }
+}
+
+impl ClassifyOutput {
+ fn label_for(&self, class_id: usize) -> Option<String> {
+ self.labels
+ .get(class_id.checked_add(self.label_index_offset)?)
+ .cloned()
+ }
+
+ pub(crate) fn classify<'a, Context: GetProperty + GetAttribute + GetId>(
+ &self,
+ context: &Context,
+ tensors: Vec<Tensor>,
+ ) -> Result<TransformedFlowFile<'a>, ProcessError> {
+ let score_floats =
+ tensor_as_f32(&tensors,
self.score_output_index).route_err_to_failure()?;
+ if score_floats.is_empty() {
+ return Err(MinifiError::custom("Score tensor is empty; nothing to
classify").into());
+ }
+
+ // A classifier head is a single score vector: shape [num_classes] or
+ // [1, .., num_classes]. We rank over the flattened class axis, so any
+ // leading axis > 1 (a real batch) would silently mix rows and yield
+ // class ids past num_classes. Reject it rather than produce garbage.
+ // (`ImageToTensor` emits batch=1 today; this just enforces the
contract.)
+ let shape = tensor_shape(&tensors,
self.score_output_index).route_err_to_failure()?;
+ if shape.iter().rev().skip(1).any(|&d| d != 1) {
+ return Err(MinifiError::custom(format!(
+ "ClassifyOutput expects a single score vector (shape
[num_classes] or \
+ [1, .., num_classes]); got {shape:?}. A batch dimension > 1
is not supported."
+ )))
+ .route_err_to_failure();
+ }
+
+ let finite: Vec<(usize, f32)> = score_floats
+ .iter()
+ .copied()
+ .enumerate()
+ .filter(|&(_, s)| s.is_finite())
+ .collect();
+
+ let (max_logit, sum_exp) = match self.score_activation {
+ ScoreActivation::Softmax => {
+ let max = finite
+ .iter()
+ .map(|&(_, s)| s)
+ .reduce(f32::max)
+ .unwrap_or(f32::NEG_INFINITY);
+ let sum = finite.iter().map(|&(_, s)| (s -
max).exp()).sum::<f32>();
+ (max, sum)
+ }
+ _ => (0.0, 1.0),
+ };
+
+ let predictions: Vec<Prediction> = top_k(finite, self.top_k)
+ .into_iter()
+ .filter_map(|(class_id, raw)| {
+ let confidence = match self.score_activation {
+ ScoreActivation::Softmax => (raw - max_logit).exp() /
sum_exp,
+ ScoreActivation::Sigmoid => 1.0 / (1.0 + (-raw).exp()),
+ ScoreActivation::None => raw,
+ };
+
+ if confidence >= self.confidence_threshold {
+ Some(Prediction {
+ class_id,
+ confidence,
+ class_name: self.label_for(class_id),
+ })
+ } else {
+ None
+ }
+ })
+ .collect();
+
+ let (content, extra_attribute) = match
context.get_property(&OUTPUT_ATTRIBUTE_NAME)? {
+ None => (
+ Some(Content::Buffer(
+ serde_json::to_vec(&predictions).route_err_to_failure()?,
+ )),
+ None,
+ ),
+ Some(output_attr) => (
+ None,
+ Some((output_attr,
serde_json::to_string(&predictions).unwrap())),
Review Comment:
This introduces a potential panic via `serde_json::to_string(...).unwrap()`.
Even if it’s unlikely in practice, this is still a hard crash path in a
processor. Please serialize with proper error handling (returning a
`ProcessError` routed to failure, consistent with the surrounding
`route_err_to_failure()` usage).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
