Spaces:
Running
Running
| """Descarga de "jugadas" (secuencias) en CSV para los gráficos de Ejes y Recursos. | |
| Lee ``vendor/data/plays/<slug>/<season>/plays.parquet`` (1 fila por secuencia con partido, | |
| local/visitante, resultado, tiempos, pases, tiro, gol, xG/pv acumulados, tipo de jugada, | |
| bloque y flags de recurso). Filtra por equipo (y por recurso/bloque en Recursos), agrega la | |
| columna ``recursos`` (qué recursos disparó la jugada = por qué está en el gráfico) y ordena | |
| según el criterio pedido. | |
| """ | |
| from __future__ import annotations | |
| import io | |
| from functools import lru_cache | |
| import pandas as pd | |
| from racing_reports import vendor_env | |
| # columnas visibles del CSV (orden y nombres para el usuario) | |
| CSV_COLS = ["matchId", "local", "visitante", "equipo", "rival", "fecha", "resultado_final", | |
| "min_inicio", "seg_inicio", "min_fin", "seg_fin", "n_pases", "tiro", "gol", | |
| "xg_acumulado", "pv_acumulado", "tipo_jugada", "recursos"] | |
| RENAME = {"matchId": "partido"} | |
| # nombre lindo de cada flag rec_* (para la columna "recursos" = por qué la jugada está ahí) | |
| REC_LABEL = { | |
| "rec_pase_entre_lineas": "Pase entre líneas", | |
| "rec_cutback": "Cutback", | |
| "rec_centro_inswinger": "Centro inswinger", | |
| "rec_centro_outswinger": "Centro outswinger", | |
| "rec_centro_raso": "Centro raso", | |
| "rec_centro_colgado": "Centro colgado", | |
| "rec_rompe_lineas_ok": "Rompe líneas", | |
| "rec_gambeta": "Gambeta", | |
| "rec_tiro_lejano": "Tiro lejano", | |
| } | |
| # criterios de orden del CSV: clave -> (columnas, ascendente) | |
| _ORDEN = { | |
| "todos": (["fecha", "matchId", "min_inicio"], True), | |
| "xg": (["xg_acumulado", "matchId", "min_inicio"], [False, True, True]), | |
| "pv": (["pv_acumulado", "matchId", "min_inicio"], [False, True, True]), | |
| "gol": (["gol", "xg_acumulado", "matchId", "min_inicio"], [False, False, True, True]), | |
| "tiro": (["tiro", "xg_acumulado", "matchId", "min_inicio"], [False, False, True, True]), | |
| } | |
| def _slug(text: str) -> str: | |
| return "".join(c.lower() if c.isalnum() else "-" for c in text).strip("-") | |
| def _df(league: str, season: str) -> pd.DataFrame: | |
| p = vendor_env.DATA_DIR / "plays" / _slug(league) / str(season) / "plays.parquet" | |
| if not p.exists(): | |
| raise FileNotFoundError(f"Sin jugadas para {league} {season}") | |
| return pd.read_parquet(p) | |
| def available(league: str, season: str) -> bool: | |
| p = vendor_env.DATA_DIR / "plays" / _slug(league) / str(season) / "plays.parquet" | |
| return p.exists() | |
| def _recursos_col(df: pd.DataFrame) -> pd.Series: | |
| """String con los recursos que disparó cada jugada (por qué está en el gráfico).""" | |
| present = [(c, lbl) for c, lbl in REC_LABEL.items() if c in df.columns] | |
| def _join(row) -> str: | |
| return " · ".join(lbl for c, lbl in present if row[c] == 1) | |
| return df.apply(_join, axis=1) if present else pd.Series("", index=df.index) | |
| def _resolve_orden(orden: str, df: pd.DataFrame): | |
| """Devuelve (columnas, ascendente, eje_col). eje_col != None cuando se ordena por un | |
| eje del gráfico 4 (ej orden='verticalidad' -> columna 'eje_verticalidad').""" | |
| if orden in _ORDEN: | |
| cols, asc = _ORDEN[orden] | |
| return [c for c in cols if c in df.columns], asc, None | |
| eje_col = f"eje_{orden}" | |
| if eje_col in df.columns: | |
| cols = [c for c in (eje_col, "matchId", "min_inicio") if c in df.columns] | |
| return cols, [False] + [True] * (len(cols) - 1), eje_col | |
| cols, asc = _ORDEN["todos"] | |
| return [c for c in cols if c in df.columns], asc, None | |
| def _to_csv(df: pd.DataFrame, orden: str = "todos") -> str: | |
| df = df.copy() | |
| df["recursos"] = _recursos_col(df) | |
| cols, asc, eje_col = _resolve_orden(orden, df) | |
| if cols: | |
| df = df.sort_values(cols, ascending=asc, na_position="last") | |
| out_cols = [c for c in CSV_COLS if c in df.columns] | |
| rename = dict(RENAME) | |
| if eje_col and eje_col in df.columns: # muestro el valor del eje por el que se ordenó | |
| df[eje_col] = pd.to_numeric(df[eje_col], errors="coerce").round(3) | |
| out_cols.append(eje_col) | |
| rename[eje_col] = eje_col.replace("eje_", "") | |
| out = df[out_cols].rename(columns=rename) | |
| buf = io.StringIO() | |
| # formato europeo (locale español): separador ';' + coma decimal, para que Excel/Numbers | |
| # no lea "0.113" como 113. BOM para que respete UTF-8 (acentos, · y →). | |
| out.to_csv(buf, index=False, sep=";", decimal=",") | |
| return "" + buf.getvalue() | |
| def ejes_csv(league: str, season: str, team: str, orden: str = "todos") -> str: | |
| """Todas las secuencias de ataque del equipo (para el gráfico de Ejes).""" | |
| df = _df(league, season) | |
| df = df[df["equipo"].astype(str) == team] | |
| return _to_csv(df, orden) | |
| def recursos_csv(league: str, season: str, team: str, lado: str = "of", | |
| recurso: str = "todos", bloques: str | None = None, | |
| orden: str = "todos") -> str: | |
| """Secuencias del recurso/bloque elegidos. lado 'of' = ataques del equipo; | |
| 'def' = lo que concede (ataques del rival contra el equipo).""" | |
| df = _df(league, season) | |
| key = "equipo" if lado != "def" else "rival" | |
| df = df[df[key].astype(str) == team] | |
| sel = [b.strip() for b in (bloques or "").split(",") if b.strip() in ("bajo", "medio", "alto")] | |
| if sel: | |
| df = df[df["bloque"].isin(sel)] | |
| col = f"rec_{recurso}" | |
| if col in df.columns: | |
| df = df[df[col] == 1] | |
| return _to_csv(df, orden) | |