import argparse import csv import pathlib import shutil import subprocess import sys from collections import defaultdict import numpy as np SCRIPT_DIR = pathlib.Path(__file__).resolve().parent SCALE_COLS = ["xcu", "cu", "mcu", "m", "ml", "l", "xls"] CATEGORY_LABELS = { "xcu": "Extreme close-up", "cu": "Close-up", "mcu": "Medium close-up", "m": "Medium", "ml": "Medium long", "l": "Long", "xls": "Extreme long", } COLLECTIONS = [("popular", "is_popular"), ("prestige", "is_prestige"), ("indie", "is_indie")] MIN_GENRE_MOVIES = 30 def load_genres(path): genres = {} with open(path, newline="") as f: for row in csv.DictReader(f, delimiter="\t"): if row["id"]: genres[row["id"]] = row["genres"] return genres def load_film_level(path): movies = [] with open(path, newline="") as f: for row in csv.DictReader(f): counts = {c: float(row[c]) for c in SCALE_COLS} total = sum(counts.values()) if total <= 0: continue movies.append({ "id": row["id"], "year": float(row["year"]), "is_popular": row["is_popular"] == "1", "is_prestige": row["is_prestige"] == "1", "is_indie": row["is_indie"] == "1", "shares": {c: v / total for c, v in counts.items()}, }) return movies def bootstrap_by_movie(buckets, values, group_name, n_boot=1000, seed=0): values = np.asarray(values, dtype=np.float64) if len(values) == 0: return [] if np.ndim(buckets) == 0: bucket_keys, bucket_codes = np.array([buckets]), np.zeros(len(values), dtype=np.int64) else: buckets = np.asarray(buckets) bucket_keys, bucket_codes = np.unique(buckets, return_inverse=True) rng = np.random.default_rng(seed) out = [] for code, key in enumerate(bucket_keys): vals = values[bucket_codes == code] n_movies = len(vals) if n_movies == 0: continue resamples = vals[rng.integers(0, n_movies, size=(n_boot, n_movies))] boot_means = resamples.mean(axis=1) out.append({ group_name: key.item() if isinstance(key, np.generic) else key, "mean": float(vals.mean()), "lo": float(np.percentile(boot_means, 2.5)), "hi": float(np.percentile(boot_means, 97.5)), "n_movies": n_movies, }) return out def build_bootstrapped(movies, bucket_fn, group_name): bootstrapped_values = [] for collection, flag_col in COLLECTIONS: in_collection = [m for m in movies if m[flag_col]] buckets = np.array([bucket_fn(m) for m in in_collection]) for cat in SCALE_COLS: shares = np.array([m["shares"][cat] for m in in_collection]) for row in bootstrap_by_movie(buckets, shares, group_name): row["category"] = CATEGORY_LABELS.get(cat) row["collection"] = collection bootstrapped_values.append(row) return bootstrapped_values def build_mcu_by_genre(movies, genre_by_id): movies_by_genre = defaultdict(list) for m in movies: if not (m["is_popular"] or m["is_prestige"] or m["is_indie"]): continue for genre in (genre_by_id.get(m["id"], "") or "").split(","): genre = genre.strip() if genre: movies_by_genre[genre].append(m) rows = [] for genre in sorted(movies_by_genre): shares = np.array([m["shares"]["mcu"] for m in movies_by_genre[genre]]) rows += bootstrap_by_movie(genre, shares, group_name="genre") rows.sort(key=lambda row: -row["mean"]) return rows def write_genre_table(path, rows): rows = [row for row in rows if row["n_movies"] >= MIN_GENRE_MOVIES] body = "\n".join( f"{row['genre']} & {100 * row['mean']:.1f} ({100 * row['lo']:.1f}, " f"{100 * row['hi']:.1f}) & {row['n_movies']} \\\\" for row in rows ) path.write_text( "% Generated by 01_temporal_trends.py -- do not edit by hand.\n" "% Requires \\usepackage{booktabs}.\n" "\\begin{table}[t]\n" "\\centering\n" "\\begin{tabular}{lcr}\n" "\\toprule\n" "Genre & Medium close-up (\\%) & Movies \\\\\n" "\\midrule\n" f"{body}\n" "\\bottomrule\n" "\\end{tabular}\n" "\\caption{Share of screen time at medium close-up by genre, with 95\\% bootstrap " "confidence intervals. Each movie contributes one observation and resampling is at " "the level of movies; a movie with several genres counts toward each of them, so the " f"rows are not independent. Genres with fewer than {MIN_GENRE_MOVIES} movies are " "omitted.}\n" "\\label{tab:mcu-by-genre}\n" "\\end{table}\n" ) def write_csv(path, rows): with open(path, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()) if rows else []) writer.writeheader() writer.writerows(rows) def render_figures(r_script_path): if shutil.which("Rscript") is None: sys.exit(f"Rscript not found on PATH -- the tables are written, but the figures " f"need R. Install R (with ggplot2) and run: Rscript {r_script_path}") subprocess.run(["Rscript", str(r_script_path)], check=True) def main(): parser = argparse.ArgumentParser() parser.add_argument("--aggregate", default=str( SCRIPT_DIR / ".." / "data" / "metadata" / "movie_metadata - Aggregate.tsv")) parser.add_argument("--film_level", default=str( SCRIPT_DIR / ".." / "data" / "analysis_outputs" / "film_level_categories.csv")) parser.add_argument("--intermediate_dir", default=str(SCRIPT_DIR / ".." / "data" / "analysis_outputs")) parser.add_argument("--article_dir", default=str(SCRIPT_DIR / ".." / "figures" / "article")) args = parser.parse_args() genre_by_id = load_genres(args.aggregate) movies = load_film_level(args.film_level) intermediate_dir = pathlib.Path(args.intermediate_dir) intermediate_dir.mkdir(parents=True, exist_ok=True) article_dir = pathlib.Path(args.article_dir) article_dir.mkdir(parents=True, exist_ok=True) write_csv(intermediate_dir / "bootstrap_intervals_by_year.csv", build_bootstrapped(movies, lambda m: m["year"], "year")) write_csv(intermediate_dir / "bootstrap_intervals_by_decade.csv", build_bootstrapped(movies, lambda m: np.floor(m["year"] / 10) * 10, "decade")) mcu_by_genre = build_mcu_by_genre(movies, genre_by_id) write_csv(intermediate_dir / "mcu_by_genre.csv", mcu_by_genre) write_genre_table(article_dir / "mcu_by_genre.tex", mcu_by_genre) render_figures(SCRIPT_DIR / "01_temporal_trends.R") if __name__ == "__main__": main()