Update app.py

README.md

@@ -4,8 +4,8 @@ emoji: 💞
 colorFrom: green
 colorTo: purple
 sdk: gradio
-python_version: 3.11
-sdk_version: 5.12.0
+python_version: 3.14
+sdk_version: 6.5.1
 app_file: app.py
 pinned: true
 license: apache-2.0

app.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python3
 
+from __future__ import annotations
+
 # Copyright 2023 Dmitry Ustalov
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -17,8 +19,7 @@
 __author__ = "Dmitry Ustalov"
 __license__ = "Apache 2.0"
 
-from collections.abc import Callable
-from typing import BinaryIO, cast
+from typing import TYPE_CHECKING, Protocol, cast
 
 import evalica
 import gradio as gr
@@ -27,7 +28,9 @@ import numpy as np
 import pandas as pd
 import plotly.express as px
 from evalica import Winner
-from plotly.graph_objects import Figure
+
+if TYPE_CHECKING:
+    from plotly.graph_objects import Figure
 
 TOLERANCE, LIMIT = 1e-6, 100
 
@@ -42,49 +45,87 @@ def visualize(df_pairwise: pd.DataFrame) -> Figure:
     return fig
 
 
-def counting(xs: "pd.Series[str]", ys: "pd.Series[str]",
-             ws: "pd.Series[Winner]", index: dict[str, int]) -> "pd.Series[float]":  # type: ignore[type-var]
+def counting(
+    xs: pd.Series[str],
+    ys: pd.Series[str],
+    ws: pd.Series[Winner],
+    index: pd.Index,
+) -> pd.Series[float]:
     result = evalica.counting(xs, ys, ws, index=index)
     return result.scores
 
 
-def average_win_rate(xs: "pd.Series[str]", ys: "pd.Series[str]",
-                     ws: "pd.Series[Winner]", index: dict[str, int]) -> "pd.Series[float]":  # type: ignore[type-var]
-    result = evalica.counting(xs, ys, ws, index=index)
+def average_win_rate(
+    xs: pd.Series[str],
+    ys: pd.Series[str],
+    ws: pd.Series[Winner],
+    index: pd.Index,
+) -> pd.Series[float]:
+    result = evalica.average_win_rate(xs, ys, ws, index=index)
     return result.scores
 
 
-def bradley_terry(xs: "pd.Series[str]", ys: "pd.Series[str]",
-                  ws: "pd.Series[Winner]", index: dict[str, int]) -> "pd.Series[float]":  # type: ignore[type-var]
+def bradley_terry(
+    xs: pd.Series[str],
+    ys: pd.Series[str],
+    ws: pd.Series[Winner],
+    index: pd.Index,
+) -> pd.Series[float]:
     result = evalica.bradley_terry(xs, ys, ws, index=index, tolerance=TOLERANCE, limit=LIMIT)
     return result.scores
 
 
-def elo(xs: "pd.Series[str]", ys: "pd.Series[str]",
-        ws: "pd.Series[Winner]", index: dict[str, int]) -> "pd.Series[float]":  # type: ignore[type-var]
+def elo(
+    xs: pd.Series[str],
+    ys: pd.Series[str],
+    ws: pd.Series[Winner],
+    index: pd.Index,
+) -> pd.Series[float]:
     result = evalica.elo(xs, ys, ws, index=index)
     return result.scores
 
 
-def eigen(xs: "pd.Series[str]", ys: "pd.Series[str]",
-          ws: "pd.Series[Winner]", index: dict[str, int]) -> "pd.Series[float]":  # type: ignore[type-var]
+def eigen(
+    xs: pd.Series[str],
+    ys: pd.Series[str],
+    ws: pd.Series[Winner],
+    index: pd.Index,
+) -> pd.Series[float]:
     result = evalica.eigen(xs, ys, ws, index=index, tolerance=TOLERANCE, limit=LIMIT)
     return result.scores
 
 
