src/redstack/features/extraction.py

from __future__ import annotations

import math
from collections import defaultdict
from collections.abc import Iterable, Mapping
from datetime import date
from types import MappingProxyType
from typing import Final, Protocol, runtime_checkable

import numpy as np
import numpy.typing as npt

from redstack.domain.candidate.behavioral import BehavioralProfile
from redstack.domain.candidate.career import (
    CareerProfile,
    CareerRecency,
    PositionFact,
    TenureStats,
)
from redstack.domain.candidate.credibility import CredibilityProfile, SkillTrust
from redstack.domain.candidate.logistics import LogisticsProfile, SalaryBand
from redstack.domain.enums import (
    CareerTrack,
    CompanySize,
    EvidenceKind,
    LocationFit,
    NoticeFit,
    Proficiency,
    SignalAvailability,
)
from redstack.domain.errors import CQVInvariantError
from redstack.domain.ids import LpaAmount, Months, Similarity, SkillName, UnitScore
from redstack.domain.provenance import EvidenceRef
from redstack.domain.source import RawCandidate
from redstack.features import career, education, geography, honeypot, latents, signals
from redstack.features.registry import FeatureRegistry
from redstack.features.skills import CompetencyConcept, CompetencyLexicon
from redstack.features.skills import extract as extract_skills
from redstack.features.view import (
    FeatureCell,
    cell,
    clamp_unit,
    days_between,
    make_evidence,
)

__all__: tuple[str, ...] = (
    "build_behavioral_profile",
    "build_career_profile",
    "build_cells",
    "build_credibility_profile",
    "build_logistics_profile",
    "extract_row",
    "extract_row_with_base_cells",
    "extract_row_with_base_cells_and_logistics",
    "fold_semantic",
)

_DAYS_PER_MONTH: Final[float] = 30.4375
_SEMANTIC_DEPENDENT_GROUPS: Final[frozenset[str]] = frozenset(
    {"retr", "rank", "recsys", "ir", "nlp", "llm", "mle", "mlops", "eval", "jd"}
)

_PRODUCT_INDUSTRIES: Final[frozenset[str]] = frozenset(
    {
        "software",
        "product",
        "saas",
        "internet",
        "technology",
        "fintech",
        "e-commerce",
        "edtech",
        "ai/ml",
        "adtech",
        "healthtech",
        "healthtech ai",
        "conversational ai",
        "voice ai",
        "ai services",
        "insurance tech",
        "gaming",
        "consumer electronics",
        "media",
        "food delivery",
        "transportation",
    }
)

_CONSULTING_FIRMS: Final[frozenset[str]] = frozenset(
    {
        "tcs",
        "tata consultancy",
        "infosys",
        "wipro",
        "accenture",
        "cognizant",
        "capgemini",
        "hcl",
        "mindtree",
        "tech mahindra",
        "mphasis",
    }
)
_FOUNDER_TITLE_TOKENS: Final[tuple[str, ...]] = ("founder", "co-founder", "cofounder")
_SMALL_COMPANY_SIZES: Final[frozenset[CompanySize]] = frozenset(
    {CompanySize.S_1_10, CompanySize.S_11_50}
)


_LEXICON_TOKENS: Final[Mapping[str, frozenset[str]]] = MappingProxyType(
    {
        "retr": frozenset({"retrieval", "indexing", "elasticsearch", "solr"}),
        "rank": frozenset({"ranking", "ranker", "relevance", "reranking"}),
        "recsys": frozenset(
            {"recommendation", "recommender", "recsys", "personalization"}
        ),
        "ir": frozenset({"embeddings", "ann", "faiss", "retrieval"}),
        "nlp": frozenset({"nlp", "tokenization", "transformer"}),
        "llm": frozenset({"llm", "gpt", "langchain", "openai", "prompt"}),
        "mle": frozenset({"pytorch", "tensorflow", "scikit"}),
        "mlops": frozenset({"mlops", "mlflow", "kubeflow"}),
        "eval": frozenset({"benchmarking", "ndcg", "mrr"}),
    }
)


def _default_lexicon() -> CompetencyLexicon:
    return CompetencyLexicon(
        concepts={
            group: CompetencyConcept(tokens=tokens, anchor_id=f"jd.{group}")
            for group, tokens in _LEXICON_TOKENS.items()
        }
    )


_LEXICON: Final[CompetencyLexicon] = _default_lexicon()


