fblanke/tvstud-download-digital
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: fblanke:d3c424d90aa5e2fe29f6b0306efada43bd1f3774
Branches Tags
fblanke:main
..
compare: fblanke:0292db36e2f3c8ba09290ec6c6121773d2c72686
Branches Tags
fblanke:main

No commits in common. "d3c424d90aa5e2fe29f6b0306efada43bd1f3774" and "0292db36e2f3c8ba09290ec6c6121773d2c72686" have entirely different histories.
d3c424d90a ... 0292db36e2

1 changed files with 55 additions and 107 deletions
Show Stats Expand all files Collapse all files

162
wsgi.py
View File

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