app.py

import ibis
from ibis import _

import streamlit as st
import altair as alt
import os
import pandas as pd
import matplotlib.pyplot as plt
from pandasai.llm.openai import OpenAI
from pandasai import Agent
from pandasai.responses.streamlit_response import StreamlitResponse
import leafmap.maplibregl as leafmap

st.set_page_config(layout="wide",
                   page_title="TPL LandVote",
                   page_icon=":globe:")

'''
# LandVote Prototype

An experimental platform for visualizing data on ballot measures for conservation, based on data from <https://landvote.org/> curated by the Trust for Public Land.

'''

COLORS = {
    "dark_orange": "#ab5601",
    "light_orange": "#f3d3b1",
    "grey": "#d3d3d3",
    "light_green": "#c3dbc3",
    "dark_green": "#417d41",
    "dem_blue": "#1b46c2",
    "rep_red": "#E81B23"
}


## chatbot
llm = OpenAI(api_token=st.secrets["OPENAI_API_KEY"])
df1 = pd.read_csv("data.csv")
agent = Agent([df1], config={"verbose": True, "response_parser": StreamlitResponse, "llm": llm})

year = st.slider("Select a year", 1988, 2024, 2022, 1)

 

votes_pmtiles = "https://huggingface.co/datasets/boettiger-lab/landvote/resolve/main/votes.pmtiles"
votes_parquet = "https://huggingface.co/datasets/boettiger-lab/landvote/resolve/main/votes.parquet"

# get parquet data for charts
con = ibis.duckdb.connect(extensions=["spatial"])
votes = (con
         .read_parquet(votes_parquet)
         .cast({"geometry": "geometry"})
        )


# generate altair chart with df
def create_chart(df, y_column, ylab, title, color):
    # color encoding - color is a list or single value 
    color_encoding = (
        alt.Color('party:N', scale=alt.Scale(domain=["DEMOCRAT", "REPUBLICAN"], range=color))
        if isinstance(color, list) else alt.value(color)
    )

    return alt.Chart(df).mark_line(strokeWidth=3).encode(
        x=alt.X('year:N', title='Year'),
        y=alt.Y(f'{y_column}:Q', title=ylab),
        color=color_encoding  
    ).properties(
        title=title
    )

# percentage of measures passing, per party
def get_passes(votes):
    return (votes
        # .filter(_.year >= 2000)
        .group_by("year", "party")
        .aggregate(total=_.count(), passes=_.Status.isin(["Pass", "Pass*"]).sum())
        .mutate(percent_passed=(_.passes / _.total).round(2),
                color=ibis.case().when(_.party == "DEMOCRAT", COLORS["dem_blue"]).else_(COLORS["rep_red"]).end())
        .to_pandas())


# cumulative funding over time 
def funding_chart(votes):
   return (votes
           # .filter(_.year >= 2000)
          .mutate(amount=_.amount.replace('$', '')
                  .replace(',', '')
                  .cast('float64'))
          .filter(_.Status.isin(["Pass", "Pass*"]))
          .group_by("year")
          .aggregate(total_funding=_.amount.sum())
          .order_by("year")
          .mutate(cumulative_funding=_.total_funding.cumsum()/1e9)
          .to_pandas()
         )
    

#color fill for measure status
paint_fill = {
    "fill-color": [
        "case",
        ["==", ["get", "Status"], "Pass"],
        [
            "interpolate", ["linear"], [
                "to-number", ["slice", ["get", "yes"], 0, -1]  # convert 'yes' string to number
            ],
            50, COLORS["grey"],
            55, COLORS["light_green"],
            100, COLORS["dark_green"] # higher yes % -> darker green
        ],
        ["==", ["get", "Status"], "Fail"],
        [
            "interpolate", ["linear"], [
                "to-number", ["slice", ["get", "yes"], 0, -1]
            ],
            0, COLORS["dark_orange"],
            50, COLORS["light_orange"], # lower yes % -> darker orange
            67, COLORS["grey"] # 67 is max in our data 
        ],
        COLORS["grey"]
    ]
}

# for status, height depends on funding 
paint_extrusion = {
    "fill-extrusion-color": paint_fill["fill-color"],
    "fill-extrusion-height": ["*", ["to-number", ["get", "log_amount"]], 5000]
}



# pmtiles style for status 
def get_style_status(jurisdiction):
    if jurisdiction == "State":
        name = "state"
        label = "States"
        paint_type = paint_fill
        layer_type = "fill"
    elif jurisdiction == "County":
        name = "county"
        label = "Counties"
        paint_type = paint_extrusion
        layer_type = "fill-extrusion"
    else:  # Municipal
        name = "municipal"
        label = "Cities"
        paint_type = paint_extrusion
        layer_type = "fill-extrusion"

    return {
        "layers": [
            {
                "id": label,
                "source": name,
                "source-layer": name,
                "type": layer_type,
                "filter": ["==", ["get", "year"], year],
                "paint": paint_type
            }
        ]
    }


