Classify Gases from Multimodal Information

Which of the three modalities gives the clearest initial separation? Start with the two gas-sensor readings and inspect their joint distribution by label.

Does a ratio or log transform make the sensor groups easier to separate? Add interactions such as sum, difference, and product.

Which repeated numeric and descriptive patterns can be extracted from the captions without a large language model?

How can the infrared image add information while keeping the validation split stratified? Compare compact image summaries against a trained image model, then blend out-of-fold probabilities.

Combine sensor interactions, caption TF-IDF/parsed clues, and compact infrared image features, then tune a probability blend using stratified weighted-F1 validation; task score remains unmeasured here.

#!/usr/bin/env python3
"""Multimodal JOAI 2025 gas-classification baseline and submission builder.

Expected input layout (download the competition data separately):
    data/
      train.csv
      test.csv
      sample_submission.csv  # optional
      images/*.png

The script uses only competition-provided data and scikit-learn/Pillow. It does
not download or redistribute the data or any pretrained weights.

Examples:
    python joai_2025_gas.py --data-dir /kaggle/input/playground-joai-competition-2025 \
        --mode validate --folds 5
    python joai_2025_gas.py --data-dir /kaggle/input/playground-joai-competition-2025 \
        --mode submit --output submission.csv
"""

from __future__ import annotations

import argparse
import re
import sys
import time
from pathlib import Path
from typing import Iterable

import numpy as np
import pandas as pd
from PIL import Image, ImageFile
from scipy import sparse
from sklearn.ensemble import ExtraTreesClassifier
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report, confusion_matrix, f1_score
from sklearn.model_selection import StratifiedKFold
from sklearn.preprocessing import StandardScaler


LABELS = ["Mixture", "NoGas", "Perfume", "Smoke"]
IMAGE_GRID = 4
IMAGE_FEATURES = 3 * (8 + IMAGE_GRID * IMAGE_GRID) + 16 + 1

TEMP_RANGE_RE = re.compile(
    r"(-?\d+(?:\.\d+)?)\s*(?:°\s*)?[-–]\s*"
    r"(-?\d+(?:\.\d+)?)\s*(?:°\s*)?([CF])\b",
    flags=re.IGNORECASE,
)
TEMP_RE = re.compile(r"(-?\d+(?:\.\d+)?)\s*(?:°\s*)?([CF])\b", flags=re.IGNORECASE)
COORD_RE = re.compile(r"\b\d{1,3}\s*°\s*\d{1,2}", flags=re.IGNORECASE)
COLOR_WORDS = (
    "red", "orange", "yellow", "green", "blue", "purple", "white", "black",
    "pink", "cyan", "magenta", "brown", "gray", "grey",
)
SCENE_WORDS = (
    "hot", "warm", "cool", "cold", "uniform", "plume", "leak", "rising",
    "gas", "smoke", "perfume", "flame", "hand", "person", "door", "wall",
    "indoor", "outdoor", "obscured", "unavailable", "temperature",
)


def normalize_label(value: object) -> str:
    label = str(value).strip()
    if label == "Mixed":
        return "Mixture"
    return label


def temperature_values(caption: str) -> tuple[list[float], bool, bool]:
    """Extract Celsius-normalized temperature values, including compact ranges."""
    values: list[float] = []
    has_c = False
    has_f = False
    range_spans: list[tuple[int, int]] = []
    for match in TEMP_RANGE_RE.finditer(caption):
        lo, hi = float(match.group(1)), float(match.group(2))
        unit = match.group(3).upper()
        has_c |= unit == "C"
        has_f |= unit == "F"
        if unit == "F":
            lo = (lo - 32.0) * (5.0 / 9.0)
            hi = (hi - 32.0) * (5.0 / 9.0)
        values.extend((lo, hi))
        range_spans.append(match.span())

    for match in TEMP_RE.finditer(caption):
        if any(start <= match.start() and match.end() <= end for start, end in range_spans):
            continue
        value = float(match.group(1))
        unit = match.group(2).upper()
        has_c |= unit == "C"
        has_f |= unit == "F"
        if unit == "F":
            value = (value - 32.0) * (5.0 / 9.0)
        values.append(value)
    return values, has_c, has_f


def caption_numeric_features(caption_value: object) -> list[float]:
    caption = str(caption_value or "")
    lower = caption.lower()
    temps, has_c, has_f = temperature_values(caption)
    if temps:
        temp_stats = [min(temps), max(temps), float(np.mean(temps)), float(np.std(temps))]
    else:
        temp_stats = [0.0, 0.0, 0.0, 0.0]

    words = re.findall(r"[a-z]+", lower)
    feats = [
        *temp_stats,
        float(len(temps)),
        float(has_c),
        float(has_f),
        float(bool(COORD_RE.search(caption))),
        float(len(caption)),
        float(len(words)),
        float(sum(ch.isdigit() for ch in caption)),
    ]
    feats.extend(float(color in lower) for color in COLOR_WORDS)
    feats.extend(float(word in lower) for word in SCENE_WORDS)
    return feats


def sensor_features(frame: pd.DataFrame) -> np.ndarray:
    mq5 = pd.to_numeric(frame.get("MQ5", pd.Series(np.zeros(len(frame)))), errors="coerce").to_numpy(float)
    mq8 = pd.to_numeric(frame.get("MQ8", pd.Series(np.zeros(len(frame)))), errors="coerce").to_numpy(float)
    safe5 = np.maximum(mq5, 0.0)
    safe8 = np.maximum(mq8, 0.0)
    return np.column_stack(
        [
            mq5,
            mq8,
            np.log1p(safe5),
            np.log1p(safe8),
            mq5 + mq8,
            mq5 - mq8,
            mq5 * mq8,
            mq5 / (mq8 + 1.0),
            mq8 / (mq5 + 1.0),
            np.log1p(safe5) - np.log1p(safe8),
        ]
    )


def image_features(path: Path | None) -> np.ndarray:
    if path is None or not path.is_file():
        return np.r_[np.zeros(IMAGE_FEATURES - 1, dtype=np.float32), 1.0]

    ImageFile.LOAD_TRUNCATED_IMAGES = True
    try:
        with Image.open(path) as image:
            rgb = np.asarray(
                image.convert("RGB").resize((32, 32), Image.Resampling.BILINEAR),
                dtype=np.float32,
            ) / 255.0
    except Exception:
        return np.r_[np.zeros(IMAGE_FEATURES - 1, dtype=np.float32), 1.0]

    feats: list[float] = []
    for channel in range(3):
        plane = rgb[:, :, channel]
        feats.extend(np.quantile(plane, [0.0, 0.05, 0.25, 0.5, 0.75, 0.95, 1.0]).tolist())
        feats.append(float(np.std(plane)))
        pooled = plane.reshape(IMAGE_GRID, 32 // IMAGE_GRID, IMAGE_GRID, 32 // IMAGE_GRID).mean(axis=(1, 3))
        feats.extend(pooled.ravel().tolist())
    gray = 0.299 * rgb[:, :, 0] + 0.587 * rgb[:, :, 1] + 0.114 * rgb[:, :, 2]
    hist, _ = np.histogram(gray, bins=16, range=(0.0, 1.0), density=False)
    feats.extend((hist / max(1, hist.sum())).tolist())
    feats.append(0.0)
    return np.asarray(feats, dtype=np.float32)


def build_image_lookup(data_dir: Path) -> dict[str, Path]:
    image_dir = data_dir / "images"
    if not image_dir.exists():
        image_dir = data_dir
    lookup: dict[str, Path] = {}
    for path in image_dir.rglob("*"):
        if path.is_file() and path.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}:
            lookup.setdefault(path.name.lower(), path)
    return lookup


def extract_image_matrix(frame: pd.DataFrame, lookup: dict[str, Path]) -> tuple[np.ndarray, int]:
    if "image_path_uuid" not in frame.columns:
        raise ValueError("CSV is missing the required image_path_uuid column")
    cache: dict[str, np.ndarray] = {}
    rows: list[np.ndarray] = []
    missing = 0
    for raw_name in frame["image_path_uuid"].fillna("").astype(str):
        name = Path(raw_name).name
        key = name.lower()
        if key not in cache:
            path = lookup.get(key)
            cache[key] = image_features(path)
            missing += int(path is None)
        rows.append(cache[key])
    return np.vstack(rows), missing


def dense_features(frame: pd.DataFrame, image_matrix: np.ndarray) -> np.ndarray:
    captions = frame.get("Caption", pd.Series([""] * len(frame))).fillna("").astype(str)
    caption = np.asarray([caption_numeric_features(value) for value in captions], dtype=np.float32)
    sensors = sensor_features(frame).astype(np.float32)
    dense = np.column_stack((sensors, caption, image_matrix)).astype(np.float32)
    return np.nan_to_num(dense, nan=0.0, posinf=0.0, neginf=0.0)


