Spaces:

fliarbi
/

web-technology-streamlit

Running

web-technology-streamlit / common /data.py

Streamlit app running in the browser with stlite

a1a01f7 verified 2 days ago

3.58 kB

"""Data loading: published HF parquet samples and the small aggregate bundles committed under data/."""
from __future__ import annotations

import io
import json
import os
import sys
import urllib.request
from pathlib import Path

import pandas as pd
import streamlit as st

ROOT = Path(__file__).resolve().parents[1]
META = json.loads((ROOT / "common" / "meta.json").read_text(encoding="utf-8"))
# For offline tests: a directory holding <key>.parquet copies of the published files.
LOCAL_DIR = os.environ.get("FLIAR_PARQUET_DIR")
# True when running in the browser (stlite / Pyodide). There is no socket networking there, so files are
# downloaded with a synchronous XMLHttpRequest (allowed inside the Web Worker stlite runs Python in), and parquet is
# read with fastparquet, which stlite already loads (pyarrow would be a large extra download).
IS_PYODIDE = sys.platform == "emscripten" or "pyodide" in sys.modules
# Optional JSON map {dataset key: URL} that overrides the published parquet URL, e.g. a column-trimmed copy
# served next to a static page. Ignored when unset.
URL_OVERRIDES = json.loads(os.environ.get("FLIAR_PARQUET_URLS") or "{}")


def meta(key: str) -> dict:
    return META[key]


def _browser_get(url: str) -> bytes:
    from js import XMLHttpRequest  # type: ignore  # Pyodide only

    xhr = XMLHttpRequest.new()
    xhr.open("GET", url, False)
    xhr.responseType = "arraybuffer"
    xhr.send(None)
    if xhr.status != 200:
        raise RuntimeError(f"Download failed ({xhr.status}): {url}")
    return xhr.response.to_bytes()


