diff --git a/tools/src/pop_fem_audit_tools/commands/build_db.py b/tools/src/pop_fem_audit_tools/commands/build_db.py index a5ebaf0..7beea58 100644 --- a/tools/src/pop_fem_audit_tools/commands/build_db.py +++ b/tools/src/pop_fem_audit_tools/commands/build_db.py @@ -77,10 +77,12 @@ from sqlalchemy.orm import Session from ..database import Base, ds from ..models import ( + Annotation, Artist, ChartEntry, CodeGroup, Coding, + Pattern, Role, Song, SongArtist, @@ -132,6 +134,13 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace: parser.add_argument( "--gender-corrections", type=Path, default=None, help="the gender correction table CSV file to apply") + parser.add_argument( + "--patterns", type=Path, default=None, + help="the pattern definition table CSV file to import") + parser.add_argument( + "--annotations", type=Path, default=None, + help="the settled pattern annotation table CSV file to" + " import") return parser.parse_args(argv) @@ -986,6 +995,216 @@ class GroupImporter: f"{path}: missing column(s): {', '.join(missing)}") +class PatternImporter: + """The pattern-import job: loads the pattern definition table + into the working store.""" + + COLUMNS: tuple[str, ...] = ( + "Pattern", "Group", "Name", "Description") + """The required columns of the pattern CSV file.""" + + def __init__(self, session: Session) -> None: + """Initialize the importer. + + :param session: The database session. + """ + self.__session: Session = session + + def import_patterns(self, path: Path | None) -> None: + """Load the pattern definition table into the store. + + A None input leaves the patterns unloaded. Otherwise + every row of the CSV file stores one pattern, with the + pattern ID, the group, the name, and the description + stored verbatim. When the method returns, the imported + patterns are queryable in the session. + + :param path: The pattern definition table CSV file to + import, or None to skip the patterns. + :return: None. + :raises BuildError: When the file lacks a required + column, or two rows name the same pattern ID. + :raises OSError: When the file cannot be read. + """ + if path is None: + return + seen: set[str] = set() + with open(path, encoding="utf-8", newline="") as file: + reader: csv.DictReader[str] = csv.DictReader(file) + self.__check_columns(path, reader.fieldnames) + row: dict[str, str] + for row in reader: + self.__import_pattern_row(path, seen, row) + self.__session.flush() + + def __import_pattern_row(self, path: Path, seen: set[str], + row: dict[str, str]) -> None: + """Store one pattern row. + + :param path: The pattern CSV file, for the error messages. + :param seen: The pattern IDs already stored, updated with + this row's pattern ID. + :param row: The pattern CSV row. + :return: None. + :raises BuildError: When the pattern ID repeats an + earlier row. + """ + if row["Pattern"] in seen: + raise BuildError( + f"{path}: duplicated pattern \"{row['Pattern']}\"") + seen.add(row["Pattern"]) + self.__session.add(Pattern( + pattern=row["Pattern"], group=row["Group"], + name=row["Name"], description=row["Description"])) + + @classmethod + def __check_columns(cls, path: Path, + fieldnames: Sequence[str] | None) -> None: + """Verify the pattern CSV file has the required columns. + + :param path: The pattern CSV file. + :param fieldnames: The header row of the file, or None + when the file is empty. + :return: None. + :raises BuildError: When a required column is absent. + """ + header: Sequence[str] = fieldnames or () + missing: list[str] = [ + x for x in cls.COLUMNS if x not in header] + if len(missing) > 0: + raise BuildError( + f"{path}: missing column(s): {', '.join(missing)}") + + +class AnnotationImporter: + """The annotation-import job: loads the settled pattern + annotation table onto the stored songs.""" + + COLUMNS: tuple[str, ...] = ( + "Song", "Artist Credit", "Pattern", "Votes") + """The required columns of the annotation CSV file.""" + + def __init__(self, session: Session) -> None: + """Initialize the importer. + + :param session: The database session. + """ + self.__session: Session = session + + def import_annotations(self, path: Path | None) -> None: + """Load the settled pattern annotation table onto the + stored songs. + + A None input leaves the annotations unloaded. Otherwise + every row of the CSV file yields one annotation linking + the song named by its title and artist credit to the + stored pattern named by its pattern ID, with the integer + vote count stored verbatim. Import the pattern table + first (see `PatternImporter`), so every named pattern is + already stored. When the method returns, the imported + annotations are queryable in the session. + + :param path: The settled pattern annotation table CSV + file to import, or None to skip the annotations. + :return: None. + :raises BuildError: When the file lacks a required + column, a row names a song that the store does not + have, a row names a pattern that the store does not + have, a votes field is not an integer, or two rows + name the same song and pattern. + :raises OSError: When the file cannot be read. + """ + if path is None: + return + songs: dict[tuple[str, str], Song] = { + (x.title, x.artist_credit): x + for x in self.__session.scalars(sa.select(Song))} + patterns: dict[str, Pattern] = { + x.pattern: x + for x in self.__session.scalars(sa.select(Pattern))} + seen: set[tuple[int, str]] = set() + with open(path, encoding="utf-8", newline="") as file: + reader: csv.DictReader[str] = csv.DictReader(file) + self.__check_columns(path, reader.fieldnames) + row: dict[str, str] + for row in reader: + self.__import_annotation( + path, songs, patterns, seen, row) + self.__session.flush() + + def __import_annotation( + self, path: Path, songs: dict[tuple[str, str], Song], + patterns: dict[str, Pattern], + seen: set[tuple[int, str]], + row: dict[str, str]) -> None: + """Store one annotation row. + + :param path: The annotation CSV file, for the error + messages. + :param songs: The stored songs, keyed by the title and + the artist credit. + :param patterns: The stored patterns, keyed by the + pattern ID. + :param seen: The (song ID, pattern ID) pairs already + stored, updated with the pair of this row. + :param row: The annotation CSV row. + :return: None. + :raises BuildError: When the row names a song that the + store does not have, names a pattern that the store + does not have, its votes field is not an integer, or + its song and pattern repeat an earlier row. + """ + key: tuple[str, str] = (row["Song"], row["Artist Credit"]) + song: Song | None = songs.get(key) + if song is None: + raise BuildError( + f"{path}: no song \"{row['Song']}\" by" + f" \"{row['Artist Credit']}\"") + pattern: Pattern | None = patterns.get(row["Pattern"]) + if pattern is None: + raise BuildError( + f"{path}: no pattern \"{row['Pattern']}\"") + annotation_key: tuple[int, str] = ( + song.id, pattern.pattern) + if annotation_key in seen: + raise BuildError( + f"{path}: duplicated annotation: \"{row['Song']}\"" + f" by \"{row['Artist Credit']}\", pattern" + f" \"{row['Pattern']}\"") + seen.add(annotation_key) + votes: int + try: + votes = int(row["Votes"]) + except ValueError as error: + raise BuildError( + f"{path}: \"{row['Song']}\" by" + f" \"{row['Artist Credit']}\", pattern" + f" \"{row['Pattern']}\": votes" + f" \"{row['Votes']}\" is not an integer" + ) from error + self.__session.add(Annotation( + song=song, pattern=pattern, votes=votes)) + + @classmethod + def __check_columns(cls, path: Path, + fieldnames: Sequence[str] | None) -> None: + """Verify the annotation CSV file has the required + columns. + + :param path: The annotation CSV file. + :param fieldnames: The header row of the file, or None + when the file is empty. + :return: None. + :raises BuildError: When a required column is absent. + """ + header: Sequence[str] = fieldnames or () + missing: list[str] = [ + x for x in cls.COLUMNS if x not in header] + if len(missing) > 0: + raise BuildError( + f"{path}: missing column(s): {', '.join(missing)}") + + @dataclass class StoreCounts: """The row counts of the working store, for the build summary.""" @@ -998,6 +1217,10 @@ class StoreCounts: """The number of the settled codings.""" groups: int """The number of the settled code group members.""" + patterns: int + """The number of the stored patterns.""" + annotations: int + """The number of the settled pattern annotations.""" @classmethod def get_instance(cls, session: Session) -> Self: @@ -1020,7 +1243,12 @@ class StoreCounts: codings=count( sa.select(sa.func.count()).select_from(Coding)), groups=count( - sa.select(sa.func.count()).select_from(CodeGroup))) + sa.select(sa.func.count()).select_from(CodeGroup)), + patterns=count( + sa.select(sa.func.count()).select_from(Pattern)), + annotations=count( + sa.select(sa.func.count()) + .select_from(Annotation))) def reset_store(session: Session) -> None: @@ -1030,8 +1258,8 @@ def reset_store(session: Session) -> None: :return: None. """ model: type[Base] - for model in (CodeGroup, Coding, SongArtist, ChartEntry, Song, - Artist): + for model in (CodeGroup, Coding, Annotation, SongArtist, + ChartEntry, Song, Pattern, Artist): session.execute(sa.delete(model)) @@ -1197,6 +1425,9 @@ def main(argv: list[str] | None = None) -> int: args.gender_corrections) CodingImporter(session).import_codings(args.codings) GroupImporter(session).import_groups(args.groups) + PatternImporter(session).import_patterns(args.patterns) + AnnotationImporter(session).import_annotations( + args.annotations) counts = StoreCounts.get_instance(session) CSVExporter(session, args.derived_dir).write() session.commit() @@ -1209,6 +1440,8 @@ def main(argv: list[str] | None = None) -> int: elapsed: str = format_duration(time.monotonic() - started) print(f"Done. {counts.songs} songs/{counts.artists} artists" f"/{counts.codings} codings/{counts.groups} group" - f" members. {elapsed} elapsed.", + f" members/{counts.patterns} patterns" + f"/{counts.annotations} annotations. {elapsed}" + " elapsed.", file=sys.stderr) return 0 diff --git a/tools/src/pop_fem_audit_tools/models.py b/tools/src/pop_fem_audit_tools/models.py index ae53220..6af6256 100644 --- a/tools/src/pop_fem_audit_tools/models.py +++ b/tools/src/pop_fem_audit_tools/models.py @@ -55,6 +55,9 @@ class Song(Base): codings: Mapped[list[Coding]] \ = relationship(back_populates="song") """The settled codings of the song.""" + annotations: Mapped[list[Annotation]] \ + = relationship(back_populates="song") + """The settled pattern annotations of the song.""" __table_args__ = (sa.UniqueConstraint(title, artist_credit),) """The table-level constraints.""" @@ -153,3 +156,43 @@ class CodeGroup(Base): votes: Mapped[int] = mapped_column() """The number of the selection runs that selected the keyword for the group.""" + + +class Pattern(Base): + """A stored gendered-language pattern definition.""" + __tablename__ = "patterns" + """The table name.""" + + pattern: Mapped[str] = mapped_column(primary_key=True) + """The pattern ID, as the pattern table carries it.""" + group: Mapped[str] = mapped_column() + """The pattern group: male, female, or mixed.""" + name: Mapped[str] = mapped_column() + """The pattern name.""" + description: Mapped[str] = mapped_column() + """The pattern description.""" + annotations: Mapped[list[Annotation]] \ + = relationship(back_populates="pattern") + """The settled annotations assigning the pattern to a song.""" + + +class Annotation(Base): + """A settled pattern annotation of a song, with its vote + count.""" + __tablename__ = "annotations" + """The table name.""" + + song_id: Mapped[int] = mapped_column(sa.ForeignKey(Song.id), + primary_key=True) + """The ID of the annotated song.""" + pattern_id: Mapped[str] = mapped_column( + sa.ForeignKey(Pattern.pattern), primary_key=True) + """The ID of the assigned pattern.""" + votes: Mapped[int] = mapped_column() + """The number of the annotation runs that assigned the + pattern to the song.""" + song: Mapped[Song] = relationship(back_populates="annotations") + """The annotated song.""" + pattern: Mapped[Pattern] \ + = relationship(back_populates="annotations") + """The assigned pattern.""" diff --git a/tools/tests/test_build_db.py b/tools/tests/test_build_db.py index 9e9deec..f60ed46 100644 --- a/tools/tests/test_build_db.py +++ b/tools/tests/test_build_db.py @@ -19,10 +19,12 @@ from pop_fem_audit_tools import config from pop_fem_audit_tools.commands import build_db from pop_fem_audit_tools.database import DataSource from pop_fem_audit_tools.models import ( + Annotation, Artist, ChartEntry, CodeGroup, Coding, + Pattern, Role, Song, ) @@ -272,6 +274,8 @@ class TestBuildDB(unittest.TestCase): self.__groups: Path = self.__dir / "groups.csv" self.__gender_corrections: Path = \ self.__dir / "gender_corrections.csv" + self.__patterns: Path = self.__dir / "patterns.csv" + self.__annotations: Path = self.__dir / "annotations.csv" self.__write_chart(self.CHART_CSV) config.set_settings(config.Settings( SQLALCHEMY_DATABASE_URL="sqlite://", @@ -1348,3 +1352,153 @@ class TestBuildDB(unittest.TestCase): self.assertIn(str(self.__gender_corrections), stderr) session: Session = self.__session() self.assertEqual(list(session.scalars(sa.select(Song))), []) + + PATTERNS_CSV: str = ( + "Pattern,Group,Name,Description\n" + "M1,male,Dominance,A pattern of asserted dominance.\n" + "F1,female,Nurture,A pattern of caretaking language.\n") + """The pattern CSV fixture: one male and one female pattern.""" + + ANNOTATIONS_CSV: str = ( + "Song,Artist Credit,Pattern,Votes\n" + "Hello,Adele,M1,3\n" + "One Dance,Drake featuring Wizkid,F1,2\n") + """The annotation CSV fixture linking two songs, each to one + of the pattern fixture's patterns.""" + + def __write_patterns(self, content: str) -> None: + """Write the pattern CSV fixture. + + :param content: The CSV content. + :return: None. + """ + self.__patterns.write_text(content, encoding="utf-8") + + def __write_annotations(self, content: str) -> None: + """Write the annotation CSV fixture. + + :param content: The CSV content. + :return: None. + """ + self.__annotations.write_text(content, encoding="utf-8") + + def __stored_patterns(self) -> dict[str, tuple[str, str, str]]: + """Read the stored patterns keyed by the pattern ID. + + :return: The stored (group, name, description) tuples, + keyed by the pattern ID. + """ + session: Session = self.__session() + return {x.pattern: (x.group, x.name, x.description) + for x in session.scalars(sa.select(Pattern))} + + def __stored_annotations(self) -> dict[tuple[str, str], int]: + """Read the stored annotations keyed by song and pattern. + + :return: The stored votes, keyed by the song title and + the pattern ID. + """ + session: Session = self.__session() + return {(x.song.title, x.pattern.pattern): x.votes + for x in session.scalars(sa.select(Annotation))} + + def test_patterns_and_annotations_imported(self) -> None: + """Test that the pattern and annotation CSVs import + together, reflected in the final counts message.""" + self.__write_patterns(self.PATTERNS_CSV) + self.__write_annotations(self.ANNOTATIONS_CSV) + status: int + stderr: str + status, stderr = self.__run_build( + "--patterns", str(self.__patterns), + "--annotations", str(self.__annotations)) + self.assertEqual(status, 0) + self.assertIn("2 patterns", stderr) + self.assertIn("2 annotations", stderr) + self.assertEqual( + self.__stored_patterns(), + {"M1": ("male", "Dominance", + "A pattern of asserted dominance."), + "F1": ("female", "Nurture", + "A pattern of caretaking language.")}) + self.assertEqual( + self.__stored_annotations(), + {("Hello", "M1"): 3, + ("One Dance", "F1"): 2}) + + def test_omitted_patterns_and_annotations_import_nothing( + self) -> None: + """Test that omitting both options leaves the pattern and + annotation tables empty.""" + self.__write_patterns(self.PATTERNS_CSV) + self.__write_annotations(self.ANNOTATIONS_CSV) + status: int + stderr: str + status, stderr = self.__run_build() + self.assertEqual(status, 0) + self.assertIn("0 patterns", stderr) + self.assertIn("0 annotations", stderr) + self.assertEqual(self.__stored_patterns(), {}) + self.assertEqual(self.__stored_annotations(), {}) + + def test_annotations_unknown_song_fails(self) -> None: + """Test that an annotation row naming an unknown song + fails the build, naming the file and the offending title + and credit.""" + self.__write_patterns(self.PATTERNS_CSV) + self.__write_annotations( + "Song,Artist Credit,Pattern,Votes\n" + "Nowhere,Nobody,M1,3\n") + status: int + stderr: str + status, stderr = self.__run_build( + "--patterns", str(self.__patterns), + "--annotations", str(self.__annotations)) + self.assertNotEqual(status, 0) + self.assertIn(str(self.__annotations), stderr) + self.assertIn("no song \"Nowhere\" by \"Nobody\"", stderr) + self.assertEqual(self.__stored_annotations(), {}) + + def test_annotations_unknown_pattern_fails(self) -> None: + """Test that an annotation row naming a pattern the store + does not have fails the build, naming the file and the + offending pattern ID.""" + self.__write_annotations( + "Song,Artist Credit,Pattern,Votes\n" + "Hello,Adele,X9,3\n") + status: int + stderr: str + status, stderr = self.__run_build( + "--annotations", str(self.__annotations)) + self.assertNotEqual(status, 0) + self.assertIn(str(self.__annotations), stderr) + self.assertIn("no pattern \"X9\"", stderr) + self.assertEqual(self.__stored_annotations(), {}) + + def test_patterns_missing_column_fails(self) -> None: + """Test that a pattern CSV missing a required column fails + the build.""" + self.__write_patterns( + "Pattern,Group,Name\n" + "M1,male,Dominance\n") + status: int + stderr: str + status, stderr = self.__run_build( + "--patterns", str(self.__patterns)) + self.assertNotEqual(status, 0) + self.assertIn("missing column(s): Description", stderr) + self.assertEqual(self.__stored_patterns(), {}) + + def test_annotations_missing_column_fails(self) -> None: + """Test that an annotation CSV missing a required column + fails the build.""" + self.__write_annotations( + "Song,Artist Credit,Pattern\n" + "Hello,Adele,M1\n") + status: int + stderr: str + status, stderr = self.__run_build( + "--annotations", str(self.__annotations)) + self.assertNotEqual(status, 0) + self.assertIn("missing column(s): Votes", stderr) + self.assertEqual(self.__stored_annotations(), {})