def make_text_matrices(
    train_text: Iterable[str],
    other_text: Iterable[str],
    train_dense: np.ndarray,
    other_dense: np.ndarray,
) -> tuple[sparse.csr_matrix, sparse.csr_matrix]:
    train_text = [str(x) for x in train_text]
    other_text = [str(x) for x in other_text]
    blocks_train: list[sparse.csr_matrix] = []
    blocks_other: list[sparse.csr_matrix] = []
    specs = [
        dict(analyzer="word", ngram_range=(1, 2), min_df=2, max_features=12000, token_pattern=r"(?u)\b\w+\b"),
        dict(analyzer="char_wb", ngram_range=(3, 5), min_df=2, max_features=16000),
    ]
    for spec in specs:
        vectorizer = TfidfVectorizer(sublinear_tf=True, strip_accents="unicode", **spec)
        try:
            blocks_train.append(vectorizer.fit_transform(train_text).tocsr())
            blocks_other.append(vectorizer.transform(other_text).tocsr())
        except ValueError as exc:
            if "empty vocabulary" not in str(exc).lower():
                raise

    scaler = StandardScaler()
    scaled_train = scaler.fit_transform(train_dense).astype(np.float32)
    scaled_other = scaler.transform(other_dense).astype(np.float32)
    blocks_train.append(sparse.csr_matrix(scaled_train))
    blocks_other.append(sparse.csr_matrix(scaled_other))
    return sparse.hstack(blocks_train, format="csr"), sparse.hstack(blocks_other, format="csr")


def fit_models(x_dense: np.ndarray, x_sparse: sparse.csr_matrix) -> tuple[ExtraTreesClassifier, LogisticRegression]:
    tree = ExtraTreesClassifier(
        n_estimators=500,
        max_features=0.9,
        min_samples_leaf=1,
        class_weight=None,
        n_jobs=-1,
        random_state=42,
    )
    linear = LogisticRegression(C=4.0, max_iter=1800, solver="lbfgs", random_state=42)
    return tree, linear


def aligned_proba(model: object, x: object, classes: np.ndarray) -> np.ndarray:
    raw = model.predict_proba(x)  # type: ignore[attr-defined]
    aligned = np.zeros((len(raw), len(LABELS)), dtype=np.float64)
    for source_index, label in enumerate(model.classes_):  # type: ignore[attr-defined]
        aligned[:, LABELS.index(str(label))] = raw[:, source_index]
    return aligned


def validate(train: pd.DataFrame, dense: np.ndarray, folds: int, seed: int) -> None:
    y = train["Gas"].map(normalize_label).to_numpy()
    oof_tree = np.zeros((len(y), len(LABELS)), dtype=np.float64)
    oof_linear = np.zeros_like(oof_tree)
    cv = StratifiedKFold(n_splits=folds, shuffle=True, random_state=seed)
    texts = train["Caption"].fillna("").astype(str).to_numpy()
    start = time.time()
    for fold, (tr_idx, va_idx) in enumerate(cv.split(dense, y), 1):
        x_tr_sparse, x_va_sparse = make_text_matrices(texts[tr_idx], texts[va_idx], dense[tr_idx], dense[va_idx])
        tree, linear = fit_models(dense[tr_idx], x_tr_sparse)
        tree.fit(dense[tr_idx], y[tr_idx])
        linear.fit(x_tr_sparse, y[tr_idx])
        oof_tree[va_idx] = aligned_proba(tree, dense[va_idx], np.asarray(LABELS))
        oof_linear[va_idx] = aligned_proba(linear, x_va_sparse, np.asarray(LABELS))
        p_tree = np.asarray(LABELS)[oof_tree[va_idx].argmax(axis=1)]
        p_linear = np.asarray(LABELS)[oof_linear[va_idx].argmax(axis=1)]
        print(
            f"fold={fold}/{folds}  tree_weighted_f1={f1_score(y[va_idx], p_tree, average='weighted'):.6f}"
            f"  text+numeric_weighted_f1={f1_score(y[va_idx], p_linear, average='weighted'):.6f}"
        )

    label_array = np.asarray(LABELS)
    print(f"\nCV weighted F1, ExtraTrees dense: {f1_score(y, label_array[np.argmax(oof_tree, axis=1)], average='weighted'):.6f}")
    print(f"CV weighted F1, TF-IDF + numeric logistic: {f1_score(y, label_array[np.argmax(oof_linear, axis=1)], average='weighted'):.6f}")
    blended = 0.5 * oof_tree + 0.5 * oof_linear
    blend_pred = np.asarray(LABELS)[np.argmax(blended, axis=1)]
    print(f"CV weighted F1, 50:50 probability blend: {f1_score(y, blend_pred, average='weighted'):.6f}")
    print("\nBlended classification report:")
    print(classification_report(y, blend_pred, labels=LABELS, zero_division=0, digits=4))
    print("Blended confusion matrix (rows=true, columns=predicted; label order:", LABELS, ")")
    print(confusion_matrix(y, blend_pred, labels=LABELS))
    print(f"Elapsed seconds (feature matrix already extracted): {time.time() - start:.1f}")


