Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Stage 3: Statistical bigram scoring.
3Note: ``from __future__ import annotations`` is intentionally omitted because
4this module is compiled with mypyc, which does not support PEP 563 string
5annotations.
6"""
8from chardet.models import (
9 BigramProfile,
10 _get_model_norms,
11 get_enc_index,
12 get_rowmax,
13 score_best_language,
14 score_with_profile,
15)
16from chardet.pipeline import DetectionResult
17from chardet.pipeline.postprocess import forced_encodings, scoring_floor
18from chardet.registry import EncodingInfo
20# Below this many distinct bigrams the upper-bound prescreen costs about as
21# much as the full dot products it would avoid, so score everything directly.
22_MIN_NONZERO_FOR_PRESCREEN = 64
25def _score_all(
26 data: bytes,
27 candidates: tuple[EncodingInfo, ...],
28 profile: BigramProfile,
29) -> list[tuple[str, float, str | None]]:
30 """Score every candidate fully (no pruning). Returns (enc, score, lang)."""
31 scores: list[tuple[str, float, str | None]] = []
32 for enc in candidates:
33 s, lang = score_best_language(data, enc.name, profile=profile)
34 if s > 0.0:
35 scores.append((enc.name, s, lang))
36 return scores
39def _split_variants(
40 candidates: tuple[EncodingInfo, ...],
41 profile: BigramProfile,
42) -> tuple[
43 list[tuple[str, str | None, bytes, str, int]],
44 list[tuple[float, int, str, str | None, bytes, str]],
45]:
46 """Flatten candidate model variants for pruned scoring.
48 Returns ``(mb_entries, sb_entries)`` where multi-byte entries are
49 ``(enc, lang, table, key, variant_index)`` and single-byte entries are
50 ``(upper_bound, variant_index, enc, lang, table, key)`` sorted by
51 descending bound. The upper bound multiplies each lead byte's total
52 profile weight by the model's maximum weight for that lead byte — at
53 most 256 terms versus one term per distinct bigram for a full score.
54 ``variant_index`` is the variant's position in the encoding index, so
55 exact score ties resolve to the same variant the full path keeps.
56 """
57 index = get_enc_index()
58 norms = _get_model_norms()
59 rowmax = get_rowmax()
60 row_freq = profile.row_freq
61 nonzero_rows = profile.nonzero_rows
62 input_norm = profile.input_norm
64 mb_entries: list[tuple[str, str | None, bytes, str, int]] = []
65 sb_entries: list[tuple[float, int, str, str | None, bytes, str]] = []
66 for enc in candidates:
67 variants = index.get(enc.name)
68 if variants is None:
69 continue
70 if enc.is_multibyte:
71 for vi, (lang, table, key) in enumerate(variants):
72 mb_entries.append((enc.name, lang, table, key, vi))
73 continue
74 for vi, (lang, table, key) in enumerate(variants):
75 rm = rowmax[key]
76 ub_dot = 0
77 for b1 in nonzero_rows:
78 ub_dot += rm[b1] * row_freq[b1]
79 model_norm = norms.get(key)
80 if model_norm is None:
81 # Unknown norm: cannot bound the score, so never skip.
82 ub = float("inf")
83 elif model_norm > 0.0:
84 ub = ub_dot / (model_norm * input_norm)
85 else:
86 # Zero norm: score_with_profile returns 0.0 for this model,
87 # so bound it at 0.0 instead of dividing by zero.
88 ub = 0.0
89 sb_entries.append((ub, vi, enc.name, lang, table, key))
90 sb_entries.sort(key=lambda e: e[0], reverse=True)
91 return mb_entries, sb_entries
94def _score_pruned(
95 candidates: tuple[EncodingInfo, ...],
96 profile: BigramProfile,
97) -> list[tuple[str, float, str | None]]:
98 """Score candidates, skipping single-byte variants that provably cannot matter.
100 Multi-byte variants are always scored fully — the orchestrator may later
101 boost their confidence based on structural coverage, so no raw-score
102 bound can rule them out. Single-byte variants are scored in descending
103 upper-bound order (see :func:`_split_variants`) and skipped once their
104 bound falls below the pruning contract's ``scoring_floor``: such
105 variants can affect neither the winner, nor position 1, nor any
106 candidate a rank correction can examine.
108 Encodings that ``postprocess_results`` inspects regardless of rank
109 (the common Western Latin trio for niche-Latin demotion, KOI8-T for the
110 KOI8-R promotion) are force-scored when their trigger could fire.
111 Because confusion resolution can promote position 1 or any candidate
112 within the band into position 0 before those triggers are evaluated,
113 the trigger check covers every encoding near the top, not just the
114 statistical winner.
116 Returns (enc, score, lang) tuples for encodings scoring above zero, in
117 candidate order.
118 """
119 index = get_enc_index()
120 mb_entries, sb_entries = _split_variants(candidates, profile)
122 best_score: dict[str, float] = {}
123 best_lang: dict[str, str | None] = {}
124 best_vi: dict[str, int] = {}
125 # Running top-2 scores across distinct encodings; the pruning threshold
126 # trails the second-best so the top two encodings stay exact.
127 top1_enc = ""
128 top1 = 0.0
129 top2 = 0.0
131 def record(enc_name: str, s: float, lang: str | None, vi: int) -> None:
132 nonlocal top1_enc, top1, top2
133 prev = best_score.get(enc_name)
134 if prev is not None and (s < prev or (s == prev and vi >= best_vi[enc_name])):
135 # On exact ties keep the variant that comes first in index
136 # order, matching score_best_language on the full path.
137 return
138 best_score[enc_name] = s
139 best_lang[enc_name] = lang
140 best_vi[enc_name] = vi
141 if enc_name == top1_enc:
142 top1 = s
143 elif s > top1:
144 top2 = top1
145 top1 = s
146 top1_enc = enc_name
147 elif s > top2:
148 top2 = s
150 for enc_name, lang, table, key, vi in mb_entries:
151 record(enc_name, score_with_profile(profile, table, key), lang, vi)
153 for ub, vi, enc_name, lang, table, key in sb_entries:
154 # The floor below which no rank correction can examine a candidate;
155 # everything above it must be scored exactly or detect() would
156 # diverge from the unpruned full ranking.
157 threshold = scoring_floor(top1, top2)
158 if ub < threshold:
159 # Sorted by descending bound and the threshold only rises, so
160 # no later entry can matter either.
161 break
162 record(enc_name, score_with_profile(profile, table, key), lang, vi)
164 # Force-score the encodings the corrections look up by name, when their
165 # trigger could fire. The trigger scan covers everything at or above
166 # the contract floor, because confusion resolution may promote any of
167 # those candidates to the top before postprocess evaluates its own
168 # trigger conditions.
169 trigger_floor = scoring_floor(top1, top2)
170 near_top = [e for e, s in best_score.items() if s >= trigger_floor]
171 forced = forced_encodings(near_top)
172 if forced:
173 for enc in candidates:
174 if enc.name not in forced:
175 continue
176 # Score every variant: a partially-pruned encoding may otherwise
177 # carry an understated best score into the demotion comparison.
178 for vi, (lang, table, key) in enumerate(index.get(enc.name, [])):
179 record(enc.name, score_with_profile(profile, table, key), lang, vi)
181 return [
182 (enc.name, best_score[enc.name], best_lang[enc.name])
183 for enc in candidates
184 if best_score.get(enc.name, 0.0) > 0.0
185 ]
188def score_candidates(
189 data: bytes,
190 candidates: tuple[EncodingInfo, ...],
191 *,
192 full_ranking: bool = False,
193) -> list[DetectionResult]:
194 """Score all candidates and return results sorted by confidence descending.
196 :param data: The raw byte data to score.
197 :param candidates: Encoding candidates to evaluate.
198 :param full_ranking: When ``True``, score every candidate fully so the
199 returned list is complete (needed by ``detect_all``). When ``False``
200 (the default), single-byte candidates that provably cannot affect the
201 top of the ranking may be skipped; the winner, position 1, and all
202 candidates within the confusion band of the top score are identical
203 to the full ranking.
204 :returns: A list of :class:`DetectionResult` sorted by confidence.
205 """
206 if not data or not candidates:
207 return []
209 profile = BigramProfile(data)
210 if profile.input_norm == 0.0:
211 return []
213 if full_ranking or len(profile.nonzero) < _MIN_NONZERO_FOR_PRESCREEN:
214 scores = _score_all(data, candidates, profile)
215 else:
216 scores = _score_pruned(candidates, profile)
218 scores.sort(key=lambda x: x[1], reverse=True)
219 return [
220 DetectionResult(encoding=name, confidence=s, language=lang)
221 for name, s, lang in scores
222 ]