app.py

import streamlit as st
import streamlit.components.v1 as components
import base64
import leafmap.maplibregl as leafmap
import altair as alt
import ibis
from ibis import _
import ibis.selectors as s


from typing import Optional
def to_streamlit(
    self,
    width: Optional[int] = None,
    height: Optional[int] = 600,
    scrolling: Optional[bool] = False,
    **kwargs,
    ):

    try:
        import streamlit.components.v1 as components
        import base64

        raw_html = self.to_html().encode("utf-8")
        raw_html = base64.b64encode(raw_html).decode()
        return components.iframe(
            f"data:text/html;base64,{raw_html}",
            width=width,
            height=height,
            scrolling=scrolling,
            **kwargs,
        )

    except Exception as e:
        raise Exception(e)


ca_pmtiles = "https://huggingface.co/datasets/boettiger-lab/ca-30x30/resolve/main/ca2024-30m-tippe.pmtiles"
ca_parquet = "https://huggingface.co/datasets/boettiger-lab/ca-30x30/resolve/main/ca2024-30m.parquet"



ca_area_acres = 1.014e8 #acres 
style_choice = "GAP Status Code"

con = ibis.duckdb.connect(extensions=["spatial"])

ca = (con
      .read_parquet(ca_parquet)
      .cast({"geom": "geometry"})
     )
 


private_access_color = "#DE881E" # orange 
public_access_color = "#3388ff" # blue
tribal_color = "#BF40BF" # purple
mixed_color = "#005a00" # green
year2023_color = "#26542C" # green
year2024_color = "#F3AB3D" # orange 

federal_color = "#529642" # green
state_color = "#A1B03D" # light green
local_color = "#365591" # blue
special_color = "#529642" # brown
private_color = "#7A3F1A" # brown
joint_color = "#DAB0AE" # pink
county_color = "#BFD76B" # green
city_color = "#BDC368" #green
hoa_color = "#A89BBC" # purple
nonprofit_color =  "#D77031" #orange


from functools import reduce

def summary_table(column, colors, filter_cols, filter_vals):
    filters = [_.reGAP < 3] #only compute with gap codes 1 and 2
    
    if filter_cols and filter_vals: #if a filter is selected, add to list of filters 
        for filter_col, filter_val in zip(filter_cols, filter_vals):
            if len(filter_val) > 1:
                filters.append(getattr(_, filter_col).isin(filter_val))
            else:
                filters.append(getattr(_, filter_col) == filter_val[0])
                
    combined_filter = reduce(lambda x, y: x & y, filters)
    
    df = (ca
        .filter(combined_filter)
        .group_by(column)
        .aggregate(percent_protected=100 * _.Acres.sum() / ca_area_acres)
        .mutate(percent_protected=_.percent_protected.round(1))
        .inner_join(colors, column)
    )
    df = df.to_pandas()
    df[column] = df[column].astype(str)
    return df



def area_plot(df, column):
    base = alt.Chart(df).encode(
        alt.Theta("percent_protected:Q").stack(True),
    )
    pie = ( base
           .mark_arc(innerRadius= 40, outerRadius=100)
           .encode(alt.Color("color:N").scale(None).legend(None),
                   tooltip=['percent_protected', column])
    )
    text = ( base
            .mark_text(radius=80, size=14, color="white")
            .encode(text = column + ":N")
    )
    plot = pie # pie + text
    return plot.properties(width="container", height=300)


    
def get_pmtiles_style(paint, alpha, cols, values):
    filters = []
    
    for col, val in zip(cols, values):
        filter_condition = ["match", ["get", col], val, True, False]
        filters.append(filter_condition)
    combined_filter = ["all"] + filters

    return {
        "version": 8,
        "sources": {
            "ca": {
                "type": "vector",
                "url": "pmtiles://" + ca_pmtiles,
            }
        },
        "layers": [{
            "id": "ca30x30",
            "source": "ca",
            "source-layer": "ca202430m",
            # "source-layer": "ca2024",
            "type": "fill",
            "filter": combined_filter,  # Use the combined filter
            "paint": {
                "fill-color": paint,
                "fill-opacity": alpha
            }
        }]
    }