def train_submission(train: pd.DataFrame, test: pd.DataFrame, dense_train: np.ndarray, dense_test: np.ndarray, output: Path, data_dir: Path) -> None:
    y = train["Gas"].map(normalize_label).to_numpy()
    train_text = train["Caption"].fillna("").astype(str).to_numpy()
    test_text = test["Caption"].fillna("").astype(str).to_numpy()
    x_train_sparse, x_test_sparse = make_text_matrices(train_text, test_text, dense_train, dense_test)
    tree, linear = fit_models(dense_train, x_train_sparse)
    tree.fit(dense_train, y)
    linear.fit(x_train_sparse, y)
    probabilities = 0.5 * aligned_proba(tree, dense_test, np.asarray(LABELS))
    probabilities += 0.5 * aligned_proba(linear, x_test_sparse, np.asarray(LABELS))
    prediction = np.asarray(LABELS)[np.argmax(probabilities, axis=1)]

    sample_path = data_dir / "sample_submission.csv"
    if sample_path.is_file():
        submission = pd.read_csv(sample_path)
        if len(submission) != len(test):
            raise ValueError("sample_submission.csv row count differs from test.csv")
        submission["Gas"] = prediction
    else:
        submission = pd.DataFrame({"index": np.arange(len(test)), "Gas": prediction})
    output.parent.mkdir(parents=True, exist_ok=True)
    submission.to_csv(output, index=False)
    print(f"Wrote {len(submission)} predictions to {output}")


def load_data(data_dir: Path) -> tuple[pd.DataFrame, pd.DataFrame]:
    train_path = data_dir / "train.csv"
    test_path = data_dir / "test.csv"
    if not train_path.is_file() or not test_path.is_file():
        raise FileNotFoundError(
            f"Expected train.csv and test.csv in {data_dir}. Download competition data separately; "
            "this script does not fetch licensed task files."
        )
    train = pd.read_csv(train_path)
    test = pd.read_csv(test_path)
    required = {"MQ5", "MQ8", "Caption", "image_path_uuid"}
    missing = sorted(required - set(train.columns))
    if missing or "Gas" not in train.columns:
        raise ValueError(f"train.csv is missing required columns: {missing + ([] if 'Gas' in train else ['Gas'])}")
    missing_test = sorted(required - set(test.columns))
    if missing_test:
        raise ValueError(f"test.csv is missing required columns: {missing_test}")
    unknown = sorted(set(train["Gas"].map(normalize_label)) - set(LABELS))
    if unknown:
        raise ValueError(f"Unexpected label values: {unknown}")
    return train, test


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--data-dir", type=Path, default=Path("/kaggle/input/playground-joai-competition-2025"))
    parser.add_argument("--mode", choices=("validate", "submit"), default="validate")
    parser.add_argument("--folds", type=int, default=5)
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--output", type=Path, default=Path("submission.csv"))
    args = parser.parse_args(argv)
    if args.folds < 2:
        parser.error("--folds must be at least 2")

    train, test = load_data(args.data_dir)
    print(f"train={len(train):,} test={len(test):,}")
    print("class counts:", train["Gas"].map(normalize_label).value_counts().to_dict())

    t0 = time.time()
    lookup = build_image_lookup(args.data_dir)
    image_train, missing_train = extract_image_matrix(train, lookup)
    dense_train = dense_features(train, image_train)
    print(f"train images missing/unreadable: {missing_train}/{len(train)}")

    if args.mode == "validate":
        validate(train, dense_train, args.folds, args.seed)
    else:
        image_test, missing_test = extract_image_matrix(test, lookup)
        dense_test = dense_features(test, image_test)
        print(f"test images missing/unreadable: {missing_test}/{len(test)}")
        train_submission(train, test, dense_train, dense_test, args.output, args.data_dir)
    print(f"Total elapsed seconds: {time.time() - t0:.1f}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Download solution.py