Compare commits

...

5 Commits

Author SHA1 Message Date
Felix Blanke
d3c424d90a Add route for single states 2023-09-01 14:47:40 +02:00
Felix Blanke
aaf4ee1863 Add option to not drop NA 2023-09-01 14:46:58 +02:00
Felix Blanke
f9c14e442b Make create_fig more general 2023-09-01 14:46:37 +02:00
Felix Blanke
ef8cab6330 Add option to not fix lims in plot 2023-09-01 14:46:10 +02:00
Felix Blanke
dbf477ed17 Refactor 2023-09-01 14:24:45 +02:00

162
wsgi.py
View File

@ -11,10 +11,10 @@ import matplotlib.pyplot as plt
import matplotlib.ticker as mtick
import numpy as np
import pandas as pd
from flask import Flask, Markup, render_template, request
from flask import Flask, Markup, abort, render_template, request
from flask_caching import Cache
from download_digital import construct_dataframe, get_bez_data, get_landesbezirk
from download_digital import construct_dataframe, get_bez_data, get_landesbezirk, landesbezirk_dict
config = {
"CACHE_TYPE": "FileSystemCache",
@ -23,6 +23,14 @@ config = {
"CACHE_DIR": "cache",
}
abbrev_dict = {
"BBR": "Berlin-Brandenburg",
"BaWü": "Baden-Württemberg",
"NRW": "Nordrhein-Westfalen",
"RLP": "Rheinland-Pfalz-Saarland",
"SAT": "Sachsen, Sachsen-Anhalt, Thüringen",
}
os.environ["TZ"] = "Europe/Berlin"
time.tzset()
@ -96,6 +104,7 @@ def plot(
total_targets: tuple[int, ...] = (1500, 2500, 3500),
alpha: float | None = None,
landesbez_str: str | None = None,
fix_lims: bool = True,
) -> str:
fig = plt.figure(dpi=300)
@ -125,7 +134,7 @@ def plot(
ls="--",
marker="o",
lw=1,
color="#e4004e" if bez is None else None,
color="#e4004e" if bez is None or not fix_lims else None,
markersize=4,
label=bez if bez is not None else "Bundesweit",
)
@ -152,14 +161,15 @@ def plot(
plt.title("Teilnahme an Digitaler Beschäftigtenbefragung")
plt.ylabel("# Teilnahmen")
max_val = df.sum(axis=1).max().item()
if fix_lims:
max_val = df.sum(axis=1).max().item()
nearest_target = np.array(total_targets, dtype=np.float32) - max_val
nearest_target[nearest_target <= 0] = np.inf
idx = np.argmin(nearest_target)
nearest_target = np.array(total_targets, dtype=np.float32) - max_val
nearest_target[nearest_target <= 0] = np.inf
idx = np.argmin(nearest_target)
ceil_val = max(max_val, total_targets[idx])
plt.ylim(0, ceil_val * 1.025)
ceil_val = max(max_val, total_targets[idx])
plt.ylim(0, ceil_val * 1.025)
plt.legend()
# use timezone offset to center tick labels
@ -184,8 +194,9 @@ def plot(
sec_ax.set_ylabel("# Teilnahmen [% Erfolg]")
sec_ax.yaxis.set_major_formatter(mtick.PercentFormatter())
for total_target in total_targets:
plt.axhline(y=total_target, color="#48a9be", linestyle="--")
if fix_lims:
for total_target in total_targets:
plt.axhline(y=total_target, color="#48a9be", linestyle="--")
plt.tight_layout()
@ -195,6 +206,8 @@ def plot(
def create_fig(
url: str = "https://beschaeftigtenbefragung.verdi.de/",
importance_factor: float = 1.0,
landesbez_strs: list[str | None] | None = None,
fix_lims: bool = True,
):
curr_datetime = datetime.datetime.now()
try:
@ -230,17 +243,19 @@ def create_fig(
timestamp = Markup(f'<font color="red">{key} 10:00:00</font>')
total = plot_df.loc[curr_datetime].sum()
landesbez_strs = [None] + [
bez
for bez in plot_df.columns
if plot_df.loc[curr_datetime][bez] >= importance_factor * total
]
if landesbez_strs is None:
landesbez_strs = [None] + [
bez
for bez in plot_df.columns
if plot_df.loc[curr_datetime][bez] >= importance_factor * total
]
return (
plot(
curr_datetime,
plot_df,
annotate_current=annotate_current,
landesbez_str=landesbez_strs,
fix_lims=fix_lims,
),
df,
df_state,
@ -257,73 +272,110 @@ def convert_fig_to_svg(fig: plt.Figure) -> str:
return imgdata.read()
@app.route("/")
@cache.cached(query_string=True)
def tables():
def _print_as_html(df: pd.DataFrame, total: int | None = None) -> None:
df = df.astype({"Digitale Befragung": "Int32"})
def _print_as_html(df: pd.DataFrame, output_str: list[str], total: int | None = None, dropna: bool = True) -> list[str]:
df = df.astype({"Digitale Befragung": "Int32"})
if dropna:
df = df.dropna()
with pd.option_context("display.max_rows", None):
table = df.to_html(
index_names=False,
justify="left",
index=False,
classes="sortable dataframe",
)
with pd.option_context("display.max_rows", None):
table = df.to_html(
index_names=False,
justify="left",
index=False,
classes="sortable dataframe",
)
tfoot = [
" <tfoot>",
" <tr>",
" <td>Gesamt</td>",
tfoot = [
" <tfoot>",
" <tr>",
" <td>Gesamt</td>",
]
for i in range(len(df.columns) - 2):
tfoot.append(" <td></td>")
tfoot.extend(
[
f" <td>{df['Digitale Befragung'].sum()}</td>",
" </tr>",
]
)
if total:
tfoot.extend([
" <tr>",
" <td>Weitere Bezirke</td>",
])
for i in range(len(df.columns) - 2):
tfoot.append(" <td></td>")
tfoot.extend(
[
f" <td>{df['Digitale Befragung'].sum()}</td>",
f" <td>{total - df['Digitale Befragung'].sum()}</td>",
" </tr>",
]
)
if total:
tfoot.extend([
" <tr>",
" <td>Weitere Bezirke</td>",
])
for i in range(len(df.columns) - 2):
tfoot.append(" <td></td>")
tfoot.extend(
[
f" <td>{total - df['Digitale Befragung'].sum()}</td>",
" </tr>",
]
)
tfoot.append(" </tfoot>")
tfoot.append(" </tfoot>")
tfoot = "\n".join(tfoot)
idx = table.index("</table>")
output_str.append(table[: idx - 1])
output_str.append(tfoot)
output_str.append(table[idx:])
tfoot = "\n".join(tfoot)
idx = table.index("</table>")
output_str.append(table[: idx - 1])
output_str.append(tfoot)
output_str.append(table[idx:])
return output_str
@app.route("/<state>")
@cache.cached(query_string=True)
def state_dashboard(state: str):
if state in abbrev_dict:
state = abbrev_dict[state]
if state not in landesbezirk_dict.values():
abort(404)
importance_factor = request.args.get("importance")
if not importance_factor:
importance_factor = 1.0
else:
importance_factor = float(importance_factor)
fig, df, df_state, timestamp = create_fig(landesbez_strs=[state], fix_lims=False)
svg_string = convert_fig_to_svg(fig)
plt.close()
df["Bundesland"] = df.index.map(get_landesbezirk)
df = df.rename(columns={"Bundesland": "Landesbezirk"})
df_state = df_state.loc[df_state["Landesbezirk"] == state]
df = df.loc[df["Landesbezirk"] == state]
output_str = []
output_str = _print_as_html(df_state, output_str, dropna=False)
output_str = _print_as_html(df, output_str, total=df_state['Digitale Befragung'].sum(), dropna=False)
return render_template(
"base.html",
tables="\n".join(output_str),
timestamp=timestamp,
image=svg_string,
)
@app.route("/")
@cache.cached(query_string=True)
def dashboard():
importance_factor = request.args.get("importance")
if not importance_factor:
importance_factor = 1.0
else:
importance_factor = float(importance_factor)
fig, df, df_state, timestamp = create_fig(importance_factor=importance_factor)
svg_string = convert_fig_to_svg(fig)
plt.close()
_print_as_html(df_state)
df["Bundesland"] = df.index.map(get_landesbezirk)
df = df.rename(columns={"Bundesland": "Landesbezirk"})
_print_as_html(df, total=df_state['Digitale Befragung'].sum())
output_str = []
output_str = _print_as_html(df_state, output_str, dropna=False)
output_str = _print_as_html(df, output_str, total=df_state['Digitale Befragung'].sum())
return render_template(
"base.html",