<@ULVA73B9P> How, can I download or create a visua...
# ask-marvin
r
@Marvin How, can I download or create a visual that will display the times of the day when the flows execute?
m
thought for 262.3 seconds
Great question. If you want a “what times of day do my flows actually run?” view, there isn’t a built-in download in the UI for that specific chart, but it’s easy to generate one from the API and either save it locally or publish it as an artifact. Below are two practical approaches in Prefect 3.x: Option A — Quick local CSV + chart (last 30 days) - Exports flow runs to CSV (for Excel/BI tools) - Produces a histogram of starts by hour-of-day and a simple heatmap (day-of-week vs hour)
Copy code
import asyncio
import datetime as dt
from zoneinfo import ZoneInfo

import pandas as pd
import matplotlib.pyplot as plt

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime
from prefect.client.schemas.sorting import FlowRunSort

# Settings
DAYS_BACK = 30
TZ = ZoneInfo("America/New_York")  # change to your local TZ, e.g., ZoneInfo("UTC")

async def fetch_runs():
    since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=DAYS_BACK)
    fr_filter = FlowRunFilter(
        start_time=FlowRunFilterStartTime(after_=since)
    )

    runs = []
    limit = 200
    offset = 0
    async with get_client() as client:
        while True:
            batch = await client.read_flow_runs(
                flow_run_filter=fr_filter,
                sort=FlowRunSort.START_TIME_ASC,
                limit=limit,
                offset=offset,
            )
            if not batch:
                break
            runs.extend(batch)
            offset += limit
    return runs

def main():
    runs = asyncio.run(fetch_runs())
    if not runs:
        print("No runs found.")
        return

    # Build a DataFrame
    rows = []
    for r in runs:
        if r.start_time:
            rows.append({
                "flow_run_id": str(r.id),
                "name": r.name,
                "deployment_id": str(r.deployment_id) if r.deployment_id else None,
                "state_type": r.state.type.value if r.state else None,
                "start_time_utc": pd.to_datetime(r.start_time),
                "end_time_utc": pd.to_datetime(r.end_time) if r.end_time else None,
            })
    df = pd.DataFrame(rows)
    if df.empty:
        print("No runs with start_time.")
        return

    # Convert to your local time zone
    df["start_time_local"] = df["start_time_utc"].dt.tz_convert(TZ)
    df["hour_of_day"] = df["start_time_local"].dt.hour
    df["dow"] = df["start_time_local"].dt.dayofweek  # 0=Mon, 6=Sun
    df["date"] = df["start_time_local"].dt.date

    # Save CSV
    df.to_csv("prefect_flow_runs_last_30d.csv", index=False)
    print("Wrote prefect_flow_runs_last_30d.csv")

    # Histogram of starts by hour-of-day
    ax = df["hour_of_day"].value_counts().sort_index().plot(kind="bar", figsize=(10,4))
    ax.set_title("Flow starts by hour of day")
    ax.set_xlabel("Hour of day")
    ax.set_ylabel("Run count")
    plt.tight_layout()
    plt.savefig("prefect_run_hours.png", dpi=150)
    plt.close()
    print("Wrote prefect_run_hours.png")

    # Heatmap: day-of-week vs hour-of-day
    pivot = df.pivot_table(index="dow", columns="hour_of_day", values="flow_run_id", aggfunc="count", fill_value=0)
    plt.figure(figsize=(12,4))
    plt.imshow(pivot.values, aspect="auto", cmap="Blues")
    plt.colorbar(label="Run count")
    plt.yticks(range(7), ["Mon","Tue","Wed","Thu","Fri","Sat","Sun"])
    plt.xticks(range(24), range(24))
    plt.title("Flow starts heatmap (DOW x Hour)")
    plt.xlabel("Hour of day")
    plt.ylabel("Day of week")
    plt.tight_layout()
    plt.savefig("prefect_run_heatmap.png", dpi=150)
    plt.close()
    print("Wrote prefect_run_heatmap.png")

if __name__ == "__main__":
    main()
Notes: - Adjust TZ to your local timezone. - Add filters for specific deployments/tags if needed by extending the
FlowRunFilter
(e.g., tags, deployment_id). - This will create CSV + two PNGs you can download and share.
Option B — Use the flow_runs/history endpoint for large ranges If you have a lot of runs and want aggregated counts by fixed intervals (e.g., hourly bins), use the history endpoint and then chart it.
Copy code
import asyncio
import datetime as dt
from zoneinfo import ZoneInfo

import pandas as pd

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime

TZ = ZoneInfo("America/New_York")
DAYS_BACK = 30
INTERVAL_SECONDS = 3600  # 1 hour

async def fetch_hourly_history():
    history_start = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=DAYS_BACK)
    history_end = dt.datetime.now(dt.timezone.utc)

    # Optional: additional filters (tags, deployments, etc.)
    fr_filter = FlowRunFilter(
        start_time=FlowRunFilterStartTime(after_=history_start, before_=history_end)
    )

    async with get_client() as client:
        resp = await <http://client._client.post|client._client.post>(
            "/flow_runs/history",
            json={
                "history_start": history_start.isoformat(),
                "history_end": history_end.isoformat(),
                "history_interval_seconds": INTERVAL_SECONDS,
                "flow_runs": fr_filter.model_dump(mode="json", exclude_unset=True),
            },
        )
        data = resp.json()

    # Build a DataFrame of total starts per interval (sum across states)
    rows = []
    for bucket in data:
        start = pd.to_datetime(bucket["interval_start"], utc=True).tz_convert(TZ)
        total = sum(s.get("count_runs", 0) for s in bucket.get("states", []))
        rows.append({"interval_start": start, "total_runs": total})
    return pd.DataFrame(rows)

def main():
    df = asyncio.run(fetch_hourly_history())
    if df.empty:
        print("No history data returned.")
        return

    # Derive hour-of-day distribution
    df["hour_of_day"] = df["interval_start"].dt.hour
    hourly = df.groupby("hour_of_day")["total_runs"].sum()

    ax = hourly.plot(kind="bar", figsize=(10,4), title="Starts per hour (aggregated via history)")
    ax.set_xlabel("Hour of day")
    ax.set_ylabel("Run count")
    ax.figure.tight_layout()
    ax.figure.savefig("prefect_run_hours_from_history.png", dpi=150)
    print("Wrote prefect_run_hours_from_history.png")

if __name__ == "__main__":
    main()
Optional — Show the visual inside Prefect - If you can host the image (e.g., on S3) at a URL, you can attach it to the run as an image artifact so it appears in the UI: - Upload the PNG to a public or signed URL (e.g., S3 presigned URL) - Then create an image artifact that points to that URL:
Copy code
from prefect.artifacts import create_image_artifact

create_image_artifact(
    image_url="<https://your-bucket.s3.amazonaws.com/prefect_run_hours.png>",
    key="run-hours-visual",
    description="Starts by hour-of-day for the last 30 days"
)
Would you prefer: - all flows vs a specific deployment/tag? - a simple histogram, or a day-of-week x hour heatmap? - and what timezone should we use? If you share that, I can tailor the snippet exactly to your case.