-def pagerank(xs: "pd.Series[str]", ys: "pd.Series[str]",
-             ws: "pd.Series[Winner]", index: dict[str, int]) -> "pd.Series[float]":  # type: ignore[type-var]
+def pagerank(
+    xs: pd.Series[str],
+    ys: pd.Series[str],
+    ws: pd.Series[Winner],
+    index: pd.Index,
+) -> pd.Series[float]:
     result = evalica.pagerank(xs, ys, ws, index=index, tolerance=TOLERANCE, limit=LIMIT)
     return result.scores
 
 
-def newman(xs: "pd.Series[str]", ys: "pd.Series[str]",
-           ws: "pd.Series[Winner]", index: dict[str, int]) -> "pd.Series[float]":  # type: ignore[type-var]
+def newman(
+    xs: pd.Series[str],
+    ys: pd.Series[str],
+    ws: pd.Series[Winner],
+    index: pd.Index,
+) -> pd.Series[float]:
     result = evalica.newman(xs, ys, ws, index=index, tolerance=TOLERANCE, limit=LIMIT)
     return result.scores
 
 
-ALGORITHMS = {
+class CallableAlgorithm(Protocol):
+    def __call__(
+        self,
+        xs: pd.Series[str],
+        ys: pd.Series[str],
+        ws: pd.Series[Winner],
+        index: pd.Index,
+    ) -> pd.Series[float]: ...
+
+
+ALGORITHMS: dict[str, CallableAlgorithm] = {
     "Counting": counting,
     "Average Win Rate": average_win_rate,
     "Bradley-Terry (1952)": bradley_terry,
@@ -97,19 +138,22 @@ ALGORITHMS = {
 
 def largest_strongly_connected_component(df_pairs: pd.DataFrame) -> set[str]:
     G = nx.from_pandas_edgelist(df_pairs, source="left", target="right", create_using=nx.DiGraph)
-    H = nx.from_pandas_edgelist(df_pairs[df_pairs["winner"] == "tie"], source="right", target="left",
-                                create_using=nx.DiGraph)
+    H = nx.from_pandas_edgelist(
+        df_pairs[df_pairs["winner"] == "tie"],
+        source="right",
+        target="left",
+        create_using=nx.DiGraph,
+    )
     F = nx.compose(G, H)
     largest = max(nx.strongly_connected_components(F), key=len)
-    return cast(set[str], largest)
+    return cast("set[str]", largest)
 
 
-def estimate(df_pairs: pd.DataFrame,
-             algorithm: Callable[[  # type: ignore[type-var]
-                 "pd.Series[str]", "pd.Series[str]", "pd.Series[Winner]", dict[str, int]],
-             "pd.Series[float]",
-             ],
-             index: dict[str, int]) -> pd.DataFrame:
+def estimate(
+    df_pairs: pd.DataFrame,
+    algorithm: CallableAlgorithm,
+    index: pd.Index,
+) -> pd.DataFrame:
     scores = algorithm(df_pairs["left"], df_pairs["right"], df_pairs["winner"], index)
 
     df_result = pd.DataFrame(data={"score": scores}, index=index)
@@ -118,13 +162,12 @@ def estimate(df_pairs: pd.DataFrame,
     return df_result
 
 
-def bootstrap(df_pairs: pd.DataFrame,
-              algorithm: Callable[[  # type: ignore[type-var]
-                  "pd.Series[str]", "pd.Series[str]", "pd.Series[Winner]", dict[str, int]],
-              "pd.Series[float]",
-              ],
-              index: dict[str, int],
-              rounds: int) -> pd.DataFrame:
+def bootstrap(
+    df_pairs: pd.DataFrame,
+    algorithm: CallableAlgorithm,
+    index: pd.Index,
+    rounds: int,
+) -> pd.DataFrame:
     scores: list[pd.Series[float]] = []  # assuming model names are strings
 
     for r in range(rounds):
@@ -136,11 +179,16 @@ def bootstrap(df_pairs: pd.DataFrame,
 
     df_bootstrap = pd.DataFrame(scores, columns=index)
 
-    ratings = df_bootstrap.quantile(.5)
+    ratings = df_bootstrap.quantile(0.5)
 
-    ci = df_bootstrap.apply(lambda row: (
-        row.quantile(.025).item(), row.quantile(.975).item(),
-    ), axis=0, result_type="reduce")
+    ci = df_bootstrap.apply(
+        lambda row: (
+            row.quantile(0.025).item(),
+            row.quantile(0.975).item(),
+        ),
+        axis=0,
+        result_type="reduce",
+    )
 
     df_result = pd.DataFrame({"score": ratings, "ci": ci})
     df_result.index.name = "item"
@@ -149,11 +197,11 @@ def bootstrap(df_pairs: pd.DataFrame,
 
 
 def handler(
-        file: BinaryIO,
-        algorithm: str,
-        filtered: bool,
-        truncated: bool,
-        rounds: int,
+    file: str | None,
+    algorithm: str,
+    filtered: bool,
+    truncated: bool,
+    rounds: int,
 ) -> tuple[pd.DataFrame, Figure]:
     if file is None:
         raise gr.Error("File must be uploaded")
@@ -162,27 +210,31 @@ def handler(
         raise gr.Error(f"Unknown algorithm: {algorithm}")
 
     try:
-        df_pairs = pd.read_csv(file.name, dtype=str)
+        df_pairs = pd.read_csv(file, dtype=str)
     except ValueError as e:
         raise gr.Error(f"Parsing error: {e}") from e
 
     if not pd.Series(["left", "right", "winner"]).isin(df_pairs.columns).all():
         raise gr.Error("Columns must exist: left, right, winner")
 
-    if not df_pairs["winner"].isin(pd.Series(["left", "right", "tie"])).all():
+    if not df_pairs["winner"].str.lower().isin(pd.Series(["left", "right", "tie"])).all():
         raise gr.Error("Allowed winner values: left, right, tie")
 
     df_pairs = df_pairs[["left", "right", "winner"]]
-    df_pairs["winner"] = df_pairs["winner"].map(
-        {"left": Winner.X, "right": Winner.Y, "tie": Winner.Draw},
+    df_pairs["winner"] = (
+        df_pairs["winner"]
+        .str.lower()
+        .map(
+            {"left": Winner.X, "right": Winner.Y, "tie": Winner.Draw},
+        )
     )
 
-    df_pairs = df_pairs.dropna(axis=0)
+    df_pairs = df_pairs.loc[df_pairs.notna().all(axis=1)]
 
     if filtered:
         largest = largest_strongly_connected_component(df_pairs)
-
-        df_pairs = df_pairs.drop(df_pairs[~(df_pairs["left"].isin(largest) & df_pairs["right"].isin(largest))].index)
+        mask = df_pairs["left"].isin(largest) & df_pairs["right"].isin(largest)
+        df_pairs = df_pairs.loc[mask]
 
     *_, index = evalica.indexing(xs=df_pairs["left"], ys=df_pairs["right"])
 
@@ -191,11 +243,18 @@ def handler(
     else:
         df_result = estimate(df_pairs, ALGORITHMS[algorithm], index)
 
-    df_result["pairs"] = pd.Series(0, dtype=int, index=index).add(
-        df_pairs.groupby("left")["left"].count(), fill_value=0,
-    ).add(
-        df_pairs.groupby("right")["right"].count(), fill_value=0,
-    ).astype(int)
+    df_result["pairs"] = (
+        pd.Series(0, dtype=int, index=index)
+        .add(
+            df_pairs.groupby("left")["left"].count(),
+            fill_value=0,
+        )
+        .add(
+            df_pairs.groupby("right")["right"].count(),
+            fill_value=0,
+        )
+        .astype(int)
+    )
 
     df_result["rank"] = df_result["score"].rank(na_option="bottom", ascending=False).astype(int)
 
@@ -204,7 +263,7 @@ def handler(
     df_result = df_result.reset_index()
 
     if truncated:
-        df_result = pd.concat((df_result.head(5), df_result.tail(5)), copy=False)
+        df_result = pd.concat((df_result.head(5), df_result.tail(5)))
         df_result = df_result[~df_result.index.duplicated(keep="last")]
 
     pairwise = evalica.pairwise_scores(df_result["score"].to_numpy())
@@ -224,8 +283,64 @@ def handler(
     return df_result, fig
 
 
+def alpha_handler(file: str | None, distance: str) -> pd.DataFrame:
+    if file is None:
+        raise gr.Error("File must be uploaded")
+
+    try:
+        df_ratings = pd.read_csv(file, header=None, dtype=str)
+    except ValueError as e:
+        raise gr.Error(f"Parsing error: {e}") from e
+
+    if df_ratings.empty:
+        raise gr.Error("The file is empty")
+
+    try:
+        result = evalica.alpha(df_ratings, distance=distance)  # type: ignore[arg-type]
+    except evalica.InsufficientRatingsError as e:
+        raise gr.Error("Insufficient ratings: no units have at least 2 ratings") from e
+    except evalica.UnknownDistanceError as e:
+        raise gr.Error(f"Unknown distance: {e}") from e
+    except Exception as e:
+        raise gr.Error(f"Computation error: {e}") from e
+
+    return pd.DataFrame(
+        {
+            "Metric": ["Alpha", "Observed Disagreement", "Expected Disagreement"],
+            "Value": [result.alpha, result.observed, result.expected],
+        },
+    )
+
+
+def alpha_interface() -> gr.Interface:
+    return gr.Interface(
+        fn=alpha_handler,
+        inputs=[
+            gr.File(
+                file_types=[".csv", ".tsv"],
+                label="Ratings Matrix (CSV without header)",
+            ),
+            gr.Dropdown(
+                choices=["nominal", "ordinal", "interval", "ratio"],
+                value="nominal",
+                label="Distance Metric",
+                info="Nominal for categorical, ordinal for ordered categories, interval/ratio for numeric scales",
+            ),
+        ],
+        outputs=[
+            gr.Dataframe(
+                headers=["Metric", "Value"],
+                label="Inter-Rater Reliability",
+            ),
+        ],
+        title="Krippendorff's Alpha",
+        analytics_enabled=False,
+        flagging_mode="never",
+    )
+
+
 def main() -> None:
-    iface = gr.Interface(
+    pairwise_iface = gr.Interface(
         fn=handler,
         inputs=[
             gr.File(
@@ -233,7 +348,7 @@ def main() -> None:
                 label="Comparisons",
             ),
             gr.Dropdown(
-                choices=cast(list[str], ALGORITHMS),
+                choices=list(ALGORITHMS),
                 value="Bradley-Terry (1952)",
                 label="Algorithm",
             ),
@@ -241,15 +356,14 @@ def main() -> None:
                 value=False,
                 label="Largest SCC",
                 info="Bradley-Terry, Eigenvector, and Newman algorithms require the comparison graph "
-                     "to be strongly-connected. "
-                     "This option keeps only the largest strongly-connected component (SCC) of the input graph. "
-                     "Some items might be missing as a result of this filtering.",
+                "to be strongly-connected. "
+                "This option keeps only the largest strongly-connected component (SCC) of the input graph. "
+                "Some items might be missing as a result of this filtering.",
             ),
             gr.Checkbox(
                 value=False,
                 label="Truncate Output",
-                info="Perform the entire computation but output only five head and five tail items, "
-                     "avoiding overlap.",
+                info="Perform the entire computation but output only five head and five tail items, avoiding overlap.",
             ),
             gr.Number(
                 value=0,
@@ -307,6 +421,13 @@ As the output, this tool provides a table with items, their estimated scores, an
         analytics_enabled=False,
     )
 
+    iface = gr.TabbedInterface(
+        [pairwise_iface, alpha_interface()],
+        ["Pairwise Ranking", "Krippendorff's Alpha"],
+        title="Evalica",
+        analytics_enabled=False,
+    )
+
     iface.launch()
 
 

mypy.ini

@@ -1,4 +1,3 @@
 [mypy]
 ignore_missing_imports = True
-plugins = numpy.typing.mypy_plugin
 strict = True

requirements.txt

@@ -1,3 +1,3 @@
-evalica[gradio]
+evalica[gradio] == 0.4.0rc2
 networkx
 plotly

ruff.toml

@@ -1,5 +1,5 @@
 line-length = 120
-target-version = "py311"
+target-version = "py314"
 
 [lint]
 select = ["ALL"]