from __future__ import annotations
import copy
import re
import threading
import faicons as fa
import plotly.express as px
import plotly.graph_objects as go
import polars as pl
from shiny import ui
SCB_SOURCE_MD = (
"Source: [Swedish Occupational Register, SCB]"
"(https://www.scb.se/en/finding-statistics/statistics-by-subject-area/"
"labour-market/labour-force-supply/"
"the-swedish-occupational-register-with-statistics/)"
)
DAIOE_SOURCE_MD = "Source: [DAIOEs](https://www.ai-econlab.com/ai-exposure-daioe)"
# Matches leading emoji glyphs only (not general non-ASCII text such as Swedish
# å/ä/ö), so label text is never mistaken for a decorative prefix. Ranges cover
# the main emoji blocks, misc symbols & dingbats (e.g. U+265F "♟"), variation
# selector-16 (U+FE0F) and zero-width joiner (U+200D) for multi-codepoint emoji.
_EMOJI_PREFIX = re.compile(
r"^[\U0001F000-\U0001FAFF☀-➿️]+\s*",
)
# Brand colours from _brand.yml
_C_BG = "rgba(0,0,0,0)"
_C_GRID = "#E5E5E5"
_C_TEXT = "#1C2826"
_C_TITLE = "#0C0A3E"
_FONT_BASE = "Nunito Sans"
_FONT_HEAD = "Montserrat"
_BASE_LAYOUT: dict = {
"paper_bgcolor": _C_BG,
"plot_bgcolor": _C_BG,
"font": {"family": _FONT_BASE, "color": _C_TEXT, "size": 13},
"title_font": {"family": _FONT_HEAD, "color": _C_TITLE, "size": 15},
"hoverlabel": {"font": {"family": _FONT_BASE, "size": 12}},
"margin": {"l": 20, "r": 20, "t": 45, "b": 20},
}
_kaleido_lock = threading.Lock()
_kaleido_started = False
def _ensure_kaleido() -> None:
"""Start the kaleido server if not already running (thread-safe)."""
global _kaleido_started # noqa: PLW0603
with _kaleido_lock:
if not _kaleido_started:
import kaleido
kaleido.start_sync_server(silence_warnings=True)
_kaleido_started = True
# Pre-warm kaleido in background so the first PNG download is not blocked.
threading.Thread(target=_ensure_kaleido, daemon=True).start()
def _empty_figure() -> go.Figure:
"""Return a blank figure with a centered 'No data available' annotation."""
fig = go.Figure()
fig.add_annotation(
text="No data available",
showarrow=False,
font={"size": 16, "color": "#999"},
xref="paper",
yref="paper",
x=0.5,
y=0.5,
)
fig.update_layout(**_BASE_LAYOUT)
return fig
def _apply_xaxes(fig: go.Figure) -> None:
fig.update_xaxes(
gridcolor=_C_GRID,
zeroline=False,
tickangle=-45,
tickformat="%b %Y",
dtick="M3",
)
def _apply_yaxes(fig: go.Figure) -> None:
fig.update_yaxes(gridcolor=_C_GRID, zeroline=False)
def _hlegend() -> dict:
return {
"orientation": "h",
"yanchor": "bottom",
"y": -0.35,
"xanchor": "center",
"x": 0.5,
"title": None,
}
def build_value_boxes(summary: pl.DataFrame, occupation: str) -> ui.Tag:
"""
Build the employment summary value boxes for a given occupation.
Returns a div containing a heading, three value boxes (employment count,
1-month change, 3-month change), and a markdown source note.
Raises IndexError if summary is empty — callers must guard with is_empty().
"""
def _arrow(v: float) -> str:
return "▼" if v < 0 else "▲"
def _theme(v: float) -> str:
return "danger" if v < 0 else "success"
def _fmt_pct(v: float | None) -> str:
return f"{_arrow(v)} {v:.0f}%" if v is not None else "N/A"
def _fmt_theme(v: float | None) -> str:
return _theme(v) if v is not None else "secondary"
row = summary.row(0, named=True)
emp = row["emp_count"]
pct1 = row["pct_chg_1m"]
pct3 = row["pct_chg_3m"]
month = row["month"]
return ui.div(
ui.h6(
f"National Employment of {occupation} (All Genders)",
class_="mt-3 mb-2 fw-semibold",
),
ui.layout_columns(
ui.value_box(
title="Employment ('000)",
showcase=fa.icon_svg("users"),
value=f"{emp:,.0f}",
theme="primary",
),
ui.value_box(
title="1-month change",
value=_fmt_pct(pct1),
showcase=fa.icon_svg(
"arrow-trend-up"
if pct1 is None or pct1 >= 0
else "arrow-trend-down",
),
theme=_fmt_theme(pct1),
),
ui.value_box(
title="3-month change",
value=_fmt_pct(pct3),
showcase=fa.icon_svg(
"arrow-trend-up"
if pct3 is None or pct3 >= 0
else "arrow-trend-down",
),
theme=_fmt_theme(pct3),
),
col_widths=[4, 4, 4],
),
ui.markdown(f"Employment count as at **{month}**.\n\n{SCB_SOURCE_MD}"),
)
def build_employment_count_chart(
df: pl.DataFrame,
occupation: str,
*,
smooth: bool = False,
) -> go.Figure:
"""
Build a Plotly line chart of total monthly employment count over time.
1-month % change is shown on hover. When df contains multiple gender series,
each is drawn as a separate coloured line. Returns an empty figure if df is empty.
"""
if df.is_empty():
return _empty_figure()
multi_gender = "gender" in df.columns and df["gender"].n_unique() > 1
df = df.with_columns(
pl.when(pl.col("pct_chg_1m").is_not_null())
.then(pl.col("pct_chg_1m").round(1).cast(pl.String) + pl.lit("%"))
.otherwise(pl.lit("N/A"))
.alias("_pct_label"),
).sort(["gender", "month_date"] if multi_gender else ["month_date"])
fig = px.line(
df,
x="month_date",
y="emp_count",
color="gender" if multi_gender else None,
markers=True,
custom_data=["_pct_label", "month"],
labels={"month_date": "Month", "emp_count": "Employment", "gender": "Gender"},
)
fig.update_traces(
line={"width": 3},
marker={"size": 8},
hovertemplate=(
"Month: %{customdata[1]}
"
"Employment: %{y:,.0f}
"
"1-mo Change: %{customdata[0]}"
),
)
title_suffix = " (3-Month Moving Average)" if smooth else ""
fig.update_layout(
**_BASE_LAYOUT,
title={
"text": f"Monthly Employment of {occupation} in Sweden{title_suffix}",
"x": 0.01,
"xanchor": "left",
},
showlegend=multi_gender,
**({"legend": _hlegend()} if multi_gender else {}),
)
_apply_xaxes(fig)
_apply_yaxes(fig)
return fig
def build_employment_chart(
df: pl.DataFrame,
occupation: str,
*,
smooth: bool = False,
) -> go.Figure:
"""
Build a Plotly line chart of total 1-month employment % change over time.
Absolute employment count is shown on hover. When df contains multiple gender
series, each is drawn as a separate coloured line. Returns an empty figure if
df is empty.
"""
if df.is_empty():
return _empty_figure()
multi_gender = "gender" in df.columns and df["gender"].n_unique() > 1
df = df.sort(["gender", "month_date"] if multi_gender else ["month_date"])
fig = px.line(
df,
x="month_date",
y="pct_chg_1m",
color="gender" if multi_gender else None,
markers=True,
custom_data=["emp_count", "month"],
labels={
"month_date": "Month",
"pct_chg_1m": "Employment change (%)",
"gender": "Gender",
},
)
fig.update_traces(
line={"width": 3},
marker={"size": 8},
hovertemplate=(
"Month: %{customdata[1]}
"
"Change: %{y:.1f}%
"
"Employment: %{customdata[0]:,.0f}"
),
connectgaps=True,
)
fig.add_hline(y=0, line_color="grey", line_width=1)
title_suffix = " (3-Month Moving Average)" if smooth else ""
fig.update_layout(
**_BASE_LAYOUT,
title={
"text": f"Monthly Employment Change of {occupation} in Sweden{title_suffix}",
"x": 0.01,
"xanchor": "left",
},
yaxis={"ticksuffix": "%"},
showlegend=multi_gender,
**({"legend": _hlegend()} if multi_gender else {}),
)
_apply_xaxes(fig)
_apply_yaxes(fig)
return fig
def build_comparison_employment_plot(
df: pl.DataFrame,
*,
smooth: bool = False,
) -> go.Figure:
"""Build a line chart comparing 1-month employment % change across selected occupations."""
if df.is_empty():
return _empty_figure()
df = df.sort(["occupation", "month_date"])
fig = px.line(
df,
x="month_date",
y="pct_chg_1m",
color="occupation",
markers=True,
custom_data=["emp_count", "month"],
labels={"pct_chg_1m": "Employment Change (%)", "month_date": "Month"},
)
fig.update_traces(
line={"width": 3},
marker={"size": 8},
hovertemplate=(
"%{fullData.name}
"
"Month: %{customdata[1]}
"
"Change: %{y:.1f}%
"
"Employment: %{customdata[0]:,.0f}"
),
connectgaps=True,
)
fig.add_hline(y=0, line_color="grey", line_width=1)
title_suffix = " (3-Month Moving Average)" if smooth else ""
fig.update_layout(
**_BASE_LAYOUT,
title={
"text": f"Monthly Employment Change by Occupation in Sweden{title_suffix}",
"x": 0.01,
"xanchor": "left",
},
legend=_hlegend(),
yaxis={"ticksuffix": "%"},
)
_apply_xaxes(fig)
_apply_yaxes(fig)
return fig
def build_comparison_employment_count_plot(
df: pl.DataFrame,
*,
smooth: bool = False,
) -> go.Figure:
"""Build a line chart comparing absolute monthly employment counts across selected occupations."""
if df.is_empty():
return _empty_figure()
df = df.with_columns(
pl.when(pl.col("pct_chg_1m").is_not_null())
.then(pl.col("pct_chg_1m").round(1).cast(pl.String) + pl.lit("%"))
.otherwise(pl.lit("N/A"))
.alias("_pct_label"),
).sort(["occupation", "month_date"])
title_suffix = " (3-Month Moving Average)" if smooth else ""
fig = px.line(
df,
x="month_date",
y="emp_count",
color="occupation",
markers=True,
custom_data=["_pct_label", "month"],
labels={"emp_count": "Employment ('000)", "month_date": "Month"},
)
fig.update_traces(
line={"width": 3},
marker={"size": 8},
hovertemplate=(
"%{fullData.name}
"
"Month: %{customdata[1]}
"
"Employment: %{y:,.0f}
"
"1-mo Change: %{customdata[0]}"
),
)
fig.update_layout(
**_BASE_LAYOUT,
title={
"text": f"Monthly Employment by Occupation in Sweden{title_suffix}",
"x": 0.01,
"xanchor": "left",
},
legend=_hlegend(),
)
_apply_xaxes(fig)
_apply_yaxes(fig)
return fig
def build_comp_radar_plot(df: pl.DataFrame, metrics: dict[str, str]) -> go.Figure:
"""Build a radar chart comparing AI percentile scores across selected occupations."""
if df.is_empty():
return _empty_figure()
categories = list(metrics.values())
fig = go.Figure()
for row in df.to_dicts():
r_values = [row[f"pctl_{k}_wavg"] for k in metrics]
r_values_closed = [*r_values, r_values[0]]
categories_closed = [*categories, categories[0]]
fig.add_trace(
go.Scatterpolar(
r=r_values_closed,
theta=categories_closed,
fill="toself",
name=row["occupation"],
hovertemplate="%{theta}: %{r:.1f}%",
),
)
fig.update_layout(
**_BASE_LAYOUT,
polar={"radialaxis": {"visible": True, "range": [0, 100]}},
showlegend=True,
legend={
"orientation": "h",
"yanchor": "bottom",
"y": -0.25,
"xanchor": "center",
"x": 0.5,
},
)
return fig
def build_ai_exposure_bar(
df: pl.DataFrame,
occupation: str,
year: int,
) -> go.Figure:
"""
Build a horizontal bar chart of AI exposure level per sub-domain.
Bar colour intensity is driven by the percentile rank score.
Hover shows exposure level label, index score, and percentile rank.
"""
if df.is_empty():
return _empty_figure()
fig = go.Figure(
go.Bar(
x=df["percentile"].to_list(),
y=df["domain"].to_list(),
orientation="h",
marker={
"color": df["percentile"].to_list(),
"colorscale": "Blues",
"colorbar": {"title": "Percentile Rank"},
"showscale": True,
"cmin": 0,
"cmax": 100,
},
customdata=list(
zip(
df["level_label"].to_list(),
df["level"].to_list(),
df["score"].to_list(),
strict=False,
),
),
hovertemplate=(
"%{y}
"
"Percentile Rank: %{x:.0f}
"
"Exposure Level: %{customdata[0]} (%{customdata[1]}/5)
"
"Index Score: %{customdata[2]:.3f}"
),
),
)
fig.update_layout(
**_BASE_LAYOUT,
title={
"text": f"{occupation} Level of AI Exposure ({year})",
"x": 0.01,
"xanchor": "left",
},
xaxis={"title": "Percentile Rank", "range": [0, 100]},
yaxis={"title": None},
)
fig.update_xaxes(gridcolor=_C_GRID, zeroline=False)
fig.update_yaxes(gridcolor=_C_GRID, zeroline=False)
return fig
def _strip_emoji(val: object) -> object:
if isinstance(val, str):
return _EMOJI_PREFIX.sub("", val)
if isinstance(val, (list, tuple)):
stripped = [_EMOJI_PREFIX.sub("", v) if isinstance(v, str) else v for v in val]
return type(val)(stripped)
return val
def export_fig(fig: go.Figure, width: int = 1000, height: int = 650) -> bytes:
"""Return PNG bytes of a figure with a solid white background and no emoji labels."""
_ensure_kaleido()
fig = copy.deepcopy(fig)
for trace in fig.data:
for field in ("y", "x", "theta", "text", "name"):
val = getattr(trace, field, None)
if val is not None:
trace.update({field: _strip_emoji(val)}) # type: ignore[union-attr]
is_polar = any(getattr(t, "type", "") == "scatterpolar" for t in fig.data)
fig.update_layout(paper_bgcolor="white", plot_bgcolor="white")
if is_polar:
fig.update_layout(polar_bgcolor="white")
return fig.to_image(format="png", scale=2, width=width, height=height)