Restructure tally-groups into the house command shape

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tools/src/pop_fem_audit_tools/commands/tally_groups.py

@@ -8,46 +8,281 @@
 Settles the semantic code groups of step 4: the same group
 selection definition file is run three times independently, and
 this command counts the votes and writes the final group table
-the paper cites, as the CSV file given as the last positional
+the paper cites.  A (group, keyword) pair is written out when at
-command-line argument.  A (group, keyword) pair is written out
+least two of the three runs select it.  When an input is
-when at least two of the three runs select it, so three votes
+malformed, the tally fails and nothing is written; the error
-never tie.  A selected item that is not in the valid keyword
+message names what failed.
-list is invalid and casts no vote; every dropped occurrence is
-reported on standard error.  The group name is the record ID
-with its ``group-`` prefix dropped.  The rows are ordered by the
-group name and then by the keyword, by Unicode code point, and
-the file carries the header row ``Group,Keyword,Votes`` with
-CRLF line endings per RFC 4180.
-
-The three archives must cover exactly the same set of group IDs,
-every record ID must carry the ``group-`` prefix, and every
-record's "text" must parse to a JSON array of strings; otherwise
-the tally fails and nothing is written.
 """
 import argparse
 import csv
 import json
 import sys
 import time
+from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, ClassVar
 
 from ..utils import format_duration
 
-GROUP_ID_PREFIX: str = "group-"
-"""The prefix every group record ID must carry; the group name
-is the rest of the ID."""
-MAJORITY: int = 2
-"""The number of runs that must select a keyword for a group for
-that pair to be settled."""
-HEADER: tuple[str, str, str] = ("Group", "Keyword", "Votes")
-"""The header row of the group table CSV file."""
-
 
 class TallyError(Exception):
     """An error that fails the group tally."""
 
 
+@dataclass(frozen=True)
+class TalliedGroups:
+    """The outcome of settling the group table."""
+
+    codes: int
+    """The number of settled (group, keyword) pairs."""
+    groups: int
+    """The number of groups the three runs cover."""
+
+
+class GroupTallier:
+    """The tallier of the three group-selection runs' votes."""
+
+    __GROUP_ID_PREFIX: ClassVar[str] = "group-"
+    """The prefix every group record ID must carry; the group
+    name is the rest of the ID."""
+    __MAJORITY: ClassVar[int] = 2
+    """The number of runs that must select a keyword for a group
+    for that pair to be settled."""
+    __HEADER: ClassVar[tuple[str, str, str]] \
+        = ("Group", "Keyword", "Votes")
+    """The header row of the group table CSV file."""
+
+    def __init__(self, run_dir_1: Path, run_dir_2: Path,
+                 run_dir_3: Path, valid_keywords_txt: Path,
+                 output_csv: Path) -> None:
+        """Set up the tallier of the three selection runs.
+
+        :param run_dir_1: The first selection run's archive
+            directory, containing ``output.jsonl``.
+        :param run_dir_2: The second selection run's archive
+            directory, containing ``output.jsonl``.
+        :param run_dir_3: The third selection run's archive
+            directory, containing ``output.jsonl``.
+        :param valid_keywords_txt: The plain text file of the
+            allowed keywords, one per line.
+        :param output_csv: The output group table CSV file.
+        """
+        self.__run_dirs: list[Path] = [
+            run_dir_1, run_dir_2, run_dir_3]
+        """The three runs' archive directories, in the given
+        order."""
+        self.__valid_keywords_txt: Path = valid_keywords_txt
+        """The plain text file of the allowed keywords, one per
+        line."""
+        self.__output_csv: Path = output_csv
+        """The output group table CSV file."""
+
+    def run(self) -> TalliedGroups:
+        """Load the three runs, tally their votes, and write the
+        table.
+
+        :return: The settled code count and the group count.
+        :raises TallyError: When a file cannot be read, a line is
+            not a well-formed output record, a record is not a
+            successful result, an ID lacks the ``group-`` prefix,
+            a group has two records, a "text" does not parse to a
+            JSON array of strings, or the three runs do not cover
+            the same set of groups.
+        :raises OSError: When the output file cannot be written.
+        """
+        valid: set[str] = self.__load_valid_keywords(
+            self.__valid_keywords_txt)
+        runs: list[dict[str, set[str]]] = [
+            self.__load_run(x) for x in self.__run_dirs]
+        self.__drop_invalid(runs, self.__run_dirs, valid)
+        rows: list[tuple[str, str, int]] = self.__tally(runs)
+        self.__write_csv(rows)
+        return TalliedGroups(codes=len(rows), groups=len(runs[0]))
+
+    @staticmethod
+    def __load_valid_keywords(path: Path) -> set[str]:
+        """Load the valid keyword list.
+
+        :param path: The plain text file of the allowed keywords,
+            one per line.
+        :return: The allowed keywords.
+        :raises TallyError: When the file cannot be read or holds
+            no keyword.
+        """
+        text: str
+        try:
+            text = path.read_text(encoding="utf-8")
+        except OSError as error:
+            raise TallyError(str(error)) from error
+        keywords: set[str] = {x.strip() for x in text.split("\n")
+                              if x.strip() != ""}
+        if len(keywords) == 0:
+            raise TallyError(f"{path}: no keywords")
+        return keywords
+
+    @classmethod
+    def __load_run(cls, run_dir: Path) -> dict[str, set[str]]:
+        """Load and validate the selection records of one run.
+
+        :param run_dir: The run's archive directory, containing
+            ``output.jsonl``.
+        :return: The selected keywords of every group of the run,
+            keyed by the group name, the duplicates within one
+            record's selection collapsed.
+        :raises TallyError: When the file cannot be read, a line
+            is not a well-formed output record, a record is not a
+            successful result, an ID lacks the ``group-`` prefix,
+            a group has two records, or a "text" does not parse
+            to a JSON array of strings.
+        """
+        path: Path = run_dir / "output.jsonl"
+        text: str
+        try:
+            text = path.read_text(encoding="utf-8")
+        except OSError as error:
+            raise TallyError(str(error)) from error
+        records: dict[str, set[str]] = {}
+        line: str
+        for line in text.split("\n"):
+            if line.strip() == "":
+                continue
+            record: Any
+            try:
+                record = json.loads(line)
+            except json.JSONDecodeError as error:
+                raise TallyError(
+                    f"{path}: malformed JSON: {error}") from error
+            item_id: Any = record["id"]
+            if "text" not in record:
+                raise TallyError(
+                    f"{path}: id {item_id}: not a successful"
+                    " result")
+            if not isinstance(item_id, str) \
+                    or not item_id.startswith(
+                        cls.__GROUP_ID_PREFIX) \
+                    or item_id == cls.__GROUP_ID_PREFIX:
+                raise TallyError(
+                    f"{path}: id {item_id}: not in the"
+                    f" \"{cls.__GROUP_ID_PREFIX}<name>\" form")
+            group: str = item_id[len(cls.__GROUP_ID_PREFIX):]
+            if group in records:
+                raise TallyError(
+                    f"{path}: id {item_id}: duplicate record")
+            records[group] = cls.__parse_selection(
+                record["text"], f"{path}: id {item_id}")
+        return records
+
+    @staticmethod
+    def __parse_selection(text: Any, label: str) -> set[str]:
+        """Parse and validate the selected keywords of one record.
+
+        :param text: The "text" field of the record.
+        :param label: The location of the record, for the error
+            message.
+        :return: The selected keywords, the duplicates collapsed.
+        :raises TallyError: When the text does not parse to a
+            JSON array of strings.
+        """
+        selected: Any
+        try:
+            selected = json.loads(text)
+        except json.JSONDecodeError as error:
+            raise TallyError(
+                f"{label}: \"text\" is malformed JSON:"
+                f" {error}") from error
+        if not isinstance(selected, list) \
+                or not all(isinstance(x, str) for x in selected):
+            raise TallyError(
+                f"{label}: \"text\" does not parse to a JSON"
+                " array of strings")
+        return set(selected)
+
+    @staticmethod
+    def __drop_invalid(runs: list[dict[str, set[str]]],
+                       run_dirs: list[Path],
+                       valid: set[str]) -> None:
+        """Drop the out-of-vocabulary selections of every run.
+
+        Every dropped occurrence is reported on standard error as
+        an observable side effect.
+
+        :param runs: The runs' records, filtered in place.
+        :param run_dirs: The run directories, for the messages.
+        :param valid: The allowed keywords.
+        :return: None.
+        """
+        records: dict[str, set[str]]
+        run_dir: Path
+        for records, run_dir in zip(runs, run_dirs):
+            group: str
+            selected: set[str]
+            for group, selected in records.items():
+                keyword: str
+                for keyword in sorted(selected - valid):
+                    print(
+                        f"note: {run_dir.name} group-{group}:"
+                        f" dropped out-of-vocabulary item"
+                        f" \"{keyword}\"", file=sys.stderr)
+                records[group] = selected & valid
+
+    @classmethod
+    def __tally(cls, runs: list[dict[str, set[str]]]) \
+            -> list[tuple[str, str, int]]:
+        """Tally the keyword votes of the runs, group by group.
+
+        :param runs: The runs' records, all covering the same set
+            of groups.
+        :return: The settled rows, each the group name, the
+            keyword, and the number of votes, ordered by the
+            group name and then by the keyword, by Unicode code
+            point.
+        :raises TallyError: When the runs do not cover the same
+            set of groups.
+        """
+        groups: set[str] = set(runs[0])
+        records: dict[str, set[str]]
+        for records in runs[1:]:
+            if set(records) != groups:
+                raise TallyError(
+                    "the three runs do not cover the same"
+                    " groups: " + ", ".join(sorted(
+                        groups.symmetric_difference(
+                            set(records)))))
+        rows: list[tuple[str, str, int]] = []
+        group: str
+        for group in sorted(groups):
+            votes: dict[str, int] = {}
+            for records in runs:
+                keyword: str
+                for keyword in records[group]:
+                    votes[keyword] = votes.get(keyword, 0) + 1
+            rows.extend(
+                (group, x, votes[x])
+                for x in sorted(votes) if votes[x] >= cls.__MAJORITY)
+        return rows
+
+    def __write_csv(self, rows: list[tuple[str, str, int]]) \
+            -> None:
+        """Write the group table CSV file.
+
+        Writes an RFC 4180 CSV file, UTF-8, with CRLF line
+        endings, carrying the header row ``Group,Keyword,Votes``
+        and one row per settled (group, keyword) pair, in the row
+        order.  The parent directory is created when it does not
+        exist.
+
+        :param rows: The settled rows, in the output order.
+        :return: None.
+        :raises OSError: When the file cannot be written.
+        """
+        self.__output_csv.parent.mkdir(parents=True, exist_ok=True)
+        with open(self.__output_csv, "w", encoding="utf-8",
+                  newline="") as file:
+            writer: Any = csv.writer(file)
+            writer.writerow(self.__HEADER)
+            writer.writerows(rows)
+
+
 def parse_args(argv: list[str] | None) -> argparse.Namespace:
     """Parse the command-line arguments.
 
@@ -68,238 +303,34 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
         "run_dir_3", type=Path,
         help="the third selection run's archive directory")
     parser.add_argument(
-        "valid_keywords", type=Path,
+        "output_csv", type=Path,
+        help="the output CSV file")
+    parser.add_argument(
+        "--valid-keywords", type=Path, required=True,
         help="a plain text file of the allowed keywords, one per"
              " line")
-    parser.add_argument(
-        "output_csv", type=Path,
-        help="the output CSV file, by convention"
-             " results/groups.csv")
     return parser.parse_args(argv)
 
 
-def load_valid_keywords(path: Path) -> set[str]:
-    """Load the valid keyword list.
-
-    :param path: The plain text file of the allowed keywords, one
-        per line.
-    :return: The allowed keywords.
-    :raises TallyError: When the file cannot be read or holds no
-        keyword.
-    """
-    text: str
-    try:
-        text = path.read_text(encoding="utf-8")
-    except OSError as error:
-        raise TallyError(str(error)) from error
-    keywords: set[str] = {x.strip() for x in text.split("\n")
-                          if x.strip() != ""}
-    if len(keywords) == 0:
-        raise TallyError(f"{path}: no keywords")
-    return keywords
-
-
-def load_run(run_dir: Path) -> dict[str, set[str]]:
-    """Load and validate the selection records of one run.
-
-    :param run_dir: The run's archive directory, containing
-        ``output.jsonl``.
-    :return: The selected keywords of every group of the run,
-        keyed by the group name, the duplicates within one
-        record's selection collapsed.
-    :raises TallyError: When the file cannot be read, a line is
-        not a well-formed output record, a record is not a
-        successful result, an ID lacks the ``group-`` prefix, a
-        group has two records, or a "text" does not parse to a
-        JSON array of strings.
-    """
-    path: Path = run_dir / "output.jsonl"
-    text: str
-    try:
-        text = path.read_text(encoding="utf-8")
-    except OSError as error:
-        raise TallyError(str(error)) from error
-    records: dict[str, set[str]] = {}
-    line: str
-    for line in text.split("\n"):
-        if line.strip() == "":
-            continue
-        record: Any
-        try:
-            record = json.loads(line)
-        except json.JSONDecodeError as error:
-            raise TallyError(
-                f"{path}: malformed JSON: {error}") from error
-        if not isinstance(record, dict) or "id" not in record:
-            raise TallyError(
-                f"{path}: record without \"id\": {line}")
-        item_id: Any = record["id"]
-        if "error" in record or "text" not in record:
-            raise TallyError(
-                f"{path}: id {item_id}: not a successful result")
-        if not isinstance(item_id, str) \
-                or not item_id.startswith(GROUP_ID_PREFIX) \
-                or item_id == GROUP_ID_PREFIX:
-            raise TallyError(
-                f"{path}: id {item_id}: not in the"
-                f" \"{GROUP_ID_PREFIX}<name>\" form")
-        group: str = item_id[len(GROUP_ID_PREFIX):]
-        if group in records:
-            raise TallyError(
-                f"{path}: id {item_id}: duplicate record")
-        records[group] = _selection(
-            record["text"], f"{path}: id {item_id}")
-    if len(records) == 0:
-        raise TallyError(f"{path}: no records")
-    return records
-
-
-def _selection(text: Any, label: str) -> set[str]:
-    """Parse and validate the selected keywords of one record.
-
-    :param text: The "text" field of the record.
-    :param label: The location of the record, for the error
-        message.
-    :return: The selected keywords, the duplicates collapsed.
-    :raises TallyError: When the text does not parse to a JSON
-        array of strings.
-    """
-    if not isinstance(text, str):
-        raise TallyError(f"{label}: \"text\" is not a string")
-    selected: Any
-    try:
-        selected = json.loads(text)
-    except json.JSONDecodeError as error:
-        raise TallyError(
-            f"{label}: \"text\" is malformed JSON:"
-            f" {error}") from error
-    if not isinstance(selected, list) \
-            or not all(isinstance(x, str) for x in selected):
-        raise TallyError(
-            f"{label}: \"text\" does not parse to a JSON array"
-            " of strings")
-    return set(selected)
-
-
-def drop_invalid(runs: list[dict[str, set[str]]],
-                 run_dirs: list[Path],
-                 valid: set[str]) -> None:
-    """Drop the out-of-vocabulary selections of every run.
-
-    Every dropped occurrence is reported on standard error as an
-    observable side effect.
-
-    :param runs: The runs' records, filtered in place.
-    :param run_dirs: The run directories, for the messages.
-    :param valid: The allowed keywords.
-    :return: None.
-    """
-    records: dict[str, set[str]]
-    run_dir: Path
-    for records, run_dir in zip(runs, run_dirs):
-        group: str
-        selected: set[str]
-        for group, selected in records.items():
-            keyword: str
-            for keyword in sorted(selected - valid):
-                print(
-                    f"note: {run_dir.name} group-{group}:"
-                    f" dropped out-of-vocabulary item"
-                    f" \"{keyword}\"", file=sys.stderr)
-            records[group] = selected & valid
-
-
-def tally(runs: list[dict[str, set[str]]]) \
-        -> list[tuple[str, str, int]]:
-    """Tally the keyword votes of the runs, group by group.
-
-    :param runs: The runs' records, all covering the same set of
-        groups.
-    :return: The settled rows, each the group name, the keyword,
-        and the number of votes, ordered by the group name and
-        then by the keyword, by Unicode code point.
-    :raises TallyError: When the runs do not cover the same set
-        of groups.
-    """
-    groups: set[str] = set(runs[0])
-    records: dict[str, set[str]]
-    for records in runs[1:]:
-        if set(records) != groups:
-            raise TallyError(
-                "the three runs do not cover the same groups: "
-                + ", ".join(sorted(
-                    groups.symmetric_difference(set(records)))))
-    rows: list[tuple[str, str, int]] = []
-    group: str
-    for group in sorted(groups):
-        votes: dict[str, int] = {}
-        for records in runs:
-            keyword: str
-            for keyword in records[group]:
-                votes[keyword] = votes.get(keyword, 0) + 1
-        rows.extend(
-            (group, x, votes[x])
-            for x in sorted(votes) if votes[x] >= MAJORITY)
-    return rows
-
-
-def write_csv(output_csv: Path,
-              rows: list[tuple[str, str, int]]) -> None:
-    """Write the group table CSV file.
-
-    Writes an RFC 4180 CSV file, UTF-8, with CRLF line endings,
-    carrying the header row ``Group,Keyword,Votes`` and one row
-    per settled (group, keyword) pair, in the row order.  The
-    parent directory is created when it does not exist.
-
-    :param output_csv: The output CSV file.
-    :param rows: The settled rows, in the output order.
-    :return: None.
-    :raises OSError: When the file cannot be written.
-    """
-    output_csv.parent.mkdir(parents=True, exist_ok=True)
-    with open(output_csv, "w", encoding="utf-8",
-              newline="") as file:
-        writer: Any = csv.writer(file)
-        writer.writerow(HEADER)
-        writer.writerows(rows)
-
-
 def main(argv: list[str] | None = None) -> int:
     """Settle the code groups by a majority of the three runs.
 
-    Writes the final group table as the given CSV file, holding
-    the header row ``Group,Keyword,Votes`` and one row per
-    (group, keyword) pair at least two of the three runs select,
-    ordered by the group name and then by the keyword.  A
-    selected item that is not in the valid keyword list casts no
-    vote, each dropped occurrence reported on standard error.
-    Nothing is written when the three archives do not cover the
-    same groups, a record is not a successful result, an ID lacks
-    the ``group-`` prefix, or a record's "text" does not parse to
-    a JSON array of strings; the error message names what failed.
-
     :param argv: The command-line arguments, or None for
         ``sys.argv``.
     :return: The exit status: 0 on success, non-zero on failure.
     """
     started: float = time.monotonic()
     args: argparse.Namespace = parse_args(argv)
-    run_dirs: list[Path] = [
-        args.run_dir_1, args.run_dir_2, args.run_dir_3]
     try:
-        valid: set[str] = load_valid_keywords(args.valid_keywords)
+        tallied: TalliedGroups = GroupTallier(
-        runs: list[dict[str, set[str]]] = [
+            args.run_dir_1, args.run_dir_2, args.run_dir_3,
-            load_run(x) for x in run_dirs]
+            args.valid_keywords, args.output_csv).run()
-        drop_invalid(runs, run_dirs, valid)
-        rows: list[tuple[str, str, int]] = tally(runs)
-        write_csv(args.output_csv, rows)
     except (TallyError, OSError) as error:
         print(f"error: {error}", file=sys.stderr)
         return 1
     elapsed: str = format_duration(time.monotonic() - started)
     print(
-        f"Done.  Settled {len(rows)} codes across"
+        f"Done.  Settled {tallied.codes} codes across"
-        f" {len(runs[0])} groups.  {elapsed} elapsed.",
+        f" {tallied.groups} groups.  {elapsed} elapsed.",
         file=sys.stderr)
     return 0

tools/tests/test_tally_groups.py

@@ -64,7 +64,8 @@ class TestTallyGroups(unittest.TestCase):
         with redirect_stderr(stderr):
             status = tally_groups.main(
                 [str(x) for x in self.__runs]
-                + [str(self.__valid), str(self.__output_csv)])
+                + [str(self.__output_csv),
+                   "--valid-keywords", str(self.__valid)])
         return status, stderr.getvalue()
 
     def __read_rows(self) -> list[list[str]]: