"""Mass functions and fusion rules."""
from __future__ import annotations
import csv
import json
from collections.abc import Mapping as MappingABC
from io import StringIO
from itertools import combinations, product
from math import comb, isfinite, log2, prod
from typing import TYPE_CHECKING, Any, Iterable, Iterator, Mapping, Sequence, cast
from evidencelib.exceptions import InvalidMassError, TotalConflictError
from evidencelib.proposition import Proposition
if TYPE_CHECKING:
from evidencelib.frame import Frame
_MASS_JSON_SCHEMA = "evidencelib.mass.v2"
_LEGACY_MASS_JSON_SCHEMA = "evidencelib.mass.v1"
_EXPORT_COLUMNS = {
"m": ("Mass", "mass"),
"mass": ("Mass", "mass"),
"belief": ("Belief", "belief"),
"bel": ("Belief", "belief"),
"plausibility": ("Plausibility", "plausibility"),
"pl": ("Plausibility", "plausibility"),
"commonality": ("Commonality", "commonality"),
"q": ("Commonality", "commonality"),
}
[docs]
class MassFunction:
"""A basic belief assignment over a frame.
Parameters
----------
frame:
Frame of discernment that owns all propositions in the assignment.
values:
Mapping from propositions, proposition expressions, or iterables of
atom names to assigned masses.
validate:
Validate that masses are non-negative and sum to one.
tolerance:
Numerical tolerance used when cleaning and validating masses.
"""
normalization_tolerance = 1e-6
def __init__(
self,
frame: "Frame",
values: Mapping[Any, float],
*,
validate: bool = True,
tolerance: float = 1e-9,
) -> None:
if not isfinite(tolerance) or not 0 <= tolerance < 1:
raise ValueError("tolerance must be a finite number in the range [0, 1).")
self.frame = frame
self.tolerance = tolerance
masses: dict[Proposition, float] = {}
for key, value in values.items():
prop = frame.proposition(key)
mass = float(value)
if not isfinite(mass):
raise InvalidMassError("Mass values must be finite numbers.")
if mass < -tolerance:
raise InvalidMassError("Mass values must be non-negative.")
if abs(mass) <= tolerance:
continue
masses[prop] = masses.get(prop, 0.0) + mass
self._masses = self._clean(masses)
if validate:
self._validate_sum()
def __getitem__(self, key: str | Proposition | Iterable[str]) -> float:
return self.mass(key)
def __iter__(self) -> Iterator[tuple[Proposition, float]]:
return iter(self.items())
def __repr__(self) -> str:
body = ", ".join(f"{prop}: {value:.6g}" for prop, value in self.items())
return f"MassFunction({{{body}}})"
[docs]
def items(self) -> tuple[tuple[Proposition, float], ...]:
"""Return focal propositions and masses sorted by proposition label."""
return tuple(sorted(self._masses.items(), key=lambda item: str(item[0])))
[docs]
def focal(self) -> tuple[Proposition, ...]:
"""Return propositions with non-zero assigned mass."""
return tuple(prop for prop, _ in self.items())
[docs]
def to_dict(self, *, string_keys: bool = True) -> dict[str | Proposition, float]:
"""Return the mass assignment as a plain dictionary."""
if string_keys:
return {str(prop): value for prop, value in self.items()}
return dict(self.items())
[docs]
@classmethod
def from_dict(
cls,
frame: "Frame",
data: Mapping[Any, Any],
**kwargs: Any,
) -> "MassFunction":
"""Create a mass function from a plain or schema-wrapped dictionary.
``data`` may be a direct mapping such as ``{"A": 0.2, "A|B": 0.8}``
or the object produced by :meth:`to_json` after JSON decoding.
"""
if not isinstance(data, MappingABC):
raise TypeError("Mass data must be a mapping.")
values = cast(Mapping[Any, float], data)
if isinstance(data.get("masses"), MappingABC):
schema = data.get("schema")
if schema is not None and schema not in {
_MASS_JSON_SCHEMA,
_LEGACY_MASS_JSON_SCHEMA,
}:
raise ValueError(f"Unsupported mass JSON schema: {schema!r}.")
cls._validate_frame_metadata(frame, data.get("frame"), schema=schema)
values = cast(Mapping[Any, float], data["masses"])
return cls(frame, values, **kwargs)
[docs]
def to_json(self, *, indent: int | None = 2) -> str:
"""Serialize this mass function to a JSON string.
The JSON stores the mass assignment and lightweight frame metadata for
validation. Import still requires the caller to provide the target
frame, because hybrid DSm constraints are model semantics rather than
just mass data.
"""
data = {
"schema": _MASS_JSON_SCHEMA,
"frame": {
"atoms": list(self.frame.atoms),
"model": self.frame.model,
"region_count": self.frame.region_count,
"regions": list(self.frame.model_signature),
},
"masses": self.to_dict(),
}
return json.dumps(data, indent=indent)
[docs]
@classmethod
def from_json(
cls,
frame: "Frame",
text: str | bytes,
**kwargs: Any,
) -> "MassFunction":
"""Create a mass function from JSON produced by :meth:`to_json`."""
data = json.loads(text)
if not isinstance(data, MappingABC):
raise ValueError("Mass JSON must contain an object.")
schema = data.get("schema")
if schema is not None and schema not in {
_MASS_JSON_SCHEMA,
_LEGACY_MASS_JSON_SCHEMA,
}:
raise ValueError(f"Unsupported mass JSON schema: {schema!r}.")
return cls.from_dict(frame, data, **kwargs)
[docs]
def to_csv(
self,
*,
include_header: bool = True,
float_format: str | None = None,
) -> str:
"""Serialize this mass assignment to CSV text.
The CSV has two columns: ``proposition`` and ``mass``. It is intended
for data exchange and round trips, not for presentation tables.
"""
output = StringIO()
writer = csv.writer(output, lineterminator="\n")
if include_header:
writer.writerow(("proposition", "mass"))
for prop, value in self.items():
writer.writerow((str(prop), self._format_number(value, float_format)))
return output.getvalue()
[docs]
@classmethod
def from_csv(
cls,
frame: "Frame",
text: str,
*,
has_header: bool = True,
**kwargs: Any,
) -> "MassFunction":
"""Create a mass function from CSV text with proposition and mass columns."""
rows = csv.reader(StringIO(text))
values: dict[str, float] = {}
if has_header:
try:
header = next(rows)
except StopIteration as exc:
raise ValueError("Mass CSV is empty.") from exc
normalized = [cell.strip().lower() for cell in header]
if normalized != ["proposition", "mass"]:
raise ValueError("Mass CSV header must be: proposition,mass.")
for row_number, row in enumerate(rows, start=2 if has_header else 1):
if not row or all(not cell.strip() for cell in row):
continue
if len(row) != 2:
raise ValueError(f"Mass CSV row {row_number} must have two columns.")
proposition, value = row
try:
mass = float(value)
except ValueError as exc:
raise ValueError(f"Mass CSV row {row_number} has invalid mass {value!r}.") from exc
values[proposition] = values.get(proposition, 0.0) + mass
return cls(frame, values, **kwargs)
[docs]
def to_latex(
self,
*,
columns: Sequence[str] = ("mass",),
rows: str = "focal",
caption: str | None = None,
label: str | None = None,
float_format: str | None = ".4f",
booktabs: bool = True,
position: str = "htbp",
) -> str:
"""Export this mass function as a LaTeX table string.
Parameters
----------
columns:
Any of ``mass``, ``belief``, ``plausibility``, or ``commonality``.
Short aliases ``m``, ``bel``, ``pl``, and ``q`` are accepted.
rows:
``"focal"`` for stored non-zero masses, or ``"all"`` for every
proposition generated by the frame. ``"all"`` can be large for
DSmT frames.
caption, label:
Optional LaTeX table metadata.
float_format:
Python format specifier such as ``".4f"``. Percent-style formats
such as ``"%0.4f"`` are also accepted.
booktabs:
Use ``toprule``/``midrule``/``bottomrule`` instead of ``hline``.
position:
LaTeX table position specifier.
"""
resolved_columns = self._resolve_export_columns(columns)
row_props = self._export_rows(rows)
alignment = "l" + ("r" * len(resolved_columns))
lines = [f"\\begin{{table}}[{position}]", "\\centering"]
if caption is not None:
lines.append(f"\\caption{{{self._latex_escape_text(caption)}}}")
if label is not None:
lines.append(f"\\label{{{label}}}")
lines.append(f"\\begin{{tabular}}{{{alignment}}}")
lines.append("\\toprule" if booktabs else "\\hline")
headers = ["Proposition", *(heading for heading, _ in resolved_columns)]
lines.append(" & ".join(headers) + r" \\")
lines.append("\\midrule" if booktabs else "\\hline")
for prop in row_props:
values = [
self._format_number(self._export_value(prop, method), float_format)
for _, method in resolved_columns
]
lines.append(" & ".join((self._latex_proposition(prop), *values)) + r" \\")
lines.append("\\bottomrule" if booktabs else "\\hline")
lines.extend(["\\end{tabular}", "\\end{table}"])
return "\n".join(lines)
[docs]
def comparison_to_latex(
self,
*others: "MassFunction",
labels: Sequence[str] | None = None,
propositions: Sequence[str | Proposition | Iterable[str]] | None = None,
orientation: str = "wide",
caption: str | None = None,
label: str | None = None,
float_format: str | None = ".4f",
booktabs: bool = True,
position: str = "htbp",
source_header: str = "Source",
proposition_header: str = "Proposition",
mass_header: str = "Mass",
font_size: str | None = None,
arraystretch: float | None = None,
) -> str:
"""Export several mass assignments as one LaTeX comparison table.
``orientation="wide"`` places sources in rows and propositions in
columns. ``orientation="long"`` produces source, proposition, and mass
columns, which is more suitable when the union of focal propositions is
too wide for a page.
"""
masses = (self, *others)
self._check_sources(masses)
resolved_labels = self._resolve_comparison_labels(labels, len(masses))
if orientation not in {"wide", "long"}:
raise ValueError("orientation must be 'wide' or 'long'.")
self._validate_latex_layout(font_size, arraystretch)
if propositions is None:
selected = sorted(
{prop for mass in masses for prop in mass.focal()},
key=lambda prop: (prop.cardinality, str(prop)),
)
else:
selected = [self.frame.proposition(prop) for prop in propositions]
if not selected:
raise ValueError("At least one proposition is required.")
lines = self._latex_table_start(
caption=caption,
label=label,
position=position,
font_size=font_size,
arraystretch=arraystretch,
)
rule_top = "\\toprule" if booktabs else "\\hline"
rule_mid = "\\midrule" if booktabs else "\\hline"
rule_bottom = "\\bottomrule" if booktabs else "\\hline"
if orientation == "wide":
lines.append("\\begin{tabular}{l" + ("r" * len(selected)) + "}")
lines.append(rule_top)
headers = [
self._latex_escape_text(source_header),
*(self._latex_proposition(prop) for prop in selected),
]
lines.append(" & ".join(headers) + r" \\")
lines.append(rule_mid)
for source_label, mass in zip(resolved_labels, masses, strict=True):
values = [
self._format_number(mass.mass(prop), float_format) for prop in selected
]
lines.append(
" & ".join((self._latex_escape_text(source_label), *values)) + r" \\"
)
else:
lines.append("\\begin{tabular}{llr}")
lines.append(rule_top)
headers = [
self._latex_escape_text(source_header),
self._latex_escape_text(proposition_header),
self._latex_escape_text(mass_header),
]
lines.append(" & ".join(headers) + r" \\")
lines.append(rule_mid)
for source_index, (source_label, mass) in enumerate(
zip(resolved_labels, masses, strict=True)
):
visible = [prop for prop in selected if mass.mass(prop) > mass.tolerance]
if source_index and booktabs:
lines.append("\\addlinespace")
for prop_index, prop in enumerate(visible):
source_cell = (
self._latex_escape_text(source_label) if prop_index == 0 else ""
)
value = self._format_number(mass.mass(prop), float_format)
lines.append(
" & ".join((source_cell, self._latex_proposition(prop), value))
+ r" \\"
)
lines.extend([rule_bottom, "\\end{tabular}", "\\end{table}"])
return "\n".join(lines)
[docs]
def pignistic_comparison_to_latex(
self,
*others: "MassFunction",
labels: Sequence[str] | None = None,
hypotheses: Sequence[str] | None = None,
actions: Sequence[str] | None = None,
caption: str | None = None,
label: str | None = None,
float_format: str | None = ".4f",
booktabs: bool = True,
position: str = "htbp",
source_header: str = "Source",
action_header: str = "Action",
font_size: str | None = None,
arraystretch: float | None = None,
) -> str:
"""Export conflict and pignistic scores for several results to LaTeX."""
masses = (self, *others)
self._check_sources(masses)
resolved_labels = self._resolve_comparison_labels(labels, len(masses))
self._validate_latex_layout(font_size, arraystretch)
selected_hypotheses = tuple(hypotheses or self.frame.atoms)
if not selected_hypotheses:
raise ValueError("At least one hypothesis is required.")
unknown = [name for name in selected_hypotheses if name not in self.frame.atoms]
if unknown:
raise ValueError(f"Unknown frame hypothesis: {unknown[0]!r}.")
if actions is not None and len(actions) != len(masses):
raise ValueError("actions must have the same length as mass functions.")
lines = self._latex_table_start(
caption=caption,
label=label,
position=position,
font_size=font_size,
arraystretch=arraystretch,
)
alignment = "l" + ("r" * (len(selected_hypotheses) + 1))
if actions is not None:
alignment += "l"
lines.append(f"\\begin{{tabular}}{{{alignment}}}")
rule_top = "\\toprule" if booktabs else "\\hline"
rule_mid = "\\midrule" if booktabs else "\\hline"
rule_bottom = "\\bottomrule" if booktabs else "\\hline"
lines.append(rule_top)
headers = [
self._latex_escape_text(source_header),
r"$m(\emptyset)$",
*(
f"$\\mathrm{{BetP}}({self._latex_escape_math(name)})$"
for name in selected_hypotheses
),
]
if actions is not None:
headers.append(self._latex_escape_text(action_header))
lines.append(" & ".join(headers) + r" \\")
lines.append(rule_mid)
for index, (source_label, mass) in enumerate(
zip(resolved_labels, masses, strict=True)
):
scores = mass.pignistic()
values = [
self._format_number(mass.conflict, float_format),
*(
self._format_number(scores[name], float_format)
for name in selected_hypotheses
),
]
cells = [self._latex_escape_text(source_label), *values]
if actions is not None:
cells.append(self._latex_escape_text(actions[index]))
lines.append(" & ".join(cells) + r" \\")
lines.extend([rule_bottom, "\\end{tabular}", "\\end{table}"])
return "\n".join(lines)
@property
def total_mass(self) -> float:
"""Sum of all stored masses."""
return sum(self._masses.values())
[docs]
def mass(self, key: str | Proposition | Iterable[str]) -> float:
"""Return the direct mass assigned to a proposition."""
return self._masses.get(self.frame.proposition(key), 0.0)
[docs]
def belief(self, key: str | Proposition | Iterable[str]) -> float:
"""Return belief, the mass of propositions contained in ``key``."""
target = self.frame.proposition(key)
# Generalized bbas have m(empty)=0, so excluding empty is equivalent to
# equation (3) in that setting. The explicit guard also gives belief a
# coherent TBM interpretation for unnormalized Smets results.
return sum(value for prop, value in self._masses.items() if prop and prop <= target)
[docs]
def plausibility(self, key: str | Proposition | Iterable[str]) -> float:
"""Return plausibility, the mass of propositions intersecting ``key``."""
target = self.frame.proposition(key)
return sum(value for prop, value in self._masses.items() if prop.intersects(target))
[docs]
def commonality(self, key: str | Proposition | Iterable[str]) -> float:
"""Return commonality, the mass of propositions containing ``key``."""
target = self.frame.proposition(key)
return sum(value for prop, value in self._masses.items() if target <= prop)
@property
def conflict(self) -> float:
"""Mass assigned to the empty proposition."""
return self.mass(self.frame.empty)
# ------------------------------------------------------------------
# Uncertainty measures.
#
# Every measure uses the DSm cardinality of a proposition, its number of
# Venn regions, which equals the ordinary set cardinality |A| on Shafer
# (DST) frames and stays consistent on free and hybrid DSm models.
[docs]
def deng_entropy(self) -> float:
"""Return the Deng entropy of the assignment.
``E_d(m) = -sum_A m(A) log2(m(A) / (2^c(A) - 1))`` with ``c(A)`` the
DSm cardinality of ``A``. Equals Shannon entropy for Bayesian
assignments and :meth:`tfb_entropy` with ``order=1``.
Reference: Y. Deng, "Deng entropy", Chaos, Solitons & Fractals 91
(2016) 549-553.
"""
return self.tfb_entropy(order=1)
[docs]
def tfb_entropy(self, order: int = 1) -> float:
"""Return the k-order time fractal-based (TFB) belief entropy.
``E_k(m) = -sum_A m(A) log2(m(A) / ((k+1)^c(A) - k^c(A)))`` with
``k = order`` and ``c(A)`` the DSm cardinality of ``A``. ``order=1``
reproduces the Deng entropy. On a DST frame with ``n`` hypotheses its
maximum over assignments, ``log2((k+2)^n - (k+1)^n)``, is the k-order
higher order information volume of a mass function (HOIVMF).
Reference: Q. Zhou and Y. Deng, "Higher order information volume of
mass function", Information Sciences 586 (2022) 501-513.
"""
if order < 1:
raise ValueError("TFB entropy requires order >= 1.")
self._require_measure_input("TFB entropy")
total = 0.0
for prop, value in self._masses.items():
cardinality = len(prop.regions)
states = (order + 1) ** cardinality - order**cardinality
total -= value * log2(value / states)
return total
[docs]
def fractal_belief_entropy(self) -> float:
"""Return the fractal-based belief (FB) entropy of the assignment.
Every focal element ``G`` spreads its mass uniformly over its
``2^c(G) - 1`` non-empty sub-propositions, producing the fractal-based
assignment ``m_F``; FB entropy is the Shannon entropy of ``m_F``. It
equals Shannon entropy for Bayesian assignments and reaches
``log2(2^n - 1)`` for the vacuous one. The cost grows with
``2^c(G)``, so keep focal cardinalities moderate.
Reference: Q. Zhou and Y. Deng, "Fractal-based belief entropy",
Information Sciences (2022), preprint arXiv:2012.00235.
"""
self._require_measure_input("FB entropy")
# The spread lives on the refinement of each proposition into its Venn
# regions, the same elementary states the generalized pignistic
# transformation uses, so the 2^c(G) - 1 denominator stays consistent
# across DST, free DSm, and hybrid frames.
fractal: dict[frozenset[int], float] = {}
for prop, value in self._masses.items():
regions = sorted(prop.regions)
share = value / ((1 << len(regions)) - 1)
for size in range(1, len(regions) + 1):
for chosen in combinations(regions, size):
sub = frozenset(chosen)
fractal[sub] = fractal.get(sub, 0.0) + share
return -sum(value * log2(value) for value in fractal.values() if value > 0.0)
[docs]
def nonspecificity(self) -> float:
"""Return the generalized Hartley nonspecificity ``sum m(A) log2 c(A)``.
Reference: G. J. Klir and M. J. Wierman, "Uncertainty-Based
Information", Physica-Verlag, 1999.
"""
self._require_measure_input("Nonspecificity")
return sum(
value * log2(len(prop.regions)) for prop, value in self._masses.items()
)
[docs]
def strife(self) -> float:
"""Return Klir's strife, the conflict-based part of total uncertainty.
``S(m) = -sum_A m(A) log2(sum_B m(B) c(A & B) / c(A))``. Reduces to
Shannon entropy for Bayesian assignments.
Reference: G. J. Klir and M. J. Wierman, "Uncertainty-Based
Information", Physica-Verlag, 1999.
"""
self._require_measure_input("Strife")
total = 0.0
for prop, value in self._masses.items():
inner = sum(
other_value * len((prop & other).regions) / len(prop.regions)
for other, other_value in self._masses.items()
)
total -= value * log2(inner)
return total
def _require_measure_input(self, name: str) -> None:
if self.conflict > self.tolerance:
raise InvalidMassError(
f"{name} requires m(empty) = 0; normalize the assignment first."
)
[docs]
def conjunctive(
self,
*others: "MassFunction",
model: "Frame | None" = None,
) -> "MassFunction":
"""Unnormalized conjunctive rule.
On a free DSm frame this is the classic DSm rule (DSmC). On Shafer's
DST model, contradictory intersections are accumulated on ``empty``.
"""
return self._combine_intersection((self, *others), normalize=False, model=model)
[docs]
def dsmc(self, *others: "MassFunction") -> "MassFunction":
"""Alias for the classic conjunctive DSm rule."""
return self.conjunctive(*others)
[docs]
def smets(
self,
*others: "MassFunction",
model: "Frame | None" = None,
) -> "MassFunction":
"""Smets/TBM unnormalized rule, keeping conflict on the empty set."""
return self.conjunctive(*others, model=model)
[docs]
def dempster(
self,
*others: "MassFunction",
model: "Frame | None" = None,
) -> "MassFunction":
"""Dempster's normalized rule of combination."""
return self._combine_intersection((self, *others), normalize=True, model=model)
[docs]
def yager(
self,
*others: "MassFunction",
model: "Frame | None" = None,
) -> "MassFunction":
"""Yager's rule: transfer total conflict to total ignorance."""
conjunctive = self.conjunctive(*others, model=model)
conflict = conjunctive.conflict
masses = {prop: value for prop, value in conjunctive.items() if prop}
if conflict:
masses[conjunctive.frame.total] = (
masses.get(conjunctive.frame.total, 0.0) + conflict
)
return MassFunction(conjunctive.frame, masses, tolerance=self.tolerance)
[docs]
def dubois_prade(
self,
*others: "MassFunction",
model: "Frame | None" = None,
) -> "MassFunction":
"""Apply the static Dubois-Prade conflict-transfer rule.
Dubois-Prade is a static rule. Passing a distinct target ``model``
denotes a dynamic model change and is rejected rather than silently
returning a DSmH result.
"""
sources = (self, *others)
self._check_sources(sources)
if len(sources) != 2:
raise ValueError("Dubois-Prade requires exactly two sources.")
if model is not None and model is not self.frame:
raise ValueError(
"Dubois-Prade is defined here only for static models; "
"use dsmh(..., model=target_frame) for a dynamic model change."
)
self._require_zero_source_conflict(sources, rule="Dubois-Prade")
masses: dict[Proposition, float] = {}
for props, values in self._focal_product(sources):
amount = prod(values)
intersection = self._intersection_all(props)
target = intersection if intersection else self._union_all(props)
if not target:
raise ValueError(
"Dubois-Prade cannot preserve mass for this dynamic/non-existential case."
)
masses[target] = masses.get(target, 0.0) + amount
return MassFunction(self.frame, masses, tolerance=self.tolerance)
[docs]
def dsmh(
self,
*others: "MassFunction",
model: "Frame | None" = None,
) -> "MassFunction":
"""Apply the hybrid DSm rule, including its S1, S2, and S3 terms.
For a dynamic model change, create the source assignments on their
original frame and pass the constrained target frame as ``model``. This
preserves each focal proposition and ``u(X)`` until constraints are
applied. With no explicit model, the rule operates statically on the
sources' existing frame and rejects mass already collapsed onto empty.
"""
sources = (self, *others)
self._check_sources(sources)
target_frame = self.frame if model is None else model
self._validate_target_model(target_frame, rule="DSmH")
if any(source.conflict > source.tolerance for source in sources):
if model is None:
raise ValueError(
"DSmH cannot recover the origin of mass already collapsed onto empty. "
"Create sources on their original frame and pass the constrained "
"target explicitly with dsmh(..., model=target_frame)."
)
raise ValueError("DSmH requires source assignments with m(empty) = 0.")
masses: dict[Proposition, float] = {}
for props, values in self._focal_product(sources):
amount = prod(values)
modeled_props = tuple(
self._proposition_in_frame(prop, target_frame) for prop in props
)
intersection = self._intersection_all(modeled_props)
if intersection:
target = intersection
elif all(not prop for prop in modeled_props):
# S2 in the hybrid DSm rule: all focal elements have become
# relatively/absolutely empty. Their original atom unions u(X)
# must be retained; if those are empty too, transfer to total
# ignorance.
target = target_frame.empty
for original in props:
target = target | self._proposition_in_frame(
original.union_atoms(), target_frame
)
if not target:
target = target_frame.total
else:
# S3: a relatively empty intersection is transferred to the
# canonical disjunction of the involved focal elements.
target = self._union_all(modeled_props, frame=target_frame)
if not target:
target = target_frame.total
masses[target] = masses.get(target, 0.0) + amount
masses.pop(target_frame.empty, None)
return MassFunction(target_frame, masses, tolerance=self.tolerance)
[docs]
def pcr5(self, other: "MassFunction") -> "MassFunction":
"""PCR5 for two sources."""
return self.pcr6(other)
[docs]
def pcr6(self, *others: "MassFunction") -> "MassFunction":
"""PCR6 proportional conflict redistribution for two or more sources."""
sources = (self, *others)
self._check_sources(sources)
self._require_zero_source_conflict(sources, rule="PCR6")
masses: dict[Proposition, float] = {}
for props, values in self._focal_product(sources):
amount = prod(values)
intersection = self._intersection_all(props)
if intersection:
masses[intersection] = masses.get(intersection, 0.0) + amount
continue
denominator = sum(values)
if denominator <= self.tolerance:
continue
for prop, source_mass in zip(props, values, strict=True):
target = prop if prop else self.frame.total
if not target:
continue
share = amount * source_mass / denominator
masses[target] = masses.get(target, 0.0) + share
masses.pop(self.frame.empty, None)
return MassFunction(self.frame, masses, tolerance=self.tolerance)
[docs]
def normalize(self) -> "MassFunction":
"""Normalize a conjunctive result by removing empty-set conflict."""
conflict = self.conflict
denominator = 1.0 - conflict
if denominator <= self.tolerance:
raise TotalConflictError("Dempster normalization is undefined at total conflict.")
masses = {
prop: value / denominator
for prop, value in self._masses.items()
if prop and abs(value) > self.tolerance
}
return MassFunction(self.frame, masses, tolerance=self.tolerance)
[docs]
def pignistic_of(
self,
key: str | Proposition | Iterable[str],
*,
normalize_conflict: bool = True,
) -> float:
"""Return the generalized pignistic probability of one proposition.
This implements ``C_M(X & A) / C_M(X)`` for arbitrary ``A`` in the
frame's hyper-power set. Empty-set conflict is normalized consistently
with :meth:`pignistic` and :meth:`pignistic_regions`.
"""
target = self.frame.proposition(key)
denominator = self._pignistic_denominator(normalize_conflict)
result = 0.0
for prop, mass in self._masses.items():
if not prop:
continue
result += mass * (prop & target).cardinality / prop.cardinality / denominator
return result
[docs]
def pignistic(self, *, normalize_conflict: bool = True) -> dict[str, float]:
"""Return pignistic scores for singleton hypotheses.
This is the classical pignistic transformation on DST frames. On free
or hybrid DSmT frames, singleton hypotheses can overlap, so the returned
event scores are useful for decisions but do not have to sum to one.
If ``normalize_conflict`` is true, mass assigned to the empty proposition
is ignored and the remaining scores are rescaled by ``1 - conflict``.
This makes TBM/Smets results usable for pignistic decisions while still
allowing raw unnormalized scores with ``normalize_conflict=False``.
"""
return {
name: self.pignistic_of(atom, normalize_conflict=normalize_conflict)
for name, atom in zip(self.frame.atoms, self.frame.symbols(), strict=True)
}
[docs]
def pignistic_regions(self, *, normalize_conflict: bool = True) -> dict[str, float]:
"""Return a probability distribution over model Venn regions.
If ``normalize_conflict`` is true, empty-set conflict is excluded and
the non-empty region probabilities are rescaled by ``1 - conflict``.
"""
denominator = self._pignistic_denominator(normalize_conflict)
result = {self._format_region(region): 0.0 for region in self.frame._universe}
for prop, mass in self._masses.items():
if not prop:
continue
cardinality = prop.cardinality
if cardinality == 0:
continue
share = mass / cardinality / denominator
for region in prop.regions:
result[self._format_region(region)] += share
return result
[docs]
def decision(self) -> str:
"""Return the singleton with the largest pignistic probability."""
probabilities = self.pignistic()
return max(probabilities, key=probabilities.__getitem__)
[docs]
def plot(self, *, ax: Any = None, **kwargs: Any) -> Any:
"""Plot this mass assignment as a horizontal bar chart.
This method requires the optional plotting dependency. Install it with
``pip install 'evidencelib[plot]'``.
"""
from evidencelib.plotting import plot_mass
return plot_mass(self, ax=ax, **kwargs)
[docs]
def plot_comparison(
self,
*others: "MassFunction",
labels: Sequence[str] | None = None,
ax: Any = None,
**kwargs: Any,
) -> Any:
"""Plot a heatmap comparing this mass assignment with other sources."""
from evidencelib.plotting import plot_mass_comparison
return plot_mass_comparison((self, *others), labels=labels, ax=ax, **kwargs)
[docs]
def plot_belief_plausibility(self, *, ax: Any = None, **kwargs: Any) -> Any:
"""Plot belief-plausibility intervals for this mass assignment."""
from evidencelib.plotting import plot_belief_plausibility
return plot_belief_plausibility(self, ax=ax, **kwargs)
[docs]
def plot_pignistic_decision(self, *, ax: Any = None, **kwargs: Any) -> Any:
"""Plot the pignistic decision ranking for this mass assignment."""
from evidencelib.plotting import plot_pignistic_decision
return plot_pignistic_decision(self, ax=ax, **kwargs)
[docs]
def plot_venn(self, *, ax: Any = None, **kwargs: Any) -> Any:
"""Plot pignistic or direct mass values over disjoint Venn regions."""
from evidencelib.plotting import plot_venn
return plot_venn(self, ax=ax, **kwargs)
@classmethod
def _from_unchecked(
cls,
frame: "Frame",
values: Mapping[Any, float],
*,
tolerance: float = 1e-9,
) -> "MassFunction":
return cls(frame, values, validate=False, tolerance=tolerance)
def _combine_intersection(
self,
sources: tuple["MassFunction", ...],
*,
normalize: bool,
model: "Frame | None" = None,
) -> "MassFunction":
self._check_sources(sources)
target_frame = self.frame if model is None else model
self._validate_target_model(target_frame, rule="fusion")
masses: dict[Proposition, float] = {}
for props, values in self._focal_product(sources):
modeled_props = tuple(
self._proposition_in_frame(prop, target_frame) for prop in props
)
target = self._intersection_all(modeled_props)
masses[target] = masses.get(target, 0.0) + prod(values)
result = MassFunction(target_frame, masses, tolerance=self.tolerance)
return result.normalize() if normalize else result
def _focal_product(
self,
sources: tuple["MassFunction", ...],
) -> Iterator[tuple[tuple[Proposition, ...], tuple[float, ...]]]:
item_groups = [source.items() for source in sources]
for combo in product(*item_groups):
props = tuple(prop for prop, _ in combo)
values = tuple(value for _, value in combo)
yield props, values
def _intersection_all(self, props: Iterable[Proposition]) -> Proposition:
iterator = iter(props)
result = next(iterator)
for prop in iterator:
result = result & prop
return result
def _union_all(
self,
props: Iterable[Proposition],
*,
frame: "Frame | None" = None,
) -> Proposition:
owner = self.frame if frame is None else frame
result = owner.empty
for prop in props:
result = result | prop
return result
def _proposition_in_frame(self, prop: Proposition, frame: "Frame") -> Proposition:
if prop.frame is frame:
return prop
if prop.is_empty:
return frame.empty
return frame.proposition(str(prop))
def _pignistic_denominator(self, normalize_conflict: bool) -> float:
if not normalize_conflict:
return 1.0
denominator = 1.0 - self.conflict
if denominator <= self.tolerance:
raise TotalConflictError("Pignistic transformation is undefined at total conflict.")
return denominator
def _validate_target_model(self, target: "Frame", *, rule: str) -> None:
if target.atoms != self.frame.atoms:
raise ValueError(
f"The {rule} target model must use the same ordered frame atoms."
)
if not set(target.model_signature) <= set(self.frame.model_signature):
raise ValueError(
f"The {rule} target model may add constraints but cannot make "
"regions possible that were absent from the source frame."
)
def _check_sources(self, sources: tuple["MassFunction", ...]) -> None:
if len(sources) < 2:
raise ValueError("At least two sources are required.")
if any(source.frame is not self.frame for source in sources):
raise ValueError("All mass functions must belong to the same frame.")
@staticmethod
def _require_zero_source_conflict(
sources: tuple["MassFunction", ...],
*,
rule: str,
) -> None:
if any(source.conflict > source.tolerance for source in sources):
raise ValueError(f"{rule} requires source assignments with m(empty) = 0.")
def _validate_sum(self) -> None:
total = sum(self._masses.values())
difference = abs(total - 1.0)
if difference == 0:
return
if difference <= max(self.tolerance, self.normalization_tolerance):
self._masses = {prop: value / total for prop, value in self._masses.items()}
return
raise InvalidMassError(f"Mass values must sum to 1.0, got {total}.")
def _clean(self, masses: Mapping[Proposition, float]) -> dict[Proposition, float]:
return {
prop: value
for prop, value in masses.items()
if abs(value) > self.tolerance
}
def _format_region(self, region: int) -> str:
names = [name for i, name in enumerate(self.frame.atoms) if region & (1 << i)]
return "&".join(names)
@staticmethod
def _validate_frame_metadata(
frame: "Frame",
metadata: Any,
*,
schema: Any = None,
) -> None:
if schema == _LEGACY_MASS_JSON_SCHEMA and frame.model == "hybrid":
raise ValueError(
"Legacy v1 JSON cannot safely identify hybrid-model constraints; "
"re-export the data with schema v2."
)
if metadata is None:
if frame.model == "hybrid":
raise ValueError(
"Hybrid-model mass JSON must include exact frame-region metadata."
)
return
if not isinstance(metadata, MappingABC):
raise ValueError("Mass frame metadata must be an object.")
atoms = metadata.get("atoms")
if atoms is not None and tuple(atoms) != frame.atoms:
raise ValueError("Mass data frame atoms do not match the target frame.")
model = metadata.get("model")
if model is not None and model != frame.model:
raise ValueError("Mass data frame model does not match the target frame.")
region_count = metadata.get("region_count")
if region_count is not None and int(region_count) != frame.region_count:
raise ValueError("Mass data frame region count does not match the target frame.")
regions = metadata.get("regions")
if regions is not None and tuple(int(region) for region in regions) != frame.model_signature:
raise ValueError("Mass data model constraints do not match the target frame.")
if frame.model == "hybrid" and regions is None:
raise ValueError(
"Hybrid-model mass JSON must include exact frame-region metadata."
)
@staticmethod
def _format_number(value: float, float_format: str | None) -> str:
if float_format is None:
return str(value)
if "%" in float_format:
return float_format % value
return format(value, float_format)
@staticmethod
def _resolve_comparison_labels(
labels: Sequence[str] | None,
count: int,
) -> tuple[str, ...]:
if labels is None:
return tuple(f"source {index + 1}" for index in range(count))
if len(labels) != count:
raise ValueError("labels must have the same length as mass functions.")
return tuple(labels)
@staticmethod
def _validate_latex_layout(
font_size: str | None,
arraystretch: float | None,
) -> None:
allowed_sizes = {"small", "footnotesize", "scriptsize"}
if font_size is not None and font_size not in allowed_sizes:
choices = ", ".join(sorted(allowed_sizes))
raise ValueError(f"font_size must be one of {choices}, or None.")
if arraystretch is not None and (
not isfinite(arraystretch) or arraystretch <= 0
):
raise ValueError("arraystretch must be a positive finite number.")
@classmethod
def _latex_table_start(
cls,
*,
caption: str | None,
label: str | None,
position: str,
font_size: str | None,
arraystretch: float | None,
) -> list[str]:
lines = [f"\\begin{{table}}[{position}]", "\\centering"]
if font_size is not None:
lines.append(f"\\{font_size}")
if arraystretch is not None:
lines.append(f"\\renewcommand{{\\arraystretch}}{{{arraystretch:g}}}")
if caption is not None:
lines.append(f"\\caption{{{cls._latex_escape_text(caption)}}}")
if label is not None:
lines.append(f"\\label{{{label}}}")
return lines
@staticmethod
def _resolve_export_columns(columns: Sequence[str]) -> tuple[tuple[str, str], ...]:
if not columns:
raise ValueError("At least one export column is required.")
resolved: list[tuple[str, str]] = []
for column in columns:
key = column.lower().strip()
try:
resolved.append(_EXPORT_COLUMNS[key])
except KeyError as exc:
choices = ", ".join(sorted(_EXPORT_COLUMNS))
raise ValueError(f"Unknown export column {column!r}; choose from {choices}.") from exc
return tuple(resolved)
def _export_rows(self, rows: str) -> tuple[Proposition, ...]:
if rows == "focal":
return self.focal()
if rows == "all":
return self.frame.elements()
raise ValueError("rows must be 'focal' or 'all'.")
def _export_value(self, prop: Proposition, method: str) -> float:
if method == "mass":
return self.mass(prop)
if method == "belief":
return self.belief(prop)
if method == "plausibility":
return self.plausibility(prop)
if method == "commonality":
return self.commonality(prop)
raise AssertionError(f"Unhandled export method: {method}")
@classmethod
def _latex_proposition(cls, prop: Proposition) -> str:
if not prop:
return r"$\emptyset$"
terms = []
for term in str(prop).split("|"):
factors = [cls._latex_escape_math(part) for part in term.split("&")]
terms.append(r" \cap ".join(factors))
return "$" + r" \cup ".join(terms) + "$"
@staticmethod
def _latex_escape_text(value: str) -> str:
replacements = {
"\\": r"\textbackslash{}",
"&": r"\&",
"%": r"\%",
"$": r"\$",
"#": r"\#",
"_": r"\_",
"{": r"\{",
"}": r"\}",
"~": r"\textasciitilde{}",
"^": r"\textasciicircum{}",
}
return "".join(replacements.get(char, char) for char in value)
@classmethod
def _latex_escape_math(cls, value: str) -> str:
return cls._latex_escape_text(value).replace(r"\textbackslash{}", r"\backslash{}")