def _fetch(key: str) -> bytes | str:
    if LOCAL_DIR and (Path(LOCAL_DIR) / f"{key}.parquet").exists():
        return str(Path(LOCAL_DIR) / f"{key}.parquet")
    url = URL_OVERRIDES.get(key) or META[key]["parquet"]
    if IS_PYODIDE:
        return _browser_get(url)
    req = urllib.request.Request(url, headers={"User-Agent": "fliar-streamlit-apps/1.0"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return r.read()


def _shrink(df: pd.DataFrame) -> pd.DataFrame:
    """Low-cardinality text columns become categoricals to keep memory small."""
    for c in df.columns:
        if df[c].dtype == object or str(df[c].dtype).startswith("string"):
            if df[c].nunique(dropna=True) < max(50, len(df) // 4):
                df[c] = df[c].astype("category")
            else:
                df[c] = df[c].astype(object)
    return df


@st.cache_data(show_spinner="Loading the published sample from Hugging Face ...", ttl=24 * 3600, max_entries=8)
def load(key: str, columns: tuple[str, ...]) -> pd.DataFrame:
    """Read only the needed columns of a published dataset sample."""
    src = _fetch(key)
    f = io.BytesIO(src) if isinstance(src, bytes) else src
    if IS_PYODIDE:
        import fastparquet

        pf = fastparquet.ParquetFile(f)
        df = pf.to_pandas(columns=[c for c in columns if c in pf.columns])
    else:
        import pyarrow.parquet as pq

        pf = pq.ParquetFile(f)
        df = pf.read(columns=[c for c in columns if c in pf.schema_arrow.names]).to_pandas()
    del src
    return _shrink(df)


@st.cache_data(show_spinner=False, max_entries=32)
def load_bundle(theme: str, name: str) -> pd.DataFrame:
    """Read an aggregate parquet committed under data/<theme>/<name>.parquet."""
    return pd.read_parquet(ROOT / "data" / theme / f"{name}.parquet", engine="fastparquet" if IS_PYODIDE else "auto")


def bundle_names(theme: str) -> list[str]:
    d = ROOT / "data" / theme
    return sorted(p.stem for p in d.glob("*.parquet")) if d.exists() else []

	"""Data loading: published HF parquet samples and the small aggregate bundles committed under data/."""
	from __future__ import annotations

	import io
	import json
	import os
	import sys
	import urllib.request
	from pathlib import Path

	import pandas as pd
	import streamlit as st

	ROOT = Path(__file__).resolve().parents[1]
	META = json.loads((ROOT / "common" / "meta.json").read_text(encoding="utf-8"))
	# For offline tests: a directory holding <key>.parquet copies of the published files.
	LOCAL_DIR = os.environ.get("FLIAR_PARQUET_DIR")
	# True when running in the browser (stlite / Pyodide). There is no socket networking there, so files are
	# downloaded with a synchronous XMLHttpRequest (allowed inside the Web Worker stlite runs Python in), and parquet is
	# read with fastparquet, which stlite already loads (pyarrow would be a large extra download).
	IS_PYODIDE = sys.platform == "emscripten" or "pyodide" in sys.modules
	# Optional JSON map {dataset key: URL} that overrides the published parquet URL, e.g. a column-trimmed copy
	# served next to a static page. Ignored when unset.
	URL_OVERRIDES = json.loads(os.environ.get("FLIAR_PARQUET_URLS") or "{}")


	def meta(key: str) -> dict:
	return META[key]


	def _browser_get(url: str) -> bytes:
	from js import XMLHttpRequest # type: ignore # Pyodide only

	xhr = XMLHttpRequest.new()
	xhr.open("GET", url, False)
	xhr.responseType = "arraybuffer"
	xhr.send(None)
	if xhr.status != 200:
	raise RuntimeError(f"Download failed ({xhr.status}): {url}")
	return xhr.response.to_bytes()


	def _fetch(key: str) -> bytes \| str:
	if LOCAL_DIR and (Path(LOCAL_DIR) / f"{key}.parquet").exists():
	return str(Path(LOCAL_DIR) / f"{key}.parquet")
	url = URL_OVERRIDES.get(key) or META[key]["parquet"]
	if IS_PYODIDE:
	return _browser_get(url)
	req = urllib.request.Request(url, headers={"User-Agent": "fliar-streamlit-apps/1.0"})
	with urllib.request.urlopen(req, timeout=120) as r:
	return r.read()


	def _shrink(df: pd.DataFrame) -> pd.DataFrame:
	"""Low-cardinality text columns become categoricals to keep memory small."""
	for c in df.columns:
	if df[c].dtype == object or str(df[c].dtype).startswith("string"):
	if df[c].nunique(dropna=True) < max(50, len(df) // 4):
	df[c] = df[c].astype("category")
	else:
	df[c] = df[c].astype(object)
	return df


	@st.cache_data(show_spinner="Loading the published sample from Hugging Face ...", ttl=24 * 3600, max_entries=8)
	def load(key: str, columns: tuple[str, ...]) -> pd.DataFrame:
	"""Read only the needed columns of a published dataset sample."""
	src = _fetch(key)
	f = io.BytesIO(src) if isinstance(src, bytes) else src
	if IS_PYODIDE:
	import fastparquet

	pf = fastparquet.ParquetFile(f)
	df = pf.to_pandas(columns=[c for c in columns if c in pf.columns])
	else:
	import pyarrow.parquet as pq

	pf = pq.ParquetFile(f)
	df = pf.read(columns=[c for c in columns if c in pf.schema_arrow.names]).to_pandas()
	del src
	return _shrink(df)


	@st.cache_data(show_spinner=False, max_entries=32)
	def load_bundle(theme: str, name: str) -> pd.DataFrame:
	"""Read an aggregate parquet committed under data/<theme>/<name>.parquet."""
	return pd.read_parquet(ROOT / "data" / theme / f"{name}.parquet", engine="fastparquet" if IS_PYODIDE else "auto")


	def bundle_names(theme: str) -> list[str]:
	d = ROOT / "data" / theme
	return sorted(p.stem for p in d.glob("*.parquet")) if d.exists() else []