# --------------------------------------------------------------------------- #
# Structural-slice builders (no other builder exists for these in the repo). #
# --------------------------------------------------------------------------- #
def build_career_profile(raw: RawCandidate, *, as_of: date) -> CareerProfile:
    """Derive the structural ``CareerProfile`` directly from ``RawCandidate``.

    At most one ``PositionFact.is_current`` survives: if the raw data carries
    more than one (a preserved semantic contradiction), the chronologically
    most recent claim wins and the rest are resolved to ``False`` so the
    typed slice stays constructible -- the contradiction itself is still
    visible to the Integrity Engine via the original ``RawCandidate``.
    """
    positions_desc = sorted(
        raw.career_history, key=lambda p: p.start_date, reverse=True
    )
    facts: list[PositionFact] = []
    seen_current = False
    for pos in positions_desc:
        is_current = False
        if pos.is_current and not seen_current:
            is_current = True
            seen_current = True
        industry = pos.industry.casefold()
        description = pos.description.casefold()
        is_product = industry in _PRODUCT_INDUSTRIES
        company = pos.company.casefold()
        is_consulting = any(firm in company for firm in _CONSULTING_FIRMS) or (
            "consulting" in industry or "consulting" in description
        )
        facts.append(
            PositionFact(
                company=pos.company,
                title=pos.title,
                start_date=pos.start_date,
                end_date=pos.end_date,
                duration_months=pos.duration_months,
                is_current=is_current,
                industry=pos.industry,
                company_size=pos.company_size,
                is_product_company=is_product,
                is_consulting_firm=is_consulting,
                description_role_match=UnitScore(
                    0.6 if pos.description.strip() else 0.3
                ),
            )
        )
    current = next((f for f in facts if f.is_current), None)

    durations = [int(f.duration_months) for f in facts if f.duration_months > 0]
    hop_count = sum(1 for d in durations if d < 18)
    tenure = TenureStats(
        position_count=len(facts),
        mean_tenure_months=(math.fsum(durations) / len(durations))
        if durations
        else 0.0,
        min_tenure_months=float(min(durations)) if durations else 0.0,
        hop_rate=UnitScore(clamp_unit(hop_count / len(facts)) if facts else 0.0),
    )

    if current is not None:
        months_since_last = 0
    elif facts:
        end_ref = facts[0].end_date if facts[0].end_date is not None else as_of
        months_since_last = max(
            0, round(days_between(as_of, end_ref) / _DAYS_PER_MONTH)
        )
    else:
        months_since_last = 0
    recency = CareerRecency(
        most_recent_start=facts[0].start_date if facts else as_of,
        is_currently_employed=current is not None,
        months_since_last_role=Months(months_since_last),
    )

    total_months = sum(int(f.duration_months) for f in facts)
    derived_years = min(total_months / 12.0, 50.0)
    product_months = sum(int(f.duration_months) for f in facts if f.is_product_company)
    services_months = total_months - product_months
    if total_months == 0:
        track = CareerTrack.UNKNOWN
    elif product_months >= 2 * services_months:
        track = CareerTrack.PRODUCT
    elif services_months >= 2 * product_months:
        track = CareerTrack.SERVICES
    else:
        track = CareerTrack.MIXED

    return CareerProfile(
        stated_experience_years=min(float(raw.profile.years_of_experience), 50.0),
        derived_experience_years=derived_years,
        positions=tuple(facts),
        current_position=current,
        track=track,
        tenure=tenure,
        recency=recency,
        title_consistency=UnitScore(0.7),
    )