def getButtons(style_options, style_choice, default_gap=None):
    column = style_options[style_choice]['property']
    opts = [style[0] for style in style_options[style_choice]['stops']]   
    default_gap = default_gap or {}  
    buttons = {
        name: st.checkbox(f"{name}", value=default_gap.get(name, False), key=column + str(name))
        for name in opts
    }
    filter_choice = [key for key, value in buttons.items() if value]  # return only selected
    d = {}
    d[column] = filter_choice
    return d

default_gap = {
    1: True,  
    2: True,  
}


manager = {
            'property': 'manager_type',
            'type': 'categorical',
            'stops': [
                ['Federal', federal_color],
                ['State', state_color],
                ['Non Profit', nonprofit_color],
                ['Special District', special_color],
                ['Unknown', "grey"],
                ['County', county_color],
                ['City', city_color],
                ['Joint', joint_color],
                ['Tribal', tribal_color],
                ['Private', private_color],
                ['Home Owners Association', hoa_color]
            ]
            }


easement = {
            'property': 'Easement',
            'type': 'categorical',
            'stops': [
                [0, public_access_color],
                [1, private_access_color]
            ]
            }


year = {
            'property': 'established',
            'type': 'categorical',
            'stops': [
                [2023, year2023_color],
                [2024, year2024_color]
            ]
            }


access = {
    'property': 'access_type',
    'type': 'categorical',
    'stops': [
        ['Open Access', public_access_color],
        ['No Public Access', private_access_color],
        ['Unknown Access', "grey"],
        ['Restricted Access', tribal_color]
    ]
}



gap = {
        'property': 'reGAP',
        'type': 'categorical',
        'stops': [
            [1, "#26633d"],
            [2, "#879647"],
            [3, "#BBBBBB"],
            [4, "#F8F8F8"]
        ]
        }


area_type = {
        'property': 'type',
        'type': 'categorical',
        'stops': [
            ["Land", "green"],
            ["Water", "blue"]
        ]
        }


style_options = {
                    "GAP Status Code": gap,
                    "Manager Type": manager,
                    "Easement": easement,
                    "Public Access": access,
                    "Year": year,
                    "Type": area_type
                    }


st.set_page_config(layout="wide", page_title="CA Protected Areas Explorer", page_icon=":globe:")

'''
# CA 30X30 Prototype

'''

m = leafmap.Map(style="positron")
filters = {}

with st.sidebar:
    color_choice = st.radio("Color by:", style_options)
    alpha = st.slider("transparency", 0.0, 1.0, 0.5)
    # st.text("Filters:")
    "Filters:"
    for label in style_options:
        with st.expander(label): 
             
            if label == "GAP Status Code": # gap code 1 and 2 are on by default
                opts = getButtons(style_options, label, default_gap)
            else:
                opts = getButtons(style_options, label)
            filters.update(opts)
            
        selected = {k: v for k, v in filters.items() if v} #get selected filters
        if selected: 
            filter_cols = list(selected.keys())
            filter_vals = list(selected.values())
        else: 
            filter_cols = []
            filter_vals = []
            
    style = get_pmtiles_style(style_options[color_choice], alpha, filter_cols, filter_vals)
    m.add_pmtiles(ca_pmtiles, style=style, visible=True, name="CA", opacity=alpha, tooltip=True)


select_column = {
                "GAP Status Code": "reGAP",
                "Manager Type": "manager_type",
                "Easement": "Easement",
                "Year": "established",
                "Public Access": "access_type",
                "Type": "type"}


column = select_column[color_choice]

select_colors = {
                "Manager Type": manager["stops"],
                "Easement": easement["stops"],
                "Public Access": access["stops"],
                "Year": year["stops"],
                "GAP Status Code": gap["stops"],
                "Type": area_type["stops"]}

colors = (ibis
          .memtable(select_colors[color_choice], columns = [column, "color"])
          .to_pandas()
         )

main = st.container()
    
with main:
    map_col, stats_col = st.columns([2,1])
    with map_col:
        to_streamlit(m, height=700)
    df = summary_table(column, colors, filter_cols, filter_vals)
    total_percent = df.percent_protected.sum().round(1)
    with stats_col:
        with st.container():
            f"{total_percent}% CA Covered"
            st.altair_chart(area_plot(df, column), use_container_width=True)



st.divider()
footer = st.container()