diff options
| author | Leonard Kugis <leonard@kug.is> | 2025-12-23 03:50:47 +0100 |
|---|---|---|
| committer | Leonard Kugis <leonard@kug.is> | 2025-12-23 03:50:47 +0100 |
| commit | d342b3d00f64915685b486a68b7c7b3e2e47fde6 (patch) | |
| tree | a9e9ae070d79737e1a62b275d18472793250ae43 | |
| parent | ec7598f568ff59ecc1eb51572f84d866b0180501 (diff) | |
| download | xembu-d342b3d00f64915685b486a68b7c7b3e2e47fde6.tar.gz | |
Format corrections, english language
| -rw-r--r-- | modules/general.py | 200 | ||||
| -rw-r--r-- | xembu.py | 38 |
2 files changed, 178 insertions, 60 deletions
diff --git a/modules/general.py b/modules/general.py index 3ad587a..d5db52f 100644 --- a/modules/general.py +++ b/modules/general.py @@ -31,8 +31,8 @@ def compute_group_distribution(df: pd.DataFrame): bad_units = contrib[~contrib["unit"].apply(_is_money_unit)] if len(bad_units) > 0: raise ValueError( - "Contribution (C) muss Geld-Einheit haben (z.B. € / EUR). " - f"Problemzeilen:\n{bad_units[['date','debitor','group','position','val','unit']]}" + "Contribution (C) needs to have currency (e.g. € / EUR). " + f"Lines:\n{bad_units[['date','debitor','group','position','val','unit']]}" ) usage = work[work["flag"] == "U"].copy() @@ -238,15 +238,15 @@ class GroupChartBigFrame(BigFrame): if self.kind == "usage_cum": series_map = self.gts.usage_cum unit = "/".join(self.gts.usage_units) if self.gts.usage_units else "" - ax.set_ylabel(f"Usage cumulative {unit}".strip(), fontproperties=mono_font) + ax.set_ylabel(f"Usage cumulative [{unit}]".strip(), fontproperties=mono_font) elif self.kind == "contrib_cum": series_map = self.gts.contrib_cum - ax.set_ylabel("Contribution cumulative €", fontproperties=mono_font) + ax.set_ylabel("Contribution cumulative [€]", fontproperties=mono_font) elif self.kind == "share_cum": series_map = self.gts.share_cum - ax.set_ylabel("Share cumulative €", fontproperties=mono_font) + ax.set_ylabel("Share cumulative [€]", fontproperties=mono_font) elif self.kind == "ratio": series_map = self.gts.ratio @@ -270,7 +270,17 @@ class GroupChartBigFrame(BigFrame): else: y = y.replace([np.inf, -np.inf], np.nan).fillna(0.0) - ax.plot(self.gts.times, y.values, label=p, linewidth=1, drawstyle="steps-post") + line, = ax.plot(self.gts.times, y.values, label=p, linewidth=1, drawstyle="steps-post") + base = (min_ratio if self.kind == "ratio" else 0.0) # log-Plot kann nicht bis 0 füllen + ax.fill_between( + self.gts.times, + y.values, + base, + step="post", + alpha=0.18, + color=line.get_color(), + zorder=line.get_zorder() - 1, + ) v = y.values v = v[np.isfinite(v)] @@ -331,6 +341,106 @@ class PlotBigFrame(BigFrame): ax.xaxis.label.set_fontproperties(mono_font) ax.yaxis.label.set_fontproperties(mono_font) +def _fmt_groups_table(group_summary: pd.DataFrame) -> List[str]: + gs = group_summary.sort_values("group").copy() + + g_list = gs["group"].astype(str).tolist() + g_w = max([len(g) for g in g_list] + [5]) + + header = f"{'group':<{g_w}} | {'total [€]':>12} | {'mode':<12} | {'participants':>12}" + sep = "-" * len(header) + + lines = [header, sep] + for _, r in gs.iterrows(): + g = str(r["group"]) + total_c = float(r.get("total_contrib", 0.0) or 0.0) + u_count = int(r.get("u_count", 0) or 0) + mode = "usage" if u_count > 0 else "equal" + participants = r.get("participants", []) or [] + lines.append(f"{g:<{g_w}} | {total_c:>12.2f} | {mode:<12} | {len(participants):>12d}") + + return lines + + +def _fmt_debitors_table(per_debitor: pd.DataFrame) -> List[str]: + pdv = per_debitor.sort_values("debitor").copy() + + name_list = pdv["debitor"].astype(str).tolist() + name_w = max([len(n) for n in name_list] + [7]) + + header = f"{'debitor':<{name_w}} | {'contrib [€]':>12} | {'share [€]':>12} | {'balance [€]':>12}" + sep = "-" * len(header) + + lines = [header, sep] + for _, r in pdv.iterrows(): + deb = str(r["debitor"]) + lines.append( + f"{deb:<{name_w}} | {float(r['contributed']):>12.2f} | {float(r['share']):>12.2f} | {float(r['balance']):>12.2f}" + ) + return lines + + +def _fmt_compensation_table(payments: List[Tuple[str, str, float]]) -> List[str]: + if not payments: + return ["(No compensation required)"] + + payer_w = max([len(p) for p, _, _ in payments] + [5]) + recv_w = max([len(r) for _, r, _ in payments] + [8]) + + header = f"{'payer':<{payer_w}} | {'receiver':<{recv_w}} | {'amount [€]':>12}" + sep = "-" * len(header) + + lines = [header, sep] + for p, r, a in payments: + lines.append(f"{p:<{payer_w}} | {r:<{recv_w}} | {float(a):>12.2f}") + return lines + + +def _fmt_group_detail_block(group: str, group_summary: pd.DataFrame, detail: pd.DataFrame) -> List[str]: + gdet = detail[detail["group"] == group].sort_values("debitor") + + if (group_summary["group"] == group).any(): + gs = group_summary[group_summary["group"] == group].iloc[0] + total_c = float(gs.get("total_contrib", 0.0) or 0.0) + total_u = float(gs.get("total_usage", 0.0) or 0.0) + u_count = int(gs.get("u_count", 0) or 0) + mode = "usage" if u_count > 0 else "equal" + participants = gs.get("participants", []) or [] + usage_units = gs.get("usage_units", []) or [] + if isinstance(usage_units, float) and pd.isna(usage_units): + usage_units = [] + units_str = "/".join(usage_units) if isinstance(usage_units, list) else str(usage_units) + else: + total_c = 0.0 + total_u = 0.0 + mode = "equal" + participants = [] + units_str = "" + + # Kopf + lines: List[str] = [] + lines.append(f"Group: {group}") + lines.append(f"Total contribution: {total_c:.2f} €") + lines.append(f"Total usage: {total_u:.4f} {units_str}".rstrip()) + lines.append(f"Mode: {mode}") + lines.append(f"Participants: {len(participants)}") + lines.append("") + + # Tabelle + name_w = max([len(str(x)) for x in gdet["debitor"].astype(str).tolist()] + [7]) + header = f"{'debitor':<{name_w}} | {'contrib [€]':>10} | {'usage':>12} | {'share [€]':>10} | {'balance [€]':>10}" + sep = "-" * len(header) + + lines.append(header) + lines.append(sep) + + for _, r in gdet.iterrows(): + deb = str(r["debitor"]) + lines.append( + f"{deb:<{name_w}} | {r['contributed']:>10.2f} | {r['usage']:>12.4f} | {r['share']:>10.2f} | {r['balance']:>10.2f}" + ) + + return lines class GeneralModule: name = "general" @@ -346,31 +456,32 @@ class GeneralModule: payments = self._minimize_payments(balance) summary_lines = [] - summary_lines.append("General") + summary_lines.append("# GeneralModule") summary_lines.append("") - summary_lines.append("Goups:") - for _, r in group_summary.sort_values("group").iterrows(): - g = r["group"] - total_c = float(r.get("total_contrib", 0.0)) - u_count = int(r.get("u_count", 0)) - mode = "usage" if u_count > 0 else "equal" - participants = r.get("participants", []) or [] - summary_lines.append(f" - {g}: {total_c:.2f} €; mode={mode}; participants={len(participants)}") + summary_lines.append("## Groups") summary_lines.append("") - summary_lines.append("Debitors (total):") - for _, r in per_debitor.sort_values("debitor").iterrows(): - summary_lines.append( - f" - {r['debitor']}: contributed={r['contributed']:.2f} €; share={r['share']:.2f} €; balance={r['balance']:.2f} €" - ) + summary_lines.extend(_fmt_groups_table(group_summary)) summary_lines.append("") - summary_lines.append("Compensation (minimized):") - if payments: - for p, r, a in payments: - summary_lines.append(f" - {p} → {r}: {a:.2f} €") - else: - summary_lines.append(" (No compensation required)") + summary_lines.append("## Group details") + summary_lines.append("") + for g in sorted(detail["group"].unique().tolist()): + # optional: Markdown-Überschrift zusätzlich + summary_lines.append(f"### {g}") + summary_lines.append("") + summary_lines.extend(_fmt_group_detail_block(g, group_summary, detail)) + summary_lines.append("") + + summary_lines.append("") + summary_lines.append("## Debitors (total)") + summary_lines.append("") + summary_lines.extend(_fmt_debitors_table(per_debitor)) + + summary_lines.append("") + summary_lines.append("## Compensation (minimized)") + summary_lines.append("") + summary_lines.extend(_fmt_compensation_table(payments)) summary_text = "\n".join(summary_lines) @@ -444,28 +555,16 @@ class GeneralModule: return out def _make_frames(self, group_summary: pd.DataFrame, per_debitor: pd.DataFrame, payments: List[Tuple[str,str,float]]) -> List[Frame]: - lines = ["Groups:"] - for _, r in group_summary.sort_values("group").iterrows(): - g = r["group"] - total_c = float(r.get("total_contrib", 0.0)) - u_count = int(r.get("u_count", 0)) - parts = r.get("participants", []) - mode = "usage" if u_count > 0 else "equal" - lines.append(f"- {g}: {total_c:.2f} €; mode={mode}; participants={len(parts)}") - + lines = ["Groups:", ""] + lines.extend(_fmt_groups_table(group_summary)) f1 = TextFrame(title="General: Groups", text="\n".join(lines)) - lines = ["Debitor total:", "debitor | contributed | share | balance"] - for _, r in per_debitor.iterrows(): - lines.append(f"{r['debitor']} | {r['contributed']:.2f} € | {r['share']:.2f} € | {r['balance']:.2f} €") + lines = ["Debitors (total):", ""] + lines.extend(_fmt_debitors_table(per_debitor)) f2 = TextFrame(title="General: Debitors", text="\n".join(lines)) - lines = ["Compensation (minimized):"] - if payments: - for p, r, a in payments: - lines.append(f"{p} → {r}: {a:.2f} €") - else: - lines.append("(No compensation required)") + lines = ["Compensation (minimized):", ""] + lines.extend(_fmt_compensation_table(payments)) f3 = TextFrame(title="General: Compensation", text="\n".join(lines)) return [f1, f2, f3] @@ -479,18 +578,7 @@ class GeneralModule: u_count = int(group_summary[group_summary["group"] == g]["u_count"].iloc[0]) if (group_summary["group"] == g).any() else 0 mode = "usage" if u_count > 0 else "equal" - lines = [ - f"Group: {g}", - f"Total Contribution: {total_c:.2f} €", - f"Mode: {mode}", - "", - "debitor | contributed | usage | share | balance", - ] - for _, r in gdet.iterrows(): - lines.append( - f"{r['debitor']} | {r['contributed']:.2f} € | {r['usage']:.4f} | {r['share']:.2f} € | {r['balance']:.2f} €" - ) - + lines = _fmt_group_detail_block(g, group_summary, detail) fig, ax = plt.subplots(figsize=(8.27, 11.69)) ax.axis("off") ax.text(0, 1, "\n".join(lines), va="top", ha="left", fontproperties=mono_font) @@ -12,9 +12,13 @@ from matplotlib import font_manager from modules.base import Frame, BigFrame, Module, ModuleResult from modules.general import GeneralModule +from modules.drug import DrugModule from datetime import datetime +import logging +import sys + CSV_COLUMNS = [ "date", "debitor", @@ -27,6 +31,28 @@ CSV_COLUMNS = [ "receipt", ] +logger = logging.getLogger("xembu") + +def setup_logging(verbose: int = 0): + level = logging.ERROR + if verbose == 1: + level = logging.WARNING + elif verbose == 2: + level = logging.INFO + elif verbose >= 3: + level = logging.DEBUG + + logger.setLevel(level) + + handler = logging.StreamHandler(sys.stderr) + handler.setLevel(level) + + fmt = logging.Formatter("[%(levelname)s] %(message)s") + handler.setFormatter(fmt) + + logger.handlers.clear() + logger.addHandler(handler) + def _pick_mono_font(size: int = 8) -> font_manager.FontProperties: for fam in ["Inconsolata", "DejaVu Sans Mono", "monospace"]: try: @@ -386,8 +412,11 @@ def main(): parser.add_argument("--title", "-t", help="PDF header title (optional)") parser.add_argument("--pdf", "-p", help="PDF path (optional)") parser.add_argument("--bundle", "-b", help="Path to bundle (.tar.zst), containing CSV, PDF and receipts (optional)") + parser.add_argument("-v", "--verbose", action="count", default=0, help="Logging verbosity (-v=warning, -vv=info, -vvv=debug)") args = parser.parse_args() + setup_logging(args.verbose) + csv_path = os.path.abspath(args.csv) base_dir = os.path.dirname(csv_path) or "." @@ -396,13 +425,14 @@ def main(): df = parse_csv(csv_path) if df["date"].isna().any(): bad = df[df["date"].isna()][CSV_COLUMNS] - raise ValueError(f"Ungültige Datumsangaben in folgenden Zeilen:\n{bad}") + raise ValueError(f"Invalid dates:\n{bad}") want_pdf = bool(args.pdf) mono_font = _pick_mono_font(size=8) modules: Dict[str, Module] = { "general": GeneralModule(), + "drug": DrugModule(), } rows_for_module: Dict[str, List[int]] = {} @@ -419,7 +449,7 @@ def main(): continue mod = modules.get(mod_name) if not mod: - print(f"[INFO] Unbekanntes Modul '{mod_name}' – ignoriert (noch nicht registriert).") + logger.warning("Unknown module {} - ignoring".format(mod_name)) continue subdf = df.loc[indices].copy() results.append(mod.process(subdf, context={"base_dir": base_dir, "want_pdf": want_pdf, "mono_font": mono_font})) @@ -439,11 +469,11 @@ def main(): module_pages.extend(r.pages) create_pdf(df, module_frames, module_bigframes, module_pages, args.pdf, mono_font, base_dir=base_dir, title=title) - print(f"[OK] PDF geschrieben: {args.pdf}") + logger.info("PDF written to {}".format(args.pdf)) if args.bundle: create_bundle(args.bundle, csv_path, df, base_dir=base_dir, pdf_path=args.pdf if args.pdf else None) - print(f"[OK] Bundle geschrieben: {args.bundle}") + logger.info("Bundle written to {}".format(args.bundle)) if __name__ == "__main__": main() |
