modules/binary_classification.py

from __future__ import annotations
from typing import Optional
import tensorflow as tf
import os


BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
MODEL_DIR_DEFAULT = os.path.join(BASE_DIR, 'data', 'model', 'sentiment_cnn')

# Cache di processo (il modello viene caricato una volta sola)
_INFER: Optional[tf.types.experimental.ConcreteFunction] = None
_MODEL_DIR_LOADED: Optional[str] = None

def _load_infer(model_dir: str) -> tf.types.experimental.ConcreteFunction:
    """
    Load SavedModel once and return serving function.
    TF 2.12 compatible, expects signature 'serving_default'.
    """
    sm = tf.saved_model.load(model_dir)
    return sm.signatures["serving_default"]

def _get_infer(model_dir: str) -> tf.types.experimental.ConcreteFunction:
    global _INFER, _MODEL_DIR_LOADED
    model_dir = str(model_dir)
    if _INFER is None or _MODEL_DIR_LOADED != model_dir:
        _INFER = _load_infer(model_dir)
        _MODEL_DIR_LOADED = model_dir
    return _INFER


def _predict_prob_positive(text: str, model_dir: str) -> float:
    infer = _get_infer(model_dir)
    x = tf.constant([text], dtype=tf.string)
    out = infer(text=x)
    # Output key stabilized by our exporter
    y = out["prob"]
    return float(y.numpy()[0][0])


def binary_classification(text: str):
    """
    IDENTICAL behavior to legacy:
    - if text is empty -> ({"error": "Sentence is required"}, 415)
    - else -> {"positive": "0.xx", "negative": "0.yy"}  (strings, 2 decimals)
    """
    try:
        if text == "":
            raise Exception
    except:
        return {"error": "Sentence is required"}, 415

    prob_pos = _predict_prob_positive(text, MODEL_DIR_DEFAULT)

    positive = prob_pos
    negative = 1.0 - positive

    labels = {
        "positive": f"{positive:.2f}",
        "negative": f"{negative:.2f}",
    }
    return labels