def build_credibility_profile(raw: RawCandidate) -> CredibilityProfile:
    """Derive the structural ``CredibilityProfile`` (skill trust + stuffing signal)."""
    scores = raw.redrob_signals.skill_assessment_scores
    description_blob = " ".join(p.description.casefold() for p in raw.career_history)
    skill_trust: dict[SkillName, SkillTrust] = {}
    for skill in raw.skills:
        assessment = scores.get(skill.name)
        endorsement_norm = clamp_unit(math.log1p(skill.endorsements) / math.log1p(50.0))
        duration_norm = (
            0.0
            if skill.duration_months is None
            else clamp_unit(math.log1p(int(skill.duration_months)) / math.log1p(36.0))
        )
        assessment_norm = 0.0 if assessment is None else clamp_unit(assessment / 100.0)
        trust_value = clamp_unit(
            0.4 * endorsement_norm + 0.3 * duration_norm + 0.3 * assessment_norm
        )
        skill_trust[skill.name] = SkillTrust(
            name=skill.name,
            proficiency=skill.proficiency,
            endorsements=skill.endorsements,
            duration_months=skill.duration_months,
            assessment_score=UnitScore(assessment_norm)
            if assessment is not None
            else None,
            trust=UnitScore(trust_value),
            is_credible=trust_value >= 0.5,
        )

    if raw.skills:
        advanced_zero = sum(
            1
            for s in raw.skills
            if s.proficiency >= Proficiency.ADVANCED
            and s.endorsements == 0
            and s.duration_months in (None, 0)
        )
        stuffing = clamp_unit(advanced_zero / len(raw.skills))
        credible_fraction = clamp_unit(
            sum(1 for st in skill_trust.values() if st.is_credible) / len(raw.skills)
        )
        gap = clamp_unit(1.0 - credible_fraction)
        relevant_credibility = credible_fraction
    else:
        stuffing = 0.0
        gap = 0.0
        relevant_credibility = 0.0
    _ = description_blob  # reserved for a future in-career corroboration pass

    return CredibilityProfile(
        skill_trust=skill_trust,
        keyword_stuffing_score=UnitScore(stuffing),
        claimed_vs_assessed_gap=UnitScore(gap),
        title_description_coherence=UnitScore(0.7),
        relevant_skill_credibility=UnitScore(relevant_credibility),
    )


def build_logistics_profile(
    raw: RawCandidate, *, jd_hubs: frozenset[str] = geography.DEFAULT_JD_HUBS
) -> LogisticsProfile:
    """Derive the structural ``LogisticsProfile``, banding ``LocationFit``/
    ``NoticeFit``."""
    sig = raw.redrob_signals
    country = raw.profile.country.strip().casefold()
    city = raw.profile.location.strip().casefold()
  
    city_only = city.split(",", 1)[0].strip()
    is_india = country in ("india", "in")
    in_hub = city in jd_hubs or city_only in jd_hubs

  
    if not is_india:
        location_fit = LocationFit.OUTSIDE_INDIA_NO_SPONSOR
    elif in_hub:
        location_fit = LocationFit.PREFERRED_HUB
    elif sig.willing_to_relocate:
        location_fit = LocationFit.INDIA_RELOCATABLE
    else:
        location_fit = LocationFit.INDIA_NON_RELOCATABLE

    notice_days = int(sig.notice_period_days)
    if notice_days <= 30:
        notice_fit = NoticeFit.SUB_30_IDEAL
    elif notice_days <= 60:
        notice_fit = NoticeFit.BUYOUTABLE
    else:
        notice_fit = NoticeFit.OVER_30_HIGHER_BAR

    salary = sig.expected_salary_range_inr_lpa
    salary_band = SalaryBand(
        min_lpa=LpaAmount(float(salary.min)),
        max_lpa=LpaAmount(float(salary.max)),
        is_inverted=float(salary.min) > float(salary.max),
    )

    return LogisticsProfile(
        location=raw.profile.location,
        country=raw.profile.country,
        location_fit=location_fit,
        willing_to_relocate=sig.willing_to_relocate,
        notice_period_days=notice_days,
        notice_fit=notice_fit,
        preferred_work_mode=sig.preferred_work_mode,
        work_mode_fit=UnitScore(1.0),
        salary=salary_band,
    )


def build_behavioral_profile(raw: RawCandidate, *, as_of: date) -> BehavioralProfile:
    """Derive the structural ``BehavioralProfile``, honoring sentinel->UNKNOWN."""
    sig = raw.redrob_signals

    days_since_active = max(0, days_between(as_of, sig.last_active_date))
    engagement = clamp_unit(math.pow(0.5, days_since_active / 90.0))

    if sig.offer_acceptance_rate < 0.0:
        verification = 0.5
        verification_status = SignalAvailability.UNKNOWN
    else:
        verification = clamp_unit(sig.offer_acceptance_rate)
        verification_status = SignalAvailability.PRESENT

    return BehavioralProfile(
        availability=UnitScore(1.0 if sig.open_to_work_flag else 0.4),
        availability_status=SignalAvailability.PRESENT,
        responsiveness=UnitScore(clamp_unit(sig.recruiter_response_rate)),
        responsiveness_status=SignalAvailability.PRESENT,
        engagement=UnitScore(engagement),
        engagement_status=SignalAvailability.PRESENT,
        reliability=UnitScore(clamp_unit(sig.interview_completion_rate)),
        reliability_status=SignalAvailability.PRESENT,
        verification=UnitScore(verification),
        verification_status=verification_status,
        raw=sig,
    )