# pmtiles style for party 
def get_style_party(jurisdiction):
    if jurisdiction == "State":
        name = "state"
        label = "States"
    elif jurisdiction == "County":
        name = "county"
        label = "Counties"
    else:  # Municipal
        name = "municipal"
        label = "Cities"

    # Return style dictionary for political party
    return {
        "layers": [
            {
                "id": label, 
                "source": name,
                "source-layer": name,
                "type": "fill",
                "filter": [
                    "==", ["get", "year"], year
                ],
                "paint": {
                    "fill-color": {
                        "property": "party",
                        "type": "categorical",
                        "stops": [
                            ["DEMOCRAT", COLORS["dem_blue"]],
                            ["REPUBLICAN", COLORS["rep_red"]]
                        ]
                    }
                }
            }
        ]
    }


sv_pmtiles = "https://data.source.coop/cboettig/social-vulnerability/svi2020_us_county.pmtiles"
sv_style =  {
        "layers": [
            {
                "id": "SVI",
                "source": "Social Vulnerability Index",
                "source-layer": "SVI2020_US_county",
                "type": "fill",
                "paint": {
                    "fill-color": 
                        ["interpolate", ["linear"], ["get", "RPL_THEMES"],
                0, "#FFE6EE",
                1, "#850101"] 
                        
                    }
                }
        ]
}

party_pmtiles = "https://huggingface.co/datasets/boettiger-lab/landvote/resolve/main/party_polygons_all.pmtiles"

recent_election_year = year - year%4 
party_style =  {
        "layers": [
            {
                "id": "Party",
                "source": "Political Parties",
                "source-layer": "county",
                "type": "fill",
                "filter": [
                    "==", ["get", "year"], recent_election_year
                ],
                "paint": {
                    "fill-color": {
                        "property": "party",
                        "type": "categorical",
                        "stops": [
                            ["DEMOCRAT", COLORS["dem_blue"]],
                            ["REPUBLICAN", COLORS["rep_red"]]
                        ]
                    }
                }
            }
        ]
    }


with st.sidebar:
    color_choice = st.radio("Color by:", ["Measure Status", "Political Party"])
    st.divider()

    social_toggle = st.toggle("Social Vulnerability Index")
    party_toggle = st.toggle("Political Parties")
    st.divider()

    '''
    ## Data Assistant (experimental)

    Ask questions about the landvote data, like:

    - What are the top states for approved conservation funds?
    - Plot the total funds spent in conservation each year.
    - What city has approved the most funds in a single measure? What was the description of that vote?
    - Which state has had largest number measures fail? What is that as a fraction of it's total measures?
    '''
    
    prompt = st.chat_input("Ask about the data")
    if prompt:
        with st.spinner():
            resp = agent.chat(prompt)
            if os.path.isfile('exports/charts/temp_chart.png'):
                im = plt.imread('exports/charts/temp_chart.png')
                st.image(im)
                os.remove('exports/charts/temp_chart.png')
            st.write(resp)

              
m = leafmap.Map(style="positron", center=(-100, 40), zoom=3)


if social_toggle:
    m.add_pmtiles(sv_pmtiles, style = sv_style ,visible=True, opacity=0.3, tooltip=True)

if party_toggle:
    m.add_pmtiles(party_pmtiles, style = party_style ,visible=True, opacity=0.3, tooltip=True)

passed = votes.filter(_.Status.isin(["Pass","Pass*"])).count().execute()
total = votes.count().execute()
overall_passed = (passed/total*100).round(2)
f"{overall_passed}% Measures Passed"  

if color_choice == "Measure Status":
    m.add_pmtiles(votes_pmtiles, style=get_style_status("State"), visible=True, opacity=0.8, tooltip=True)
    m.add_pmtiles(votes_pmtiles, style=get_style_status("County"), visible=True, opacity=1.0, tooltip=True)
    m.add_pmtiles(votes_pmtiles, style=get_style_status("Municipal"), visible=True, opacity=1.0, tooltip=True)

elif color_choice == "Political Party":
    m.add_pmtiles(votes_pmtiles, style=get_style_party("State"), visible=True, opacity=0.8, tooltip=True)
    m.add_pmtiles(votes_pmtiles, style=get_style_party("County"), visible=True, opacity=1.0, tooltip=True)
    m.add_pmtiles(votes_pmtiles, style=get_style_party("Municipal"), visible=True, opacity=1.0, tooltip=True)



m.add_layer_control()
m.to_streamlit()


# display charts
df_passes = get_passes(votes)
st.altair_chart(create_chart(df_passes, "percent_passed", "Percent Passed","% of Measures Passed", [COLORS["dem_blue"], COLORS["rep_red"]]), use_container_width=True)

df_funding = funding_chart(votes)
st.altair_chart(create_chart(df_funding, "cumulative_funding", "Billions of Dollars", "Cumulative Funding", COLORS["dark_green"]), use_container_width=True)