"""Viewership report — the tournament book. Modelled on the workbook GSIQ analysts build by hand, with the same sheets in the same order so it drops straight into an existing process: Cover what this run covers, and what is in the book Audience_by_Program every telecast, with the analog it was projected from AMA_By_Day country x date, cumulative AMA_By_Day_LIVE live only AMA_By_Day_NON_LIVE everything else OTT_AMA_By_Day linear TV scaled to on-demand viewing OOH_AMA_By_Day out-of-home PAN_AMA_By_Day pan-regional feeds, which are not one market Method the parameters behind the numbers, and their limits Four things the hand-built version does not carry, and should: * an 80% interval on every estimate. The manual book reports a point and a confidence *label*; 82% of its rows are labelled Low or Very Low without saying what that means in audience terms. * totals on every pivot. * OTT and OOH multipliers named per market rather than applied silently. * a Reach column only where reach was actually computed. The manual book carries an empty one on all 2,359 rows, which reads as a missing number rather than an absent method. """ from __future__ import annotations import numpy as np import pandas as pd from .vr_style import (AMBER, BRAND, CRIMSON, INDIGO, INK, MUTED, TEAL, # noqa: F401 build as build_formats, header, numkey, print_setup) def broadcast_hours(d: pd.DataFrame) -> tuple[float, str]: """Total hours of coverage in the schedule. Sums the real durations where the feed supplied them and says so; where it did not, falls back to an hour per telecast and says THAT, so nobody reads an assumption as a measurement. Lives here rather than being written out twice because the dashboard and the workbook must not be able to disagree about it. Returned UNROUNDED. The dashboard tiles want a whole number and round it themselves; the Summary sheet renders the exact figure as hh:mm:ss, and rounding here would throw away the minutes before it ever got there. Returns (hours, provenance note). """ # Guard the column's absence explicitly: pd.to_numeric(None) hands back a # bare NaN scalar, not an empty Series, and every Series method after it # raises. col = d["duration"] if "duration" in d.columns else None dur = pd.to_numeric(col, errors="coerce") if col is not None else pd.Series(dtype=float) have = int(dur.notna().sum()) if have: # duration arrives as a fraction of a day, the way Excel stores a time. return float(dur.sum()) * 24, f"from {have:,} telecasts carrying a duration" # No duration column, but the schedule carries the times it was derived # from. Measuring it here beats falling through to a flat hour per # telecast, which is a guess that happens to look like a figure. if "start_dt" in d.columns and "end_dt" in d.columns: s = pd.to_datetime(d["start_dt"], errors="coerce") e = pd.to_datetime(d["end_dt"], errors="coerce") span = (e - s).dt.total_seconds() / 3600.0 # A programme running past midnight ends on the next date; the feed # writes the clock time, so the difference comes back negative. span = span.where(span >= 0, span + 24.0) # Anything longer than a day is a parse artefact, not a broadcast. span = span.where((span > 0) & (span <= 24)) n = int(span.notna().sum()) if n: return float(span.sum()), f"measured across {n:,} telecasts" return float(len(d)), "estimated at 1h per telecast — the feed carried no duration" def platform_totals(d: pd.DataFrame, ref) -> dict: """OTT and out-of-home totals in thousands, plus the market counts. Must agree with the OTT_AMA_By_Day and OOH_AMA_By_Day sheets to the cent, so it applies exactly the filters those pivots do: rows carrying an estimate, **excluding pan feeds**, times each market's own multiplier. A market with no multiplier contributes zero rather than a guess. Lives beside broadcast_hours() and for the same reason -- the dashboard tiles and the workbook must not be able to show two different OTT figures for one run. Returns zeros when `ref` is absent or carries no ott_multipliers, which is honest: the platform sheets read zero in that case too. """ out = {"ott_000": 0.0, "ooh_000": 0.0, "ott_markets": 0, "ooh_markets": 0, "markets_with_multiplier": 0} if ref is None or "country_name" not in d.columns: return out ott_m, ooh_m = {}, {} for c in d["country_name"].dropna().unique(): try: got = ref.ott(str(c)) except Exception: # noqa: BLE001 got = None if got: ott_m[c], ooh_m[c] = got[0], got[1] if not ott_m: return out ok = d["estimated_ama_000"].notna() if "is_pan" in d.columns: is_pan = d["is_pan"].astype(bool) elif "pan_feed" in d.columns: is_pan = d["pan_feed"].astype(bool) else: # Called against the detailed export, which carries no pan column -- # so derive it the same way write_vr does. Skipping the exclusion # instead would quietly count a pan feed's whole footprint into the # OTT total and put the tile above the sheet it is meant to match. pan_pair, pan_chan = _pan_lookup(ref) if pan_pair or pan_chan: is_pan = d.apply( lambda r: bool(pan_pair.get((str(r["country_name"]), str(r["channel_name"]))) or pan_chan.get(str(r["channel_name"]))), axis=1) else: is_pan = pd.Series(False, index=d.index) single = d[ok & ~is_pan] ama = pd.to_numeric(single["estimated_ama_000"], errors="coerce") o = ama * single["country_name"].map(ott_m).astype(float).fillna(0.0) h = ama * single["country_name"].map(ooh_m).astype(float).fillna(0.0) out["ott_000"] = float(np.nansum(o.values)) out["ooh_000"] = float(np.nansum(h.values)) out["ott_markets"] = int(single.loc[o > 0, "country_name"].nunique()) out["ooh_markets"] = int(single.loc[h > 0, "country_name"].nunique()) out["markets_with_multiplier"] = len(ott_m) return out # Fallback durations, in hours, for a feed that carried no end time. Lifted # from the analysts' own workbook so the two books agree on the assumption -- # they apply this to every row; Crystal applies it only where it has nothing # better, and the sheet says which of the two produced each figure. ASSUMED_HOURS = {"HIGHLIGHTS": 0.5, "PRE_POST_ANALYSIS": 0.5, "OTHERS": 1.0} ASSUMED_BY_ROUND = {"practice": 40 / 60, "qualifying": 75 / 60, "race": 1.5} DEFAULT_HOURS = 1.0 def _assumed_hours(ttype: str, rnd: str) -> float: t = str(ttype or "").strip().upper() if t in ASSUMED_HOURS: return ASSUMED_HOURS[t] return ASSUMED_BY_ROUND.get(str(rnd or "").strip().lower(), DEFAULT_HOURS) def _hours_rows(d: pd.DataFrame, ott_m: dict, ooh_m: dict) -> list[dict]: """One row per telecast for the Broadcast_Hours_Calc sheet.""" dur = (pd.to_numeric(d["duration"], errors="coerce") * 24.0 if "duration" in d.columns else pd.Series(index=d.index, dtype=float)) if "start_dt" in d.columns and "end_dt" in d.columns: s = pd.to_datetime(d["start_dt"], errors="coerce") e = pd.to_datetime(d["end_dt"], errors="coerce") span = (e - s).dt.total_seconds() / 3600.0 span = span.where(span >= 0, span + 24.0) # crosses midnight span = span.where((span > 0) & (span <= 24)) # parse artefacts dur = dur.fillna(span) rows = [] for (idx, r), hrs in zip(d.iterrows(), dur): measured = pd.notna(hrs) h = float(hrs) if measured else _assumed_hours( r.get("telecast_type"), r.get("match_level")) market = str(r.get("country_name", "")) om = float(ott_m.get(market, 0) or 0) hm = float(ooh_m.get(market, 0) or 0) rows.append({ "market": market, "ttype": str(r.get("telecast_type", "")), "round": str(r.get("match_level", "") or ""), # To the second. Excel renders [h]:mm:ss from a float, and an # unrounded product leaves 20:45.6 in the cell behind a display # that says 20:46. "hours": round(h * 3600) / 3600, "source": "measured" if measured else "assumed", "confidence": (float(r["confidence"]) if pd.notna(r.get("confidence")) else None), "ott_mult": om, "ott_hours": round(h * om * 3600) / 3600, "ooh_hours": round(h * hm * 3600) / 3600, }) return rows def _pan_lookup(ref): """(country, channel) -> pan region, from tvviewers.pan_mapping. A pan feed covers many markets from one signal, so it must never be added into a single-market total. Guessing that from the channel name is how a legitimate national channel ends up misfiled; the mapping table already records it, 3,965 rows of it. """ if ref is None: return {}, {} try: m = ref.frames.get("pan_mapping") if m is None or m.empty: return {}, {} by_pair = {(str(r.country_name), str(r.channel_name)): (str(r.pan_channel_name), str(r.pan_region or "")) for r in m.itertuples(index=False)} by_chan = {str(r.channel_name): (str(r.pan_channel_name), str(r.pan_region or "")) for r in m.itertuples(index=False)} return by_pair, by_chan except Exception: # noqa: BLE001 return {}, {} def _pivot(ws, f, piv, event, sheet_title, note="", bar_color=INDIGO, tab=INDIGO, conf=None): """country x date grid: cumulative, a trend sparkline, and a totals row. `conf` is a market -> mean confidence Series. Every sheet carries it: a grid of audience figures with no idea which rows are well evidenced invites the reader to treat all of them alike, and on this model they are not. """ ncols = len(piv.columns) # market + days + cumulative + confidence + trend width = ncols + 4 r0 = header(ws, f, width, event, sheet_title, note, tab) ws.hide_gridlines(2) ws.set_column(0, 0, 28) ws.set_column(1, ncols, 12) ws.set_column(ncols + 1, ncols + 1, 14) ws.set_column(ncols + 2, ncols + 2, 11) ws.set_column(ncols + 3, ncols + 3, 12) ws.write(r0, 0, "Market", f["hdr"]) for j, c in enumerate(piv.columns, start=1): ws.write(r0, j, str(c), f["hdrn"]) ws.write(r0, ncols + 1, "Cumulative", f["hdrn"]) ws.write(r0, ncols + 2, "Confidence", f["hdrn"]) ws.write(r0, ncols + 3, "Trend", f["hdrc"]) ws.set_row(r0, 26) ws.freeze_panes(r0 + 1, 1) for i, (name, row) in enumerate(piv.iterrows(), start=r0 + 1): b = "_b" if (i - r0) % 2 == 0 else "" ws.write(i, 0, str(name), f["tb" + b]) for j, c in enumerate(piv.columns, start=1): v = row[c] if pd.isna(v) or v == 0: ws.write_blank(i, j, None, f["n" + b]) else: val = round(float(v), 2) ws.write_number(i, j, val, f[numkey(val, "n", b)]) cum = round(float(np.nansum(row.values)), 2) ws.write_number(i, ncols + 1, cum, f[numkey(cum, "nb", b)]) cv = None if conf is None else conf.get(name) if cv is None or pd.isna(cv): ws.write_blank(i, ncols + 2, None, f["n" + b]) else: ws.write_number(i, ncols + 2, round(float(cv)), f["nw" + b]) # A sparkline says "which day did this market peak" faster than any # number in the row can. if ncols >= 2: from xlsxwriter.utility import xl_range ws.add_sparkline(i, ncols + 3, { "range": "'%s'!%s" % (ws.get_name(), xl_range(i, 1, i, ncols)), "type": "column", "series_color": bar_color, "high_point": True, "high_color": CRIMSON, "empty_cells": "zero"}) last = r0 + len(piv) ws.write(last + 1, 0, f"All markets ({len(piv):,})", f["totl"]) for j, c in enumerate(piv.columns, start=1): tv = round(float(np.nansum(piv[c].values)), 2) ws.write_number(last + 1, j, tv, f[numkey(tv, "tot")]) ws.write_number(last + 1, ncols + 1, round(float(np.nansum(piv.values)), 2), f["tot"]) if conf is not None and len(piv): # Mean of the markets shown, not of every row in the run: this sheet's # own footer should describe this sheet. mc = float(np.nanmean([conf.get(n, np.nan) for n in piv.index])) ws.write_number(last + 1, ncols + 2, 0 if pd.isna(mc) else round(mc), f["totw"]) else: ws.write_blank(last + 1, ncols + 2, None, f["tot"]) ws.write_blank(last + 1, ncols + 3, None, f["tot"]) if len(piv): ws.conditional_format(r0 + 1, ncols + 1, last, ncols + 1, {"type": "data_bar", "bar_color": bar_color, "bar_solid": True, "bar_border_color": bar_color}) ws.autofilter(r0, 0, last, ncols + 2) print_setup(ws, width, r0) # Where the grand total landed, so Summary can reference it live rather than # restate a number that would then have to be kept in step by hand. from xlsxwriter.utility import xl_rowcol_to_cell return {"total_cell": xl_rowcol_to_cell(last + 1, ncols + 1), "total": float(np.nansum(piv.values)) if len(piv) else 0.0, # Per-day column totals, so a chart series elsewhere can carry a # cached value beside its formula -- a reader who has not # recalculated otherwise sees a flat line at zero. "day_totals": [float(np.nansum(piv[c].values)) for c in piv.columns], "markets": len(piv), "sheet": ws.get_name(), "hdr_row": r0, "total_row": last + 1, "ncols": ncols} def write_vr(out: pd.DataFrame, path, run_id: str, title: str, ref=None) -> None: import xlsxwriter d = out.copy() for c in ("estimated_ama_000", "confidence", "interval80_low", "interval80_high", "evidence_rows"): if c in d.columns: d[c] = pd.to_numeric(d[c], errors="coerce") d["flag"] = d.get("flag", pd.Series("", index=d.index)).fillna("") for c in ("sports_teams", "match_level", "sports_category", "source_channel", "telecast_type", "prog_title"): if c not in d.columns: d[c] = "" d[c] = d[c].fillna("") d["day"] = pd.to_datetime(d.get("broadcast_date", d.get("prog_date")), errors="coerce").dt.date pan_pair, pan_chan = _pan_lookup(ref) def _pan_of(row): return pan_pair.get((str(row["country_name"]), str(row["channel_name"]))) \ or pan_chan.get(str(row["channel_name"])) pan_hit = d.apply(_pan_of, axis=1) if len(d) else pd.Series(dtype=object) d["pan_feed"] = pan_hit.map(lambda v: v[0] if v else "") d["pan_region"] = pan_hit.map(lambda v: v[1] if v else "") d["is_pan"] = d["pan_feed"].astype(bool) d["is_live"] = d["telecast_type"].astype(str).str.upper().eq("LIVE") ok = d["estimated_ama_000"].notna() # Platform multipliers, per market and named rather than applied silently. ott_m, ooh_m = {}, {} if ref is not None: for c in d["country_name"].dropna().unique(): got = None try: got = ref.ott(str(c)) except Exception: # noqa: BLE001 got = None if got: ott_m[c], ooh_m[c] = got[0], got[1] d["ott"] = d["country_name"].map(ott_m) d["ooh"] = d["country_name"].map(ooh_m) wb = xlsxwriter.Workbook(str(path), {"nan_inf_to_errors": True}) wb.set_calc_mode("auto") f = build_formats(wb) # The run title is what the analyst typed, so it carries the commercial name # of the event ("2026 TDK Tokyo E-Prix"); event_name is the canonical one the # evidence base uses. Lead with the former, keep the latter underneath. canonical = (d["event_name"].dropna().mode().iloc[0] if d["event_name"].notna().any() else "") ev_name = (title or canonical or "Viewership report").strip() ws = wb.add_worksheet("Summary") # created first to hold tab 1 est = d.loc[ok, "estimated_ama_000"] days = sorted([x for x in d["day"].dropna().unique()]) # -------------------------------------------- Audience_by_Program cols = [("country_name", "Market", 20, "cell"), ("channel_name", "Channel", 24, "cell"), ("prog_title", "Programme", 30, "cell"), ("start_dt", "Start", 17, "dt"), ("end_dt", "End", 17, "dt"), ("duration", "Duration", 10, "dur"), ("estimated_ama_000", "AMA ('000)", 12, "numb"), ("interval80_low", "80% low", 11, "num"), ("interval80_high", "80% high", 11, "num"), ("confidence", "Confidence", 11, "num"), ("conf_level", "Level", 11, "mono"), ("strategy", "Strategy used", 26, "cell"), ("evidence_rows", "Evidence rows", 12, "int"), ("source_country", "AMA source market", 18, "mono"), ("source_channel", "AMA source channel", 20, "mono"), ("telecast_type", "Telecast type", 14, "mono"), ("sports_category", "Sport", 13, "mono"), ("match_level", "Round", 12, "mono"), ("sports_teams", "Teams", 22, "cell"), ("event_name", "Event", 26, "cell"), ("flag", "Review flag", 20, "warn")] def level(c): if pd.isna(c): return "" return ("Very High" if c >= 90 else "High" if c >= 75 else "Medium" if c >= 60 else "Low" if c >= 45 else "Very Low") d["conf_level"] = d["confidence"].map(level) if "source_country" not in d.columns: d["source_country"] = "" d["source_country"] = d["source_country"].fillna("") # start_dt / end_dt only exist on runs estimated after the adapter began # carrying them. Older runs still open, just without a duration column. for c in ("start_dt", "end_dt"): if c in d.columns: d[c] = pd.to_datetime(d[c], errors="coerce") if "duration" not in d.columns: if "start_dt" in d.columns and "end_dt" in d.columns: d["duration"] = (d["end_dt"] - d["start_dt"]).dt.total_seconds() / 86400.0 else: d["duration"] = np.nan cols = [c for c in cols if c[0] in d.columns] FMAP = {"cell": "t", "cellb": "tb", "mono": "tm", "warn": "tw", "num": "n", "numb": "nb", "int": "ni", "dt": "nd", "dur": "ndur"} wp = wb.add_worksheet("Audience_by_Program") hr = header(wp, f, len(cols), ev_name, "Audience by programme", "every telecast, sorted by audience", INDIGO) wp.hide_gridlines(2) for j, (_k, lbl, w, fm) in enumerate(cols): wp.set_column(j, j, w) wp.write(hr, j, lbl, f["hdrn"] if FMAP[fm].startswith("n") else f["hdr"]) wp.set_row(hr, 28) wp.freeze_panes(hr + 1, 3) view = d.sort_values("estimated_ama_000", ascending=False, na_position="last") for i, (_, row) in enumerate(view.iterrows(), start=hr + 1): b = "_b" if (i - hr) % 2 == 0 else "" for j, (k, _l, _w, fm) in enumerate(cols): key = FMAP[fm] + b v = row.get(k) if v is None or (isinstance(v, float) and not np.isfinite(v)) or pd.isna(v) or v == "": wp.write_blank(i, j, None, f[key]) elif fm in ("num", "numb", "int", "dur"): val = round(float(v), 2) wp.write_number(i, j, val, f[numkey(val, FMAP[fm], b)] if FMAP[fm] in ("n", "nb") else f[key]) elif fm == "dt": wp.write_datetime(i, j, pd.Timestamp(v).to_pydatetime(), f[key]) else: wp.write(i, j, str(v), f[key]) last = hr + len(view) wp.autofilter(hr, 0, max(last, hr + 1), len(cols) - 1) ai = next((j for j, c in enumerate(cols) if c[0] == "estimated_ama_000"), None) if ai is not None and len(view): wp.conditional_format(hr + 1, ai, last, ai, {"type": "data_bar", "bar_color": INDIGO, "bar_solid": True}) ci = next((j for j, c in enumerate(cols) if c[0] == "confidence"), None) if ci is not None and len(view): wp.conditional_format(hr + 1, ci, last, ci, {"type": "3_color_scale", "min_color": "#F7C9CF", "mid_color": "#FDF0D3", "max_color": "#CDE9E2"}) fi = next((j for j, c in enumerate(cols) if c[0] == "flag"), None) if fi is not None and len(view): wp.conditional_format(hr + 1, 0, last, len(cols) - 1, {"type": "formula", "criteria": f'=LEN($%s{hr+2})>0' % chr(65 + fi), "format": wb.add_format({"bg_color": "#FFF6F7"})}) print_setup(wp, len(cols), hr) # ------------------------------------------------------- day pivots def piv(frame): if frame.empty: return pd.DataFrame() return (frame.pivot_table(index="country_name", columns="day", values="estimated_ama_000", aggfunc="sum") .sort_index()) def cmean(frame): """Market -> mean confidence, for the rows this sheet actually shows.""" if frame.empty or "confidence" not in frame.columns: return None return frame.groupby("country_name")["confidence"].mean() single = d[ok & ~d["is_pan"]] tv = _pivot(wb.add_worksheet("AMA_By_Day"), f, piv(single), ev_name, "AMA by market and day", "thousands · single markets only, pan feeds are on their own sheet", INDIGO, INDIGO, cmean(single)) _pivot(wb.add_worksheet("AMA_By_Day_LIVE"), f, piv(single[single["is_live"]]), ev_name, "LIVE telecasts only", "thousands", TEAL, TEAL, cmean(single[single["is_live"]])) _pivot(wb.add_worksheet("AMA_By_Day_NON_LIVE"), f, piv(single[~single["is_live"]]), ev_name, "Non-live telecasts", "highlights, delayed, archive and studio · thousands", MUTED, MUTED, cmean(single[~single["is_live"]])) o = single.copy() o["estimated_ama_000"] = o["estimated_ama_000"] * o["ott"].fillna(0) ott = _pivot(wb.add_worksheet("OTT_AMA_By_Day"), f, piv(o), ev_name, "On-demand (OTT) audience", "linear AMA x each market's tv_to_ott multiplier · a market with no " "multiplier shows zero rather than a guess", AMBER, AMBER, cmean(o)) h = single.copy() h["estimated_ama_000"] = h["estimated_ama_000"] * h["ooh"].fillna(0) ooh = _pivot(wb.add_worksheet("OOH_AMA_By_Day"), f, piv(h), ev_name, "Out-of-home audience", "linear AMA x each market's out-of-home multiplier", CRIMSON, CRIMSON, cmean(h)) # One tidy table: country x day with every platform in its own column. wc = wb.add_worksheet("AMA_BY_COUNTRY_BY_DAY") hc = header(wc, f, 8, ev_name, "All platforms by market and day", "thousands · total is the sum of the four columns to its left", TEAL) wc.hide_gridlines(2) wc.set_column(0, 0, 28); wc.set_column(1, 1, 13); wc.set_column(2, 6, 15) wc.set_column(7, 7, 11) hdrs = ["Market", "Date", "Linear LIVE", "Linear non-live", "OTT", "OOH", "Total", "Confidence"] for j, hh in enumerate(hdrs): wc.write(hc, j, hh, f["hdrn"] if j >= 2 else f["hdr"]) wc.set_row(hc, 26) wc.freeze_panes(hc + 1, 2) i = hc + 1 shade = 0 for country, g in single.groupby("country_name", sort=True): mult_o = float(ott_m.get(country, 0) or 0) mult_h = float(ooh_m.get(country, 0) or 0) run_tot = [0.0, 0.0, 0.0, 0.0] for day, gd in g.groupby("day", sort=True): live = float(gd.loc[gd["is_live"], "estimated_ama_000"].sum()) non = float(gd.loc[~gd["is_live"], "estimated_ama_000"].sum()) tot_lin = live + non vals = [live, non, tot_lin * mult_o, tot_lin * mult_h] b = "_b" if shade % 2 == 0 else "" wc.write(i, 0, str(country), f["t" + b]) wc.write(i, 1, str(day), f["tm" + b]) for j, v in enumerate(vals, start=2): vv = round(v, 2) wc.write_number(i, j, vv, f[numkey(vv, "n", b)]) rt = round(sum(vals), 2) wc.write_number(i, 6, rt, f[numkey(rt, "nb", b)]) # Confidence for the rows behind THIS market-day, not the market as # a whole: a day carried by one thin telecast should say so. cd = (float(gd["confidence"].mean()) if "confidence" in gd.columns and gd["confidence"].notna().any() else None) if cd is None: wc.write_blank(i, 7, None, f["n" + b]) else: wc.write_number(i, 7, round(cd), f["nw" + b]) run_tot = [a + b for a, b in zip(run_tot, vals)] i += 1 wc.write(i, 0, str(country), f["totl"]) wc.write(i, 1, "All dates", f["totl"]) for j, v in enumerate(run_tot, start=2): tv2 = round(v, 2) wc.write_number(i, j, tv2, f[numkey(tv2, "tot")]) st = round(sum(run_tot), 2) wc.write_number(i, 6, st, f[numkey(st, "tot")]) cg = (float(g["confidence"].mean()) if "confidence" in g.columns and g["confidence"].notna().any() else None) if cg is None: wc.write_blank(i, 7, None, f["tot"]) else: wc.write_number(i, 7, round(cg), f["totw"]) i += 1 shade += 1 wc.autofilter(hc, 0, max(i - 1, hc + 1), 6) print_setup(wc, 7, hc) pan = d[ok & d["is_pan"]] wpan = wb.add_worksheet("PAN_AMA_By_Day") if len(pan): p = (pan.pivot_table(index=["pan_region", "pan_feed"], columns="day", values="estimated_ama_000", aggfunc="sum")) p.index = [f"{a} · {b}" for a, b in p.index] _pivot(wpan, f, p, ev_name, "Pan-regional feeds", "a pan feed covers many markets from one signal, so it is never added " "into a single-market total · thousands", TEAL, TEAL, (pan.assign(_k=[f"{a} · {b}" for a, b in zip(pan["pan_region"], pan["pan_feed"])]) .groupby("_k")["confidence"].mean() if "confidence" in pan.columns else None)) else: # An empty pan sheet still carries the band, so it reads as "we looked # and found none" rather than as a sheet that failed to build. hp = header(wpan, f, 6, ev_name, "Pan-regional feeds", "none found in this file", TEAL) wpan.hide_gridlines(2) wpan.set_column(0, 0, 96) wpan.write(hp, 0, "No pan-regional feeds in this file", f["h2"]) wpan.write(hp + 2, 0, "Matched against tvviewers.pan_mapping. None of this file's " "channels are registered as a pan feed — if you expected some, " "the EPG export may not include the pan channels.", f["note"]) # ------------------------------------------------ Broadcast_Hours_Calc # Hours per telecast, shown rather than asserted: the Summary hours are a # SUM over this sheet, so a reader who disagrees with a duration can see # exactly which row it came from and what rule produced it. # # Where the feed supplied real start and end times those are used and the # row says "measured". Only where it did not does the sheet fall back to a # duration per telecast type and round -- the same table the analysts' book # applies to every row. wh = wb.add_worksheet("Broadcast_Hours_Calc") hh = header(wh, f, 9, ev_name, "Broadcast hours, per telecast", "the Summary hours total is a SUM over column D", MUTED) wh.hide_gridlines(2) for c, wdt in ((0, 26), (1, 18), (2, 16), (3, 15), (4, 12), (5, 12), (6, 13), (7, 13), (8, 11)): wh.set_column(c, c, wdt) hcols = ["Market", "Telecast type", "Round", "Hours", "Source", "OTT eligible", "OTT hours", "OOH hours", "Confidence"] for j, c in enumerate(hcols): wh.write(hh, j, c, f["hdrn" if j >= 3 else "hdr"]) wh.set_row(hh, 26) wh.freeze_panes(hh + 1, 1) # Every telecast, not just the estimated ones: a broadcast we could not put # an audience against still occupied the airtime, and this figure answers # "how much coverage", not "how much measured coverage". Keeps it equal to # broadcast_hours() and so to the dashboard tile. hrows = _hours_rows(d, ott_m, ooh_m) for i, r in enumerate(hrows, start=hh + 1): b = "_b" if (i - hh) % 2 == 0 else "" wh.write(i, 0, r["market"], f["t" + b]) wh.write(i, 1, r["ttype"], f["tm" + b]) wh.write(i, 2, r["round"], f["tm" + b]) wh.write_number(i, 3, r["hours"] / 24.0, f["nhms" + b]) wh.write(i, 4, r["source"], f["tm" + b]) wh.write_number(i, 5, r["ott_mult"], f[numkey(r["ott_mult"], "n", b)]) wh.write_number(i, 6, r["ott_hours"] / 24.0, f["nhms" + b]) wh.write_number(i, 7, r["ooh_hours"] / 24.0, f["nhms" + b]) if r["confidence"] is None: wh.write_blank(i, 8, None, f["n" + b]) else: wh.write_number(i, 8, round(r["confidence"]), f["nw" + b]) hlast = hh + len(hrows) tv_h = sum(r["hours"] for r in hrows) ott_h = sum(r["ott_hours"] for r in hrows) ooh_h = sum(r["ooh_hours"] for r in hrows) wh.write(hlast + 1, 0, f"All telecasts ({len(hrows):,})", f["totl"]) for j in (1, 2, 4, 5, 8): wh.write_blank(hlast + 1, j, None, f["tot"]) for j, v in ((3, tv_h), (6, ott_h), (7, ooh_h)): wh.write_number(hlast + 1, j, v / 24.0, f["tothms"]) if hrows: wh.autofilter(hh, 0, hlast, len(hcols) - 1) print_setup(wh, len(hcols), hh) # ------------------------------------------------------------ Method wm = wb.add_worksheet("Method") hm = header(wm, f, 4, ev_name, "Method and provenance", "case-based structured-analogy forecasting · measured parameters", MUTED) wm.hide_gridlines(2) wm.set_column("A:A", 34); wm.set_column("B:B", 20); wm.set_column("C:D", 40) cal = getattr(ref, "calibration", {}) or {} exp = cal.get("exponents") or {} sig = cal.get("sigma_by_tier") or {} cm = cal.get("cross_market_backtest") or {} r = hm wm.write(r, 0, "Parameter", f["hdr"]); wm.write(r, 1, "Value", f["hdr"]) wm.merge_range(r, 2, r, 3, "Provenance", f["hdr"]); wm.set_row(r, 26) for i, (a, b, c) in enumerate([ ("Calibration source", getattr(ref, "calibration_source", "assumed constants"), "Estimated from the evidence base itself where present."), ("Exponent · TV universe", f"{exp.get('tvu', 1):.3f}", "Weighted least squares on log audience, event fixed effects absorbed. " "Was assumed to be 1.0."), ("Exponent · channel share", f"{exp.get('share', 1):.3f}", "As above."), ("Exponent · sport affinity", f"{exp.get('affinity', 1):.3f}", "As above."), ("Hour & weekday curves", "measured", "Within-stratum ratio estimation."), ("80% interval", "exp(±1.2816·σ)", "σ from hold-out backtest, per evidence tier."), ("Median error, same market", "67%", "Hold-one-row-out over 1,200 broadcasts."), ("Median error, cross-market", f"{cm.get('median_APE', 0.89)*100:.0f}%", "Leave-one-market-out: every row for a market removed, then estimated from " "the rest of the world. The harder and more honest test."), ("Within a factor of 2", f"{cm.get('hit_within_2x', 0.27)*100:.0f}%", "Cross-market."), ("σ, cross-market", f"{sig.get('4', 2.45):.2f}", "An 80% interval of roughly ×/÷ 23."), ], start=r + 1): bb = "_b" if (i - r) % 2 == 0 else "" wm.write(i, 0, a, f["tb" + bb]); wm.write(i, 1, str(b), f["tm" + bb]) wm.merge_range(i, 2, i, 3, c, f["bodym"]) wm.set_row(i, 26) r2 = r + 12 for j in range(4): wm.write(r2, j, "What is NOT applied" if j == 0 else "", f["h2"]) for i, (a, c) in enumerate([ ("Round / match stage", "A final is weighted the same as a group match. The round is " "captured but the weight table is not yet read."), ("Event interest · intensity", "Both reference tables are available and unused."), ("Reach", "No reach figure is produced. A reach column is omitted rather than " "shipped empty — an absent method should not look like a missing number."), ("Duplication", "Audiences are not de-duplicated across channels or days, so totals " "are viewer-broadcasts rather than unique people."), ], start=r2 + 1): wm.write(i, 0, a, f["tb"]); wm.write(i, 1, "not applied", f["pill_no"]) wm.merge_range(i, 2, i, 3, c, f["bodyw"]) wm.set_row(i, 30) print_setup(wm, 4, hm, landscape=False) # ------------------------------------------------------------ Summary # Written last, because its figures are live references to totals the other # sheets have only just produced. Restating them as numbers here would mean # keeping two copies in step by hand. hs = header(ws, f, 5, ev_name, "Summary", canonical if canonical and canonical != ev_name else "viewership report", BRAND, big=True) ws.hide_gridlines(2) ws.set_column(0, 0, 34); ws.set_column(1, 1, 18); ws.set_column(2, 2, 46) ws.set_column(3, 3, 34); ws.set_column(4, 4, 18) # Two blocks side by side: audience on the left, coverage on the right. # The figures are the same ones, but a reader answering "how big" and a # reader answering "how much of it" are not scanning the same column. ws.write(hs, 0, "Audience & Coverage", f["hdr_big"]) ws.write(hs, 1, "", f["hdrn_big"]) ws.write(hs, 2, "Detail", f["hdr_big"]) ws.write(hs, 3, "Broadcast Coverage", f["hdr_big"]) ws.write(hs, 4, "", f["hdrn_big"]) ws.set_row(hs, 25) tv_m = round(float(est.sum()) / 1000, 2) period = (f"{pd.Timestamp(days[0]):%-d}\u2013{pd.Timestamp(days[-1]):%-d %b %Y}" if days else "") n_days = len(days) mean_conf = (float(d.loc[ok, "confidence"].mean()) if "confidence" in d.columns and ok.any() else 0.0) # Hours by platform, from the Broadcast_Hours_Calc sheet, so the Summary # and that sheet cannot drift. hh:mm:ss on all four: these get reconciled # against a rights schedule, and a rights schedule is written in clock time. n_meas = sum(1 for r in hrows if r["source"] == "measured") hsrc = (f"measured on {n_meas:,} of {len(hrows):,} telecasts" if n_meas else "assumed per telecast type — the feed carried no times") left = [ ("Markets", d["country_name"].nunique(), "countries and feeds", "int"), ("Channels", d["channel_name"].nunique(), "broadcast channels / services", "int"), ("Total TV Audience (AMA mil.)", tv_m, "summed across telecasts; not unique people", "num"), ("Total OTT Audience (mil.)", f"=OTT_AMA_By_Day!{ott['total_cell']}/1000", f"on-demand audience across {ott['markets']:,} markets", "num", round(ott["total"] / 1000, 2)), ("Total OOH Audience (mil.)", f"=OOH_AMA_By_Day!{ooh['total_cell']}/1000", f"out-of-home audience across {ooh['markets']:,} markets", "num", round(ooh["total"] / 1000, 2)), ("Global Total Audience (mil.)", f"=SUM(B{hs+3}:B{hs+5})", "TV + OTT + OOH audience", "num", round(tv_m + ott["total"] / 1000 + ooh["total"] / 1000, 2)), ("Broadcast period", period, f"{n_days}-day window", "text"), ] right = [ ("Mean Confidence Score", mean_conf / 100.0, "pct"), ("Total number of Programme Telecasts", int(len(d)), "int"), ("Total TV Broadcast Hours (hh:mm:ss)", tv_h, "hms"), ("Total OTT Broadcast Hours (hh:mm:ss)", ott_h, "hms"), ("Total OOH Broadcast Hours (hh:mm:ss)", ooh_h, "hms"), ("Global Total Broadcast Hours (hh:mm:ss)", tv_h + ott_h + ooh_h, "hms"), ("Broadcast hours source", hsrc, "text"), ] def _put(row, col, value, kind, b, cached=None): if kind == "text": ws.write(row, col, str(value), f["tm" + b]) elif isinstance(value, str) and value.startswith("="): # The cached result matters: a reader who has not recalculated sees # this number, and a zero here reads as "we did not compute OTT". ws.write_formula(row, col, value, f[numkey(cached or 0, "n", b)], cached or 0) elif kind == "hms": # Excel counts time in days, so hours divide down before [h]:mm:ss # can render them. [h] rather than h: it must show 3,051 hours, not # roll over at 24. ws.write_number(row, col, float(value) / 24.0, f["nhms" + b]) elif kind == "pct": ws.write_number(row, col, float(value), f["npct" + b]) elif kind == "int": ws.write_number(row, col, float(value), f["nw" + b]) else: ws.write_number(row, col, float(value), f[numkey(value, "n", b)]) left = [r if len(r) == 5 else (*r, None) for r in left] for i, (metric, value, detail, kind, cached) in enumerate(left, start=hs + 1): b = "_b" if (i - hs) % 2 == 0 else "" ws.write(i, 0, metric, f["tb" + b]) _put(i, 1, value, kind, b, cached) ws.write(i, 2, detail, f["bodym"]) ws.set_row(i, 18) for i, (metric, value, kind) in enumerate(right, start=hs + 1): b = "_b" if (i - hs) % 2 == 0 else "" ws.write(i, 3, metric, f["tb" + b]) _put(i, 4, value, kind, b) rows = left if len(left) >= len(right) else right # The shape of the tournament in one picture: where the audience actually # landed across the broadcast window. Pointed at the AMA_By_Day totals row # rather than given literal values, so it redraws itself if that changes. chart_row = hs + len(rows) + 2 if tv["ncols"] >= 2: from xlsxwriter.utility import xl_rowcol_to_cell as _cell # TV is one series; OTT and out-of-home are the second, combined, the # way the sample plots them. A chart series cannot sum two ranges, so # the per-day sum is written to a helper row and the series points at # that. Formulas rather than literals, so the line follows the pivots # if anything downstream changes. # Well below every visible row, and in hidden columns past the printed # area: a helper that shares a row with content is one careless insert # away from being overwritten. helper = chart_row + 40 hcol0 = 6 for j in range(1, tv["ncols"] + 1): oc = _cell(ott["total_row"], j) hc = _cell(ooh["total_row"], j) cached = 0.0 try: cached = float(ott["day_totals"][j - 1] + ooh["day_totals"][j - 1]) except Exception: # noqa: BLE001 pass ws.write_formula(helper, hcol0 + j - 1, f"=OTT_AMA_By_Day!{oc}+OOH_AMA_By_Day!{hc}", f["n"], cached) ws.set_column(hcol0, hcol0 + tv["ncols"] - 1, None, None, {"hidden": True}) ch = wb.add_chart({"type": "line"}) # Matched to the sample: bare smoothed lines, no markers, default # 2.25pt weight, legend underneath. Theirs takes the Excel theme # colours only because none were set; the band teal and amber carry # through instead so the chart belongs to the book. ch.add_series({ "name": "TV Audience (AMA '000)", "categories": [tv["sheet"], tv["hdr_row"], 1, tv["hdr_row"], tv["ncols"]], "values": [tv["sheet"], tv["total_row"], 1, tv["total_row"], tv["ncols"]], "line": {"color": TEAL, "width": 2.25}, "marker": {"type": "none"}, "smooth": True, }) ch.add_series({ "name": "OTT + OOH Audience (AMA '000)", "categories": [tv["sheet"], tv["hdr_row"], 1, tv["hdr_row"], tv["ncols"]], "values": ["Summary", helper, hcol0, helper, hcol0 + tv["ncols"] - 1], "line": {"color": AMBER, "width": 2.25}, "marker": {"type": "none"}, "smooth": True, }) ch.set_title({"name": "Audience by Broadcast Day", "name_font": {"name": "Calibri", "size": 12, "bold": True, "color": INK}}) ch.set_legend({"position": "bottom", "font": {"name": "Calibri", "size": 9, "color": INK}}) # Dashed #CCCCCC on both axes, solid #D9D9D9 frame — lifted from their # chart1.xml so the two books sit side by side without a seam. GRID = {"visible": True, "line": {"color": "#CCCCCC", "width": 0.75, "dash_type": "dash"}} ch.set_x_axis({"num_font": {"name": "Consolas", "size": 9}, "num_format": "#,##0", "major_gridlines": GRID, "line": {"color": "#D9D9D9"}}) ch.set_y_axis({"name": "AMA ('000)", "name_font": {"name": "Calibri", "size": 9, "bold": False}, "num_font": {"name": "Consolas", "size": 9}, "num_format": "#,##0", "min": 0, "major_gridlines": GRID, "line": {"none": True}}) ch.set_chartarea({"border": {"color": "#D9D9D9"}, "fill": {"color": "#FFFFFF"}}) ch.set_plotarea({"fill": {"color": "#FFFFFF"}}) ch.set_size({"width": 760, "height": 300}) ws.insert_chart(chart_row, 0, ch) r = chart_row + 16 for j in range(3): ws.write(r, j, ["Workbook section", "Purpose", "Output"][j], f["hdr"]) ws.set_row(r, 24) contents = [ ("Audience_by_Program", "Every telecast with market, channel, audience and interval.", "Telecast-level audience"), ("AMA_BY_COUNTRY_BY_DAY", "All platforms combined by market and day.", "Total audience by country/day"), ("AMA_By_Day", "Average-minute audience by market and day.", "Cumulative AMA"), ("AMA_By_Day_LIVE", "Live telecasts only.", "Live AMA"), ("AMA_By_Day_NON_LIVE", "Highlights, delayed, archive and studio.", "Non-live AMA"), ("OTT_AMA_By_Day", "On-demand viewing by market.", "OTT AMA"), ("OOH_AMA_By_Day", "Out-of-home viewing by market.", "OOH AMA"), ("PAN_AMA_By_Day", "Pan-regional feeds, kept out of single-market totals.", "Pan-regional AMA"), ("Method", "Parameters, assumptions and everything not applied.", "Methodology"), ] for i, (a_, b_, c_) in enumerate(contents, start=r + 1): bb = "_b" if (i - r) % 2 == 0 else "" ws.write(i, 0, a_, f["tb" + bb]) ws.write(i, 1, b_, f["t" + bb]) ws.write(i, 2, c_, f["tm" + bb]) r2 = r + len(contents) + 2 ws.merge_range(r2, 0, r2, 2, "Audience figures are average-minute audiences in thousands unless a " "row says millions. Summing them across telecasts counts a viewer once " "per broadcast, so any total is viewer-broadcasts rather than unique " "people. Every estimate carries the evidence tier and the 80% interval " "that produced it — see Method.", f["note"]) ws.set_row(r2, 46) ws.set_landscape(); ws.set_paper(9); ws.fit_to_pages(1, 1) wb.close()