# --------------------------------------------------------------------------- #
# The fourteen features with no dedicated extractor module (honest, simple). #
# --------------------------------------------------------------------------- #
@runtime_checkable
class _RawCell(Protocol):
    """Structural shape shared by ``features.view.FeatureCell`` and
    ``features.parsing.FeatureCell`` -- two independently-defined but
    field-identical classes; this lets ``_normalize`` accept either."""

    @property
    def value(self) -> float: ...
    @property
    def confidence(self) -> UnitScore: ...
    @property
    def evidence(self) -> tuple[EvidenceRef, ...]: ...


def _normalize(items: Iterable[tuple[object, _RawCell]]) -> dict[str, FeatureCell]:
    """Re-wrap any ``_RawCell``-shaped items into ``features.view.FeatureCell``.

    Keys arrive typed as either ``features.parsing.FeatureId`` or plain
    ``str`` depending on the source extractor; both are ``str`` at runtime
    (the former a ``NewType``), so ``str(fid)`` is a lossless normalization,
    not a real coercion. Accepting an items iterable (rather than a
    ``Mapping``) sidesteps ``Mapping``'s key-type invariance, since the two
    source modules declare structurally-identical but nominally-distinct
    ``FeatureId``/``FeatureCell`` types.
    """
    return {
        str(fid): cell(float(c.value), float(c.confidence), c.evidence)
        for fid, c in items
    }


def _simple_groups(
    raw: RawCandidate, career_cells: Mapping[str, FeatureCell]
) -> dict[str, FeatureCell]:
    """``id.*`` / ``exp.*`` / ``sen.*`` / ``co.*`` / ``lead.*`` / ``startup.*`` /
    ``found.*`` -- the 14 features with no dedicated extractor module.

    Several reuse an already-computed ``career.*``/``pvs.*`` cell as an honest
    proxy rather than re-deriving an equivalent signal from scratch.
    """
    id_ev = make_evidence(
        EvidenceKind.PROFILE_FIELD, "candidate_id", raw.candidate_id, raw=raw
    )
    years_ev = make_evidence(
        EvidenceKind.PROFILE_FIELD,
        "profile.years_of_experience",
        float(raw.profile.years_of_experience),
        raw=raw,
    )
    authenticity = career_cells["career.experience_authenticity"]
    progression = career_cells["career.progression_quality"]
    inflation = career_cells["career.title_inflation"]
    company_progression = career_cells["career.company_progression"]
    product_density = career_cells["career.product_company_density"]
    management_only = career_cells["career.management_only"]
    production_exposure = career_cells["career.production_exposure"]

    small_co = any(p.company_size in _SMALL_COMPANY_SIZES for p in raw.career_history)
    first_position = raw.career_history[0]
    small_co_ev = make_evidence(
        EvidenceKind.CAREER_FIELD,
        "career_history[0].company_size",
        first_position.company_size.value,
        raw=raw,
    )
    founder_hit = any(
        any(token in p.title.casefold() for token in _FOUNDER_TITLE_TOKENS)
        for p in raw.career_history
    )
    founder_ev = make_evidence(
        EvidenceKind.CAREER_FIELD,
        "career_history[0].title",
        first_position.title,
        raw=raw,
    )

    return {
        "id.is_valid_id": cell(1.0, 1.0, (id_ev,)),
        # exp.years carries raw years (layout bounds 0..50), not a UnitScore.
        "exp.years": cell(float(raw.profile.years_of_experience), 0.9, (years_ev,)),
        # No JD experience band injected here -> neutral prior, low confidence.
        "exp.in_band": cell(0.5, 0.3, (years_ev,)),
        "exp.derived_vs_stated_gap": cell(
            clamp_unit(1.0 - authenticity.value),
            float(authenticity.confidence),
            authenticity.evidence,
        ),
        "sen.level": cell(
            progression.value, float(progression.confidence), progression.evidence
        ),
        "sen.title_vs_scope_gap": cell(
            inflation.value, float(inflation.confidence), inflation.evidence
        ),
        "co.scale_progression": cell(
            company_progression.value,
            float(company_progression.confidence),
            company_progression.evidence,
        ),
        "co.industry_relevance": cell(
            product_density.value,
            float(product_density.confidence),
            product_density.evidence,
        ),
        "lead.scope": cell(
            clamp_unit(1.0 - management_only.value),
            float(management_only.confidence),
            management_only.evidence,
        ),
        "lead.management_only": cell(
            management_only.value,
            float(management_only.confidence),
            management_only.evidence,
        ),
        "startup.small_co_experience": cell(
            1.0 if small_co else 0.0, 0.5, (small_co_ev,)
        ),
        "startup.shipping_signal": cell(
            production_exposure.value,
            float(production_exposure.confidence),
            production_exposure.evidence,
        ),
        "found.ownership": cell(1.0 if founder_hit else 0.0, 0.5, (founder_ev,)),
        "found.breadth": cell(0.5, 0.3, (founder_ev,)),
    }


# --------------------------------------------------------------------------- #
# Full per-candidate cell assembly + the R2/R3 public entry points.           #
# --------------------------------------------------------------------------- #
def _build_base_cells(
    raw: RawCandidate, *, as_of: date
) -> tuple[dict[str, FeatureCell], LogisticsProfile]:
    """Assemble the semantic-independent base cells; return the ``LogisticsProfile``
    computed during extraction so callers can reuse it without a second call.

    Career/pvs/geography/education/``_simple_groups``/signals/honeypot are all
    pure functions of ``(raw, as_of)`` alone -- their output is identical
    whether called from the R2 placeholder pass (``semantic={}``) or the R3
    resolved pass. A caller that runs both passes (online R2->R3) computes this
    once and feeds it to both :func:`extract_row_with_base_cells` and
    :func:`fold_semantic` instead of re-deriving it a second time.

    Order matters: ``career.*``/``pvs.*`` run first because ``_simple_groups``
    reuses their cells.
    """
    cells: dict[str, FeatureCell] = {}
    cells.update(dict(career.extract_career(raw, as_of=as_of)))
    cells.update(dict(career.extract_pvs(raw, as_of=as_of)))

    logistics = build_logistics_profile(raw)
    cells.update(_normalize(geography.extract_geography(raw, logistics).items()))
    cells.update(_normalize(education.extract_education(raw, as_of).items()))
    cells.update(_simple_groups(raw, cells))
    cells.update(dict(signals.extract(raw, as_of=as_of)))
    cells.update(dict(honeypot.extract(raw, as_of=as_of)))
    return cells, logistics


def _fold_skills_and_latents(
    base_cells: Mapping[str, FeatureCell],
    raw: RawCandidate,
    *,
    semantic: Mapping[str, Similarity],
) -> dict[str, FeatureCell]:
    """Layer the semantic-dependent skill-competency + ``jd.*`` latent cells onto
    a copy of ``base_cells``.

    ``extract_skills`` is the only extractor that reads ``semantic`` directly;
    ``latents.extract`` runs last because its ``jd.*`` cells (also semantic-
    dependent, via the competency cells) read the full accumulated map.
    """
    cells = dict(base_cells)
    cells.update(dict(extract_skills(raw, semantic=semantic, lexicon=_LEXICON)))
    cells.update(dict(latents.extract(cells)))
    return cells


def build_cells(
    raw: RawCandidate, *, as_of: date, semantic: Mapping[str, Similarity]
) -> dict[str, FeatureCell]:
    """Assemble every one of the 145 feature cells for one candidate."""
    base_cells, _logistics = _build_base_cells(raw, as_of=as_of)
    return _fold_skills_and_latents(base_cells, raw, semantic=semantic)


def _assemble(
    cells: Mapping[str, FeatureCell], registry: FeatureRegistry
) -> tuple[npt.NDArray[np.float32], npt.NDArray[np.float32]]:
    """Fold the assembled cells into the ``(D,)`` row + ``(G,)`` confidence row."""
    values = np.zeros(registry.dim, dtype=np.float32)
    conf_sum: defaultdict[str, float] = defaultdict(float)
    conf_count: defaultdict[str, float] = defaultdict(float)
    for definition in registry.definitions:
        fid = str(definition.feature_id)
        found = cells.get(fid)
        if found is None:
            raise CQVInvariantError(f"extract_row produced no cell for {fid!r}")
        value = float(found.value)
        if not math.isfinite(value):
            raise CQVInvariantError(f"feature {fid!r} is non-finite ({value!r})")
        low, high = definition.schema_.lower, definition.schema_.upper
        if value < low - 1e-6 or value > high + 1e-6:
            raise CQVInvariantError(
                f"feature {fid!r} value {value!r} outside bounds [{low}, {high}]"
            )
        values[int(definition.index)] = np.float32(value)
        group = definition.group
        conf_sum[group] += float(found.confidence)
        conf_count[group] += 1.0

    confidence = np.zeros(len(registry.groups), dtype=np.float32)
    for column, group in enumerate(registry.groups):
        count = conf_count[group]
        confidence[column] = (
            np.float32(conf_sum[group] / count) if count else np.float32(0.0)
        )
    values.setflags(write=False)
    confidence.setflags(write=False)
    return values, confidence


def extract_row(
    raw: RawCandidate, registry: FeatureRegistry, *, as_of: date
) -> tuple[npt.NDArray[np.float32], npt.NDArray[np.float32]]:

    cells = build_cells(raw, as_of=as_of, semantic={})
    return _assemble(cells, registry)


def extract_row_with_base_cells(
    raw: RawCandidate, registry: FeatureRegistry, *, as_of: date
) -> tuple[npt.NDArray[np.float32], npt.NDArray[np.float32], dict[str, FeatureCell]]:
    """Like :func:`extract_row`, but also returns the semantic-independent base
    cells (see :func:`_build_base_cells`).

    For a caller that will later call :func:`fold_semantic` on the same
    candidate (the online R2->R3 path), passing the returned ``base_cells``
    through lets R3 skip re-deriving career/geography/education/signals/
    honeypot a second time.
    """
    base_cells, _logistics = _build_base_cells(raw, as_of=as_of)
    cells = _fold_skills_and_latents(base_cells, raw, semantic={})
    values, confidence = _assemble(cells, registry)
    return values, confidence, base_cells


def extract_row_with_base_cells_and_logistics(
    raw: RawCandidate, registry: FeatureRegistry, *, as_of: date
) -> tuple[
    npt.NDArray[np.float32],
    npt.NDArray[np.float32],
    dict[str, FeatureCell],
    LogisticsProfile,
]:
    """Like :func:`extract_row_with_base_cells`, but also returns the
    :class:`~redstack.domain.candidate.logistics.LogisticsProfile` computed
    during base-cell extraction.

    Avoids calling :func:`build_logistics_profile` a second time in callers
    that need both the CQV row and the logistics profile for the same candidate
    (the online R2 loop).
    """
    base_cells, logistics = _build_base_cells(raw, as_of=as_of)
    cells = _fold_skills_and_latents(base_cells, raw, semantic={})
    values, confidence = _assemble(cells, registry)
    return values, confidence, base_cells, logistics


def fold_semantic(
    row: npt.NDArray[np.float32],
    confidence: npt.NDArray[np.float32],
    raw: RawCandidate,
    registry: FeatureRegistry,
    base_cells: Mapping[str, FeatureCell],
    *,
    semantic: Mapping[str, Similarity],
) -> dict[str, FeatureCell]:
    """Recompute the semantic-dependent cells of ``row``/``confidence`` in place (R3).

    Takes the ``base_cells`` already computed by R2's
    :func:`extract_row_with_base_cells` instead of rebuilding the full 145-cell
    set from scratch -- only ``extract_skills`` and the ``jd.*`` latents that sit
    on top of it actually read ``semantic``, so the career/geography/education/
    signals/honeypot extractors are not re-run here. Overwrites only the
    competency ``.semantic``/``.competency`` (etc., whole-group) cells and the
    ``jd.*`` latents in ``row``/``confidence``; every other index is untouched.
    Returns the full cell map it just built so the caller (R3) doesn't have to
    re-run anything a second time to get the per-candidate cell map it also
    needs.
    """
    full_cells = _fold_skills_and_latents(base_cells, raw, semantic=semantic)
    conf_sum: defaultdict[str, float] = defaultdict(float)
    conf_count: defaultdict[str, float] = defaultdict(float)
    for definition in registry.definitions:
        if definition.group not in _SEMANTIC_DEPENDENT_GROUPS:
            continue
        fid = str(definition.feature_id)
        found = full_cells[fid]
        row[int(definition.index)] = np.float32(float(found.value))
        group = definition.group
        conf_sum[group] += float(found.confidence)
        conf_count[group] += 1.0

    group_to_col: dict[str, int] = {g: col for col, g in enumerate(registry.groups)}
    for group, total in conf_sum.items():
        confidence[group_to_col[group]] = np.float32(total / conf_count[group])
    return full_cells