$ kode

← alle Dateien · als Text

kode/oberflaeche.py

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091"""Terminal-Oberflaeche von kode. Eingabe mit prompt_toolkit (Verlauf, mehrzeilig mit Alt+Enter, Shift+Tab schaltet den Freigabemodus), Ausgabe mit rich. Die Antwort des Modells laeuft roh ins Terminal, waehrend sie entsteht - erst rendern hiesse warten. """ import json import os import re import subprocess import sys import threading import time from pathlib import Path from typing import Any from prompt_toolkit import PromptSession from prompt_toolkit.completion import Completer, Completion from prompt_toolkit.formatted_text import HTML from prompt_toolkit.history import FileHistory from prompt_toolkit.key_binding import KeyBindings from rich.console import Console, Group from rich.live import Live from rich.markup import escape from rich.panel import Panel from rich.syntax import Syntax from rich.text import Text from rich.table import Table from . import agent, anbieter, freigabe, karte, katalog, konfig, protokoll, werkzeuge # mosh-Clients verbinden sich hier mit `-c 256`: True Color nur, wenn das # Terminal es ausdruecklich meldet. con = Console(highlight=False, color_system=( "truecolor" if os.getenv("COLORTERM", "").lower() in ("truecolor", "24bit") else "256")) HILFE = """[bold]Befehle[/bold] /start \\[modell] bestes Angebot bis Höchstpreis mieten, bei Fehlschlag das nächste /wahl Karte selbst aus der Liste wählen und bestätigen /maxpreis \\[gross] <usd> Höchstpreis je Stunde für den automatischen Start (gross: Modelle ab 80 GB) /stopp Karte löschen (nachgeprüft) /halten <min> Karte mindestens so lange halten /leerlauf <min> Abschaltung nach so viel Ruhe (Standard 30) /status Karte, Restzeit, Kosten /kosten letzte Läufe und Guthaben /modelle Katalog /modell <nr|name> Modell wechseln /modus \\[fragen|dateien|alles] Freigabemodus (auch Shift+Tab) /freigaben \\[weg <präfix>] dauerhafte Befehlsfreigaben /neu Gespräch leeren /kompakt Verlauf zusammenfassen lassen /kontext Belegung des Kontextfensters /rueckgaengig letzte Dateiänderung zurücknehmen /cd <pfad> Arbeitsordner wechseln /weiter letzte Sitzung in diesem Ordner fortsetzen /zeigen ganzes Gespräch in less (mosh hat kein Zurückrollen) /beenden beenden (auch Strg+D) !befehl Shell-Befehl selbst ausführen, Ausgabe geht ins Gespräch [bold]Tasten[/bold] Enter senden · Alt+Enter neue Zeile · Shift+Tab Modus Strg+C: bricht eine Antwort ab · leert die Eingabe · beendet bei leerer Eingabe""" def _usd(x: float | None) -> str: if x is None: return "?" return f"{x:.4f}" if abs(x) < 1 else f"{x:.2f}" def _dauer(s: float) -> str: s = max(0, int(s)) return f"{s // 60}:{s % 60:02d}" if s < 3600 else f"{s // 3600} h {s % 3600 // 60:02d}" # ------------------------------------------------------------ Einstellungen def einstellungen() -> dict[str, Any]: try: return json.loads(konfig.EINSTELLUNGEN.read_text(encoding="utf-8")) except (OSError, ValueError): return {} def einstellung(**felder: Any) -> None: d = einstellungen() d.update(felder) konfig.EINSTELLUNGEN.write_text(json.dumps(d, indent=1), encoding="utf-8") # ------------------------------------------------------------ Ausgabe class Terminal: """Erfuellt agent.Oberflaeche.""" def __init__(self, sitzung: "agent.Sitzung", session: PromptSession | None): self.s = sitzung self.ps = session self._im_denken = False self._im_text = False def text(self, stueck: str) -> None: if self._im_denken: sys.stdout.write("\033[0m\n") self._im_denken = False if not self._im_text: sys.stdout.write("\n") self._im_text = True sys.stdout.write(stueck) sys.stdout.flush() def denken(self, stueck: str) -> None: if not self._im_denken: sys.stdout.write("\n\033[2;3m✻ ") self._im_denken = True sys.stdout.write(stueck) sys.stdout.flush() def antwort_ende(self) -> None: if self._im_denken: sys.stdout.write("\033[0m") if self._im_denken or self._im_text: sys.stdout.write("\n") sys.stdout.flush() self._im_denken = self._im_text = False def werkzeug(self, name: str, args: dict) -> None: con.print(f"[bold cyan]●[/] [bold]{name}[/] {escape(kurz_args(name, args))}") def ergebnis(self, name: str, args: dict, text: str) -> None: if name == "edit_file" and text.startswith("Geändert"): con.print(f" [green]⎿ {escape(text)}[/]") return zeilen = text.splitlines() or [""] farbe = "red" if text.startswith(("Fehler", "[ABGELEHNT]")) else "dim" zeigen = 4 if name in werkzeuge.NUR_LESEN else 10 for z in zeilen[:zeigen]: con.print(f" [{farbe}]⎿ {escape(z[:200])}[/]") if len(zeilen) > zeigen: con.print(f" [dim] … {len(zeilen) - zeigen} weitere Zeilen[/]") def hinweis(self, text: str) -> None: con.print(f"[yellow]{escape(text)}[/]") def fragen(self, name: str, args: dict, d: str | None) -> tuple[str, str]: if d is not None: if d.strip(): con.print(Syntax(d[:20000], "diff", theme="ansi_dark", background_color="default")) else: con.print("[dim] (keine inhaltliche Änderung)[/]") if name == "bash": con.print(Panel(escape(args.get("command") or ""), title="Befehl", border_style="yellow", expand=False)) praefix = freigabe.praefix_vorschlag(args.get("command") or "") frage = f"Ausführen? [j]a · [i]mmer „{praefix} …“ in diesem Projekt · [n]ein · [s]topp: " else: praefix = "" frage = "Ändern? [j]a · [i]mmer (Modus „dateien“) · [n]ein · [s]topp: " while True: try: a = (self.ps.prompt(frage) if self.ps else input(frage)).strip().lower() except (KeyboardInterrupt, EOFError): return "stopp", "" if a in ("j", "ja", "y", ""): return "ja", "" if a in ("i", "immer"): if name == "bash": try: p = self.ps.prompt("Präfix: ", default=praefix) if self.ps else praefix except (KeyboardInterrupt, EOFError): continue return "immer", p.strip() or praefix return "immer", "" if a in ("n", "nein"): try: grund = self.ps.prompt("Begründung fürs Modell (optional): ") if self.ps else "" except (KeyboardInterrupt, EOFError): grund = "" return "nein", grund.strip() if a in ("s", "stopp"): return "stopp", "" class Stumm(Terminal): """Fuer kode -p: keine Rueckfragen moeglich, alles Ungefreite wird abgelehnt.""" def fragen(self, name: str, args: dict, d: str | None) -> tuple[str, str]: con.print(f"[red] nicht freigegeben im Modus {self.s.modus}[/]") return "nein", "Nicht freigegeben (nicht interaktiv)." def kurz_args(name: str, a: dict) -> str: if name == "bash": return (a.get("command") or "").replace("\n", " ⏎ ")[:160] if name in ("read_file", "write_file", "edit_file", "list_dir"): zusatz = "" if name == "read_file" and (a.get("offset") or a.get("limit")): zusatz = f" [{a.get('offset') or 1}+{a.get('limit') or ''}]" return f"{a.get('path') or '.'}{zusatz}" if name in ("glob", "grep"): return f"{a.get('pattern')}" + (f" in {a['path']}" if a.get("path") else "") return json.dumps(a, ensure_ascii=False)[:160] # ------------------------------------------------------------ Karte # --- Startanzeige (17.10.2026: "geht das ausfuehrlicher - mit Anzeige, wann # fertig geladen, geschaetzte Zeit?"). Schritte als Liste, beim Laden der # Gewichte GB, MB/s und Restzeit, oben "fertig ca. HH:MM". Die Schaetzung # rechnet karte.hochfahren; eine wartende zweite Sitzung liest sie aus dem Zustand. PHASEN_NAME = {"mieten": "Karte mieten", "dienst": "Ollama starten", "gewichte": "Gewichte laden", "aufwaermen": "In den Grafikspeicher"} class Startanzeige: def __init__(self, kopf: str) -> None: self.kopf = kopf self.t0 = self.phase_t = time.time() self.phase: str | None = None self.dauer: dict[str, float] = {} self.text, self.prozent, self.rest_s, self.noch_s = "", None, None, None self.fremd = False def melden(self, phase: str, text: str, prozent: float | None, rest_s: float | None = None, noch_s: float | None = None, fremd: bool = False, seit: float | None = None, phase_seit: float | None = None) -> None: jetzt = time.time() if fremd: self.fremd = True self.t0 = seit or self.t0 self.phase, self.phase_t = phase, phase_seit or self.phase_t elif phase != self.phase: if self.phase: self.dauer[self.phase] = jetzt - self.phase_t self.phase, self.phase_t = phase, jetzt self.text, self.prozent, self.rest_s, self.noch_s = text, prozent, rest_s, noch_s def zusammenfassung(self) -> str: if self.phase and self.phase not in self.dauer: self.dauer[self.phase] = time.time() - self.phase_t return " · ".join(f"{PHASEN_NAME.get(ph, ph)} {_dauer(d)}" for ph, d in self.dauer.items()) def __rich__(self) -> Group: jetzt = time.time() kopf = Text.assemble((self.kopf, "bold"), (f" · seit {_dauer(jetzt - self.t0)}", "dim")) if self.noch_s is not None: fertig = time.strftime("%H:%M", time.localtime(jetzt + self.noch_s)) kopf.append(f" · fertig ca. {fertig} (noch ~{_dauer(self.noch_s)})", "cyan") zeilen: list[Any] = [kopf] i = karte.PHASEN.index(self.phase) if self.phase in karte.PHASEN else 0 for j, ph in enumerate(karte.PHASEN): name = f"{PHASEN_NAME[ph]:23}" if j < i: d = self.dauer.get(ph) zeilen.append(Text.assemble((" ✓ ", "green"), (name, "dim"), (_dauer(d) if d else "", "dim"))) elif j == i: zeilen.append(Text.assemble((" ● ", "cyan"), (name, "bold"), f"{_dauer(jetzt - self.phase_t)} ", (self.text, ""))) if self.prozent is not None: voll = int(24 * min(max(self.prozent, 0), 100) / 100) zeilen.append(Text.assemble(" ", ("━" * voll, "cyan"), ("━" * (24 - voll), "grey30"), f" {self.prozent:3.0f} %", (f" · noch {_dauer(self.rest_s)}" if self.rest_s else "", ""))) else: zeilen.append(Text.assemble((" ○ ", "dim"), (PHASEN_NAME[ph], "dim"))) return Group(*zeilen) def modelle_zeigen(aktiv: str | None = None) -> None: t = Table(show_lines=False, header_style="bold") for spalte in ("Nr", "Kennung", "Name", "GB", "Karte", "Kontext", "denkt", "Notiz"): t.add_column(spalte) for i, (k, m) in enumerate(katalog.MODELLE.items(), 1): if k == "lokal": continue stil = "bold green" if k == aktiv else ("" if m["code"] else "dim") t.add_row(str(i), k, m["name"], f"{m['gb']:.1f}", f"{m['vram']} GB", str(m["kontext"]), "ja" if m["denkt"] else "", m["notiz"], style=stil) con.print(t) con.print("[dim]Grau = kein Coding-Modell. Werkzeugvorlage bei allen geprüft; " "Kontext ist gerechnet, nicht gemessen.[/]") def _vorab(s: "agent.Sitzung") -> str | None: """Was vor jeder Suche gilt: fester Endpunkt, schon bereit, fremder Start.""" if konfig.FESTE_BASIS: return "bereit" z = karte.lesen() if z["phase"] == "bereit": if z["modell"] == s.modell_id: return "bereit" con.print(f"[yellow]Es läuft schon eine Karte mit {katalog.MODELLE.get(z['modell'], {}).get('name', z['modell'])}.[/] " f"Erst /stopp oder /modell {z['modell']}.") return "abbruch" if z["phase"] != "aus": anz = Startanzeige(f"Andere Sitzung startet {z.get('karte') or 'die Karte'}" + (f" · {_usd(z['preis'])} $/h" if z.get("preis") else "")) with Live(anz, console=con, refresh_per_second=2, transient=True): z = karte.warten_auf_fremden_start(anz.melden) return "bereit" if z["phase"] == "bereit" and z["modell"] == s.modell_id else "abbruch" return None def _gross(m: dict[str, Any] | None) -> bool: return bool(m) and m.get("vram", 0) >= konfig.GROSS_AB_VRAM def max_preis(m: dict[str, Any] | None = None) -> float: """Grenze fuer den automatischen Start; grosse Modelle haben eine eigene.""" schluessel, standard = (("max_preis_gross", konfig.MAX_PREIS_GROSS) if _gross(m) else ("max_preis", konfig.MAX_PREIS)) try: return float(einstellungen().get(schluessel, standard)) except (TypeError, ValueError): return standard def _anbieter_text(k: dict[str, Any]) -> str: return "Vast" if k["anbieter"] == "vast" else "RunPod " + k["wolke"].title() def starten(s: "agent.Sitzung") -> bool: """Automatisch: bestes Angebot bis max_preis(), bei Fehlschlag das naechste. Wunsch vom 16.10.2026 ("das beste Angebot nehmen und so lange durchprobieren, bis es geht - maximal 0,75 USD pro Stunde"), ersetzt die Preisbestaetigung. Von Hand waehlen: /wahl. Seit 17.10.2026 zaehlt beim automatischen Start nicht mehr Tempo je Dollar, sondern die geschaetzte Zeit bis bereit (Abbild + Gewichte), bei Gleichstand der Preis; Grenze 0,50 USD/h ("schnell bereit"). Weiter zur naechsten Karte nur, wenn es an der Karte lag (nicht mietbar, startet nicht, Ollama stumm, Leitung zu langsam, nicht im Grafikspeicher). Liegt es am Modell (Datei fehlt, keine Werkzeuge), hilft keine andere Karte - dann wird abgebrochen. Jede Runde fragt die Angebote neu ab; versuchte Karten fallen raus. """ vorab = _vorab(s) if vorab: return vorab == "bereit" m = s.modell grenze = max_preis(m) versucht: set[tuple[str, str, str]] = set() t0 = time.monotonic() protokoll.ereignis("auto_start", modell=s.modell_id, max_preis=grenze, max_versuche=konfig.MAX_VERSUCHE) con.print(f"[dim]Automatischer Start: bestes Angebot bis {_usd(grenze)} USD/h, " f"bis zu {konfig.MAX_VERSUCHE} Karten. Strg+C bricht ab und löscht die Karte.[/]") for versuch in range(1, konfig.MAX_VERSUCHE + 1): try: with con.status(f"Suche Karten mit mindestens {m['vram']} GB für {m['name']} …"): liste = karte.vorschlag(s.modell_id) except KeyboardInterrupt: con.print("Nichts gemietet.") return False frei = [k for k in liste if (k["anbieter"], k["id"], k["wolke"]) not in versucht] passend = sorted((k for k in frei if k["preis"] <= grenze), key=lambda k: (k["ladezeit_s"], k["preis"])) if not passend: billigste = min((k["preis"] for k in frei), default=None) grund = (f"Keine Karte mit {m['vram']} GB frei." if billigste is None else f"Keine {'weitere ' if versucht else ''}Karte bis {_usd(grenze)} USD/h " f"(billigste freie: {_usd(billigste)} USD/h). " f"/maxpreis {'gross ' if _gross(m) else ''}erhöhen oder /wahl.") con.print(f"[red]{grund}[/]") protokoll.ereignis("auto_start_ende", ausgang="keine_karte", versuche=versuch - 1, grund=grund) return False k = passend[0] con.print(f"[bold]Versuch {versuch}:[/] {k['name']} · {_anbieter_text(k)} · " f"[bold]{_usd(k['preis'])} USD/h[/] · Tempo/USD {k['wert'] or '–'} · " f"Laden ca. {_dauer(k['ladezeit_s'])}") try: anz = Startanzeige(f"{k['name']} · {_anbieter_text(k)} · {_usd(k['preis'])} $/h") with Live(anz, console=con, refresh_per_second=2, transient=True): gut, text, art = karte.hochfahren(k, s.modell_id, anz.melden, probe=True) except KeyboardInterrupt: con.print("[yellow]Abgebrochen - Karte wurde gelöscht.[/]") protokoll.ereignis("auto_start_ende", ausgang="abgebrochen", versuche=versuch) return False con.print(f" [{'green' if gut else 'red'}]{escape(text)}[/]") con.print(f" [dim]{anz.zusammenfassung()}[/]") if gut: protokoll.ereignis("auto_start_ende", ausgang="bereit", versuche=versuch, karte=k["name"], dauer_s=round(time.monotonic() - t0)) return True if art != "karte": protokoll.ereignis("auto_start_ende", ausgang=art, versuche=versuch, grund=text) if art == "modell": con.print("[red]Das liegt am Modell, nicht an der Karte - andere Karten helfen nicht. " "/modell wechseln.[/]") return False versucht.add((k["anbieter"], k["id"], k["wolke"])) con.print(f"[red]{konfig.MAX_VERSUCHE} Karten nacheinander gingen nicht - aufgehört.[/] " f"Später noch einmal oder /wahl.") protokoll.ereignis("auto_start_ende", ausgang="versuche_aus", versuche=konfig.MAX_VERSUCHE) return False def waehlen(s: "agent.Sitzung") -> bool: """Von Hand: Liste zeigen, Karte waehlen, Preis bestaetigen (/wahl).""" ausschluss: set[tuple[str, str, str]] = set() while True: ergebnis = _waehlen_einmal(s, ausschluss) if ergebnis != "nochmal": return ergebnis == "bereit" def _waehlen_einmal(s: "agent.Sitzung", ausschluss: set) -> str: vorab = _vorab(s) if vorab: return vorab m = s.modell with con.status(f"Suche Karten mit mindestens {m['vram']} GB für {m['name']} …"): liste = karte.vorschlag(s.modell_id) if not liste: con.print(f"[red]Keine Karte mit {m['vram']} GB frei.[/] Später noch einmal oder kleineres Modell.") return "abbruch" liste = [k for k in liste if (k["anbieter"], k["id"], k["wolke"]) not in ausschluss][:10] if not liste: con.print("[red]Keine weitere Karte frei.[/]") return "abbruch" t = Table(title=f"{m['name']} · {m['gb']} GB Gewichte · Kontext {m['kontext']}", header_style="bold") for spalte in ("Nr", "Anbieter", "Karte", "VRAM", "USD/h", "Lage", "Laden ca.", "Tempo/USD"): t.add_column(spalte, justify="right" if spalte in ("USD/h", "VRAM", "Tempo/USD") else "left") for i, k in enumerate(liste, 1): t.add_row(str(i), "RunPod " + k["wolke"].title() if k["anbieter"] == "runpod" else "Vast", k["name"], f"{k['gb']} GB", _usd(k["preis"]), str(k["lage"])[:18], _dauer(k["ladezeit_s"]), str(k["wert"] or "–"), style="bold" if i == 1 else "") con.print(t) con.print("[dim]Tempo/USD = Speicherbandbreite je Dollar der ersten Stunde (Ladezeit eingerechnet). " "Vast-Wirte sind Fremdrechner, RunPod Secure sind Rechenzentren.[/]") try: wahl = input("Karte wählen [1]: ").strip() or "1" except (KeyboardInterrupt, EOFError): con.print("Nichts gemietet.") return "abbruch" if not wahl.isdigit() or not 1 <= int(wahl) <= len(liste): protokoll.ereignis("start_wahl", wahl=wahl, ergebnis="abbruch") con.print("Nichts gemietet.") return "abbruch" k = liste[int(wahl) - 1] bis_bereit = k["preis"] * k["ladezeit_s"] / 3600 + k["download_usd"] leer = karte.lesen().get("leerlauf_min") or konfig.LEERLAUF_MIN con.print(f"\n[bold]{k['name']}[/] bei {'Vast.ai' if k['anbieter'] == 'vast' else 'RunPod ' + k['wolke'].title()}: " f"[bold]{_usd(k['preis'])} USD/h[/], bis bereit ca. {_usd(bis_bereit)} USD " f"({_dauer(k['ladezeit_s'])} min geschätzt). Abschaltung nach {leer} min Ruhe.") try: ok = input("Mieten? [j/N]: ").strip().lower() in ("j", "ja", "y") except (KeyboardInterrupt, EOFError): ok = False protokoll.ereignis("start_wahl", wahl=wahl, bestaetigt=ok, karte=k["name"], anbieter=k["anbieter"], wolke=k["wolke"], preis=k["preis"], ladezeit_s=k["ladezeit_s"]) if not ok: con.print("Nichts gemietet. [dim](Gemietet wird nur mit „j“.)[/]") return "abbruch" con.print("[dim]Strg+C bricht ab und löscht die Karte wieder.[/]") try: anz = Startanzeige(f"{k['name']} · {_anbieter_text(k)} · {_usd(k['preis'])} $/h") with Live(anz, console=con, refresh_per_second=2, transient=True): gut, text, art = karte.hochfahren(k, s.modell_id, anz.melden) except KeyboardInterrupt: con.print("[yellow]Abgebrochen - Karte wurde gelöscht.[/]") return "abbruch" con.print(f"[{'green' if gut else 'red'}]{escape(text)}[/]") con.print(f"[dim]{anz.zusammenfassung()}[/]") if not gut and art == "karte" and "nicht mietbar" in text: ausschluss.add((k["anbieter"], k["id"], k["wolke"])) con.print("[yellow]Neue Liste ohne diese Karte:[/]") return "nochmal" return "bereit" if gut else "abbruch" def stoppen(grund: str = "von Hand beendet") -> None: if konfig.FESTE_BASIS: con.print("Fester Endpunkt - nichts zu löschen.") return with con.status("Karte wird gelöscht und nachgeprüft …"): ok, text = karte.herunterfahren(grund) con.print(f"[{'green' if ok else 'bold red'}]{escape(text)}[/]") _guthaben["zeit"] = 0.0 def status_zeigen() -> None: if konfig.FESTE_BASIS: con.print(f"Fester Endpunkt {konfig.FESTE_BASIS}") return z = karte.lesen() if not z.get("kennung"): con.print(f"Keine Karte. ({z['phase']})") return f = karte.frist(z) con.print(f"[bold]{z['karte']}[/] ({z['anbieter']} {z['kennung']}) · {z['phase']} · " f"{katalog.MODELLE.get(z['modell'], {}).get('name', z['modell'])}\n" f"{_usd(z['preis'])} USD/h · läuft {_dauer(time.time() - z['seit'])} · " f"bisher {_usd(karte.kosten_bisher(z))} USD · Abschaltung in " f"{_dauer((f or time.time()) - time.time())} (Leerlauf {z['leerlauf_min']} min" + (f", gehalten bis {time.strftime('%H:%M', time.localtime(z['halten_bis']))}" if z.get("halten_bis") and z["halten_bis"] > time.time() else "") + ")") def kosten_zeigen() -> None: t = Table(header_style="bold") for spalte in ("Beginn", "Karte", "Modell", "Dauer", "USD", "Grund"): t.add_column(spalte) for r in karte.laeufe(): dauer = (r["ende"] or time.time()) - r["beginn"] t.add_row(time.strftime("%d.%m. %H:%M", time.localtime(r["beginn"])), r["karte"] or "", r["modell"] or "", _dauer(dauer), _usd(r["kosten"]) if r["kosten"] is not None else "läuft", (r["grund"] or "") + ("" if r["geloescht"] in (1, None) else " [NICHT gelöscht]")) con.print(t) heute = time.mktime(time.strptime(time.strftime("%Y-%m-%d"), "%Y-%m-%d")) monat = time.mktime(time.strptime(time.strftime("%Y-%m-01"), "%Y-%m-%d")) with con.status("Guthaben abfragen …"): g = karte.guthaben() con.print(f"Heute {_usd(karte.kosten_summe(heute))} USD · Monat {_usd(karte.kosten_summe(monat))} USD · " f"Guthaben RunPod {_usd(g['runpod'])} · Vast {_usd(g['vast'])} USD") # ------------------------------------------------------------ Sitzungen def _sitzungsdatei(s: "agent.Sitzung") -> Path: return konfig.SITZUNGEN / f"{s.kennung}.json" def sitzung_sichern(s: "agent.Sitzung") -> None: if not s.verlauf: return _sitzungsdatei(s).write_text(json.dumps({ "cwd": str(s.projekt), "modell": s.modell_id, "zeit": time.time(), "verlauf": s.verlauf}, ensure_ascii=False), encoding="utf-8") def letzte_sitzung(projekt: Path, ausser: str) -> dict | None: beste = None for p in konfig.SITZUNGEN.glob("*.json"): if p.stem == ausser: continue try: d = json.loads(p.read_text(encoding="utf-8")) except (OSError, ValueError): continue if d.get("cwd") == str(projekt) and (not beste or d["zeit"] > beste["zeit"]): beste = d return beste def _andere_sitzungen_aktiv() -> bool: z = karte.lesen() lebend = [] for pid in z.get("sitzungen") or []: try: os.kill(pid, 0) if pid != os.getpid(): lebend.append(pid) except OSError: pass return bool(lebend) def _anmelden(an: bool) -> None: z = karte.lesen() liste = [p for p in (z.get("sitzungen") or []) if p != os.getpid()] if an: liste.append(os.getpid()) karte.aendern(sitzungen=liste) def beenden_fragen() -> None: if konfig.FESTE_BASIS: return z = karte.lesen() if not z.get("kennung"): return if _andere_sitzungen_aktiv(): con.print("[dim]Andere kode-Sitzung nutzt die Karte noch - sie bleibt.[/]") return leer = z.get("leerlauf_min") or konfig.LEERLAUF_MIN rest = (karte.frist(z) or time.time()) - time.time() con.print(f"\n[bold]{z['karte']}[/] läuft noch ({_usd(z['preis'])} USD/h, bisher " f"{_usd(karte.kosten_bisher(z))} USD, Abschaltung in {_dauer(rest)}).") try: a = input(f"[l]öschen · [h]alten <Minuten> · Enter/Strg+C = Leerlauf-Uhr ({leer} min) läuft weiter: ").strip().lower() except (KeyboardInterrupt, EOFError): a = "" protokoll.ereignis("beenden_wahl", antwort=a, karte=z.get("karte"), kosten=karte.kosten_bisher(z)) if a.startswith("l"): stoppen("beim Beenden gelöscht") elif a.startswith("h"): try: minuten = int(a[1:].strip() or input("Minuten: ")) except (ValueError, KeyboardInterrupt, EOFError): minuten = leer karte.aendern(halten_bis=time.time() + 60 * minuten) con.print(f"Karte bleibt bis {time.strftime('%H:%M', time.localtime(time.time() + 60 * minuten))}, " f"danach löscht der Wächter sie.") else: con.print(f"\nDer Wächter löscht die Karte nach {leer} min Ruhe.") # ------------------------------------------------------------ Schleife # Guthaben fuer die Statuszeile: im Hintergrund, hoechstens einmal je Minute. # Die Leiste wird alle 5 s neu gezeichnet - ein Abruf dort liesse jede # Eingabe auf RunPod und Vast warten. GUTHABEN_TAKT = 60 _guthaben: dict[str, Any] = {"werte": None, "zeit": 0.0, "laeuft": False} def _guthaben_holen() -> None: try: _guthaben["werte"] = karte.guthaben() except Exception: # noqa: BLE001 pass finally: _guthaben["zeit"] = time.time() _guthaben["laeuft"] = False def guthaben_text() -> str: if (time.time() - _guthaben["zeit"] > GUTHABEN_TAKT) and not _guthaben["laeuft"]: _guthaben["laeuft"] = True threading.Thread(target=_guthaben_holen, daemon=True).start() g = _guthaben["werte"] if not g: return "Guthaben …" teile = [] if konfig.RUNPOD_API_KEY: teile.append(f"RunPod {_usd(g.get('runpod'))} $") if konfig.VAST_API_KEY: teile.append(f"Vast {_usd(g.get('vast'))} $") return " · ".join(teile) MODUS_FARBE = {"fragen": "green", "dateien": "yellow", "alles": "red"} def _teile_karte(s: "agent.Sitzung") -> list[tuple[str, str | None]]: """Erste Zeile: Karte, Modell, Modus, Tempo - das, was sich waehrend der Arbeit aendert.""" if konfig.FESTE_BASIS: teile: list[tuple[str, str | None]] = [("fest", None)] else: z = karte.lesen() f = z.get("fortschritt") or {} if z.get("kennung") and z["phase"] != "bereit" and f.get("fertig_um"): teile = [(f"{z['karte']} {PHASEN_NAME.get(z['phase'], z['phase'])}" + (f" {f['prozent']:.0f} %" if f.get("prozent") is not None else "") + f" · fertig ca. {time.strftime('%H:%M', time.localtime(f['fertig_um']))}" f" · {_usd(z['preis'])} $/h", None)] elif z.get("kennung"): rest = (karte.frist(z) or time.time()) - time.time() teile = [(f"{z['karte']} {z['phase']} · {_usd(z['preis'])} $/h · " f"{_usd(karte.kosten_bisher(z))} $ · aus in {_dauer(rest)}", None)] else: teile = [("keine Karte", None)] teile.append((s.modell["name"], "bold")) teile.append((f"Modus {s.modus}", MODUS_FARBE[s.modus])) if s.stat.tok_s: teile.append((f"{s.stat.tok_s:.0f} tok/s", None)) return teile def _teile_konto(s: "agent.Sitzung") -> list[tuple[str, str | None]]: """Zweite Zeile: Guthaben (zuerst, das soll beim Scrollen sichtbar bleiben) und Kontext.""" teile: list[tuple[str, str | None]] = [] if konfig.FESTE_BASIS else [(guthaben_text(), "cyan")] teile.append((f"Kontext {s.belegung():.0%}", None)) return teile def _leiste_teile(s: "agent.Sitzung") -> list[tuple[str, str | None]]: """Inhalt der einzeiligen Statuszeile von prompt_toolkit (ohne tmux).""" return _teile_karte(s) + _teile_konto(s) def _leiste(s: "agent.Sitzung"): def leiste() -> HTML: html = [] for text, farbe in _leiste_teile(s): t = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;") if farbe == "bold": t = f"<b>{t}</b>" elif farbe: t = f"<ansi{farbe}>{t}</ansi{farbe}>" html.append(t) return HTML(" " + " │ ".join(html)) return leiste # --- Statuszeile in tmux (seit 16.10.2026: "wenn gescrollt wird, ist das # Guthaben nicht mehr zu sehen"). Die Leiste von prompt_toolkit ist Teil des # Bildschirms und rollt beim Scrollen mit weg; die tmux-Leiste bleibt stehen. # Seit 17.10.2026 ("die Ansicht von Guthaben und Status ist etwas komisch"): # zwei Zeilen unten statt einer oben, Karte zuerst, gekuerzt an Feldgrenzen # statt mitten im Wort. Ein Faden schreibt je Zeile eine Datei, die tmux per # #(cat) liest, und stoesst bei jeder Aenderung refresh-client -S an - sonst # hinkte die Leiste bis zu 5 s hinterher. LEISTE_TAKT = 1 TMUX_FARBE = {"green": "colour114", "yellow": "colour221", "red": "colour203", "cyan": "colour117"} # Rechts in Zeile 2 steht im Scroll-Modus ein Hinweis (status-format[1] in # __main__), aber erst ab LEISTE_HINWEIS_AB Spalten - darunter geht das Guthaben vor. LEISTE_HINWEIS_BREITE = 22 LEISTE_HINWEIS_AB = 80 def _leiste_datei() -> Path | None: name = os.getenv("KODE_TMUX_SITZUNG") return konfig.DATEN / "leiste" / name if name and os.getenv("TMUX") else None def _tmux_zeile(teile: list[tuple[str, str | None]], breite: int) -> str: """Ganze Felder, solange sie passen, dann " …"; ein zu langes erstes Feld wird gekuerzt.""" platz = breite - 1 stuecke: list[str] = [] for text, farbe in teile: trenner = " │ " if stuecke else "" if len(trenner) + len(text) > platz: if stuecke: if platz >= 2: stuecke.append(" …") break text = text[:max(platz - 1, 0)] + "…" platz -= len(trenner) + len(text) t = text.replace("#", "##") if farbe == "bold": t = f"#[bold]{t}#[nobold]" elif farbe: t = f"#[fg={TMUX_FARBE[farbe]}]{t}#[fg=default]" stuecke.append(trenner + t) return " " + "".join(stuecke) def _tmux_leiste_schreiben(s: "agent.Sitzung", datei: Path) -> bool: """Schreibt beide Zeilen; True, wenn sich etwas geaendert hat.""" try: breite = os.get_terminal_size(sys.__stdout__.fileno()).columns except OSError: breite = 120 neu = False for nr, (teile, rand) in enumerate(((_teile_karte(s), 0), (_teile_konto(s), LEISTE_HINWEIS_BREITE if breite >= LEISTE_HINWEIS_AB else 0)), 1): inhalt = _tmux_zeile(teile, max(breite - rand, 20)) ziel = datei.with_name(f"{datei.name}.{nr}") try: if ziel.read_text(encoding="utf-8") == inhalt: continue except OSError: pass tmp = ziel.with_name(ziel.name + ".tmp") tmp.write_text(inhalt, encoding="utf-8") tmp.replace(ziel) neu = True return neu def _tmux_leiste_starten(s: "agent.Sitzung") -> Path | None: datei = _leiste_datei() if not datei: return None datei.parent.mkdir(parents=True, exist_ok=True) def lauf() -> None: while True: try: if _tmux_leiste_schreiben(s, datei): subprocess.run(["tmux", "refresh-client", "-S"], capture_output=True, timeout=5) except Exception: # noqa: BLE001 pass time.sleep(LEISTE_TAKT) threading.Thread(target=lauf, daemon=True).start() return datei def befehl(s: "agent.Sitzung", ui: Terminal, zeile: str) -> bool: """Schraegstrich-Befehl. False = beenden.""" teile = zeile[1:].split(maxsplit=1) name, arg = (teile[0].lower() if teile else ""), (teile[1].strip() if len(teile) > 1 else "") protokoll.ereignis("befehl", name=name, arg=arg) if name in ("beenden", "exit", "quit", "q"): return False if name in ("hilfe", "help", "?"): con.print(HILFE) elif name == "start": if arg: mid = katalog.finden(arg) if not mid: con.print("Unbekanntes Modell - /modelle") return True s.modell_wechseln(mid) einstellung(modell=mid) starten(s) elif name == "wahl": waehlen(s) elif name == "maxpreis": teile = arg.split() gross = bool(teile) and teile[0].lower() in ("gross", "groß") if gross: teile = teile[1:] try: wert = float(teile[0].replace(",", ".")) except (IndexError, ValueError): con.print(f"Höchstpreis: {_usd(max_preis())} USD/h, große Modelle (ab " f"{konfig.GROSS_AB_VRAM} GB): {_usd(max_preis({'vram': konfig.GROSS_AB_VRAM}))} USD/h" f" · ändern: /maxpreis 0.5 oder /maxpreis gross 1.5") return True if gross: einstellung(max_preis_gross=wert) con.print(f"Automatischer Start nimmt für große Modelle Karten bis {_usd(wert)} USD/h.") else: einstellung(max_preis=wert) con.print(f"Automatischer Start nimmt Karten bis {_usd(wert)} USD/h.") elif name in ("stopp", "stop"): stoppen() elif name == "halten": try: minuten = int(arg) except ValueError: con.print("/halten <Minuten>") return True karte.aendern(halten_bis=time.time() + 60 * minuten) status_zeigen() elif name == "leerlauf": try: minuten = max(5, min(480, int(arg))) except ValueError: con.print("/leerlauf <Minuten>") return True karte.aendern(leerlauf_min=minuten) einstellung(leerlauf=minuten) con.print(f"Abschaltung nach {minuten} min Ruhe.") elif name in ("status", "karte"): status_zeigen() elif name == "kosten": kosten_zeigen() elif name == "modelle": modelle_zeigen(s.modell_id) elif name == "modell": if not arg: modelle_zeigen(s.modell_id) return True mid = katalog.finden(arg) if not mid: con.print("Unbekanntes Modell - /modelle zeigt Nummern und Kennungen.") return True z = karte.lesen() if z.get("kennung") and z.get("modell") != mid and not konfig.FESTE_BASIS: try: ok = input(f"Karte läuft mit {katalog.MODELLE.get(z['modell'], {}).get('name')}. " f"Löschen und {katalog.MODELLE[mid]['name']} neu starten? [j/N]: ").strip().lower() == "j" except (KeyboardInterrupt, EOFError): ok = False if not ok: return True stoppen("Modellwechsel") s.modell_wechseln(mid) einstellung(modell=mid) starten(s) else: s.modell_wechseln(mid) einstellung(modell=mid) con.print(f"Modell: {s.modell['name']} (Kontext {s.kontext})") elif name == "modus": if arg in freigabe.MODI: s.modus = arg else: s.modus = freigabe.MODI[(freigabe.MODI.index(s.modus) + 1) % 3] con.print(f"Modus {s.modus}: {freigabe.MODUS_TEXT[s.modus]}") elif name == "freigaben": if arg.startswith("weg "): p = arg[4:].strip() con.print("entfernt" if freigabe.regel_weg(s.projekt, p) else "gab es nicht") r = freigabe.regeln(s.projekt) con.print(f"Freigaben für {s.projekt}: " + (", ".join(f"„{x} …“" for x in r) if r else "keine")) elif name == "neu": sitzung_sichern(s) s.verlauf.clear() s.stat = agent.Statistik() con.print("Gespräch geleert.") elif name == "kompakt": z = karte.bereit() if z: s.verdichten(ui, z["basis"]) else: con.print("Keine Karte bereit.") elif name == "kontext": con.print(f"{s.belegung():.0%} von {s.kontext} Token (geschätzt {s.schaetzung()}, " f"zuletzt gemessen {s.stat.letzter_kontext}) · {len(s.verlauf)} Nachrichten") elif name in ("rueckgaengig", "rückgängig", "undo"): con.print(s.umgebung.rueckgaengig()) elif name == "cd": ziel = s.umgebung.pfad(arg or str(Path.home())) if ziel.is_dir(): s.umgebung.cwd = ziel.resolve() s.projekt = s.umgebung.cwd con.print(f"Arbeitsordner: {s.projekt}") else: con.print(f"Kein Ordner: {ziel}") elif name == "zeigen": zeigen(s) elif name == "weiter": d = letzte_sitzung(s.projekt, s.kennung) if not d: con.print("Keine frühere Sitzung in diesem Ordner.") else: s.verlauf = d["verlauf"] con.print(f"Sitzung vom {time.strftime('%d.%m. %H:%M', time.localtime(d['zeit']))} " f"geladen ({len(s.verlauf)} Nachrichten).") else: con.print(f"Unbekannt: /{name} - /hilfe") return True def zeigen(s: "agent.Sitzung") -> None: """Ganzes Gespraech im Pager - mosh hat keinen Scrollback.""" zeilen: list[str] = [] for m in s.verlauf: if m["role"] == "user": zeilen += ["", "\033[1;36m› " + (m.get("content") or "") + "\033[0m"] elif m["role"] == "assistant": if m.get("content"): zeilen += ["", m["content"]] for t in m.get("tool_calls") or []: f = t.get("function", {}) zeilen.append(f"\033[1;36m●\033[0m \033[1m{f.get('name')}\033[0m " f"{kurz_args(f.get('name') or '', f.get('arguments') or {})}") elif m["role"] == "tool": for z in (m.get("content") or "").splitlines()[:40]: zeilen.append(f"\033[2m ⎿ {z}\033[0m") if not zeilen: con.print("Noch kein Gespräch.") return try: subprocess.run(["less", "-R", "+G"], input="\n".join(zeilen).encode(), check=False) except (OSError, KeyboardInterrupt): pass def shell(s: "agent.Sitzung", ui: Terminal, befehl_text: str) -> None: ausgabe = werkzeuge.ausfuehren(s.umgebung, "bash", {"command": befehl_text}) protokoll.ereignis("eigener_befehl", befehl=befehl_text, ergebnis=ausgabe) con.print(escape(ausgabe)) s.verlauf.append({"role": "user", "content": f"[Ich habe selbst ausgeführt: `{befehl_text}`]\nAusgabe:\n{ausgabe}"}) def begruessung(s: "agent.Sitzung") -> None: con.print(Panel.fit( f"[bold]kode[/] · {escape(str(s.projekt))}\n" f"Modell [bold]{s.modell['name']}[/] · Modus {s.modus} · /hilfe", border_style="cyan")) if konfig.FESTE_BASIS: con.print(f"[dim]Fester Endpunkt {konfig.FESTE_BASIS} - es wird nichts gemietet.[/]") return z = karte.lesen() if z["phase"] == "bereit": status_zeigen() elif z["phase"] == "aus": con.print(f"[dim]Keine Karte. Die erste Nachricht mietet automatisch das beste Angebot bis " f"{_usd(max_preis(s.modell))} USD/h (/maxpreis, /wahl zum Selbstwählen).[/]") # --- Vorschlaege nach "/" (17.10.2026: "wenn ich ein / schicke, kommen dann # Vorschlaege?"). Die Befehle stammen aus HILFE - eine Quelle, damit Hilfe # und Liste nicht auseinanderlaufen. def _befehle_aus_hilfe() -> list[tuple[str, str, str]]: """(Name, Argumente, Beschreibung) je Zeile " /name [arg] Text" der HILFE.""" aus = [] for zeile in HILFE.splitlines(): t = re.match(r"^ /(\S+)((?: \S+)*?)\s{2,}(.+)$", zeile) if t: aus.append((t[1], t[2].replace("\\", "").strip(), t[3].strip())) return aus class BefehlVorschlaege(Completer): def __init__(self) -> None: self.befehle = _befehle_aus_hilfe() def get_completions(self, document, complete_event): # noqa: ANN001, ANN201 text = document.text_before_cursor if not text.startswith("/") or "\n" in text: return if " " not in text: wort = text[1:].lower() for name, args, beschreibung in self.befehle: if name.startswith(wort): yield Completion("/" + name, start_position=-len(text), display=f"/{name} {args}".rstrip(), display_meta=beschreibung) return name, _, rest = text[1:].partition(" ") werte: list[tuple[str, str]] = [] if name == "modus": werte = [(m, "") for m in freigabe.MODI] elif name in ("modell", "start"): werte = [(k, m["name"]) for k, m in katalog.MODELLE.items()] for wert, beschreibung in werte: if wert.startswith(rest.lower()) or (beschreibung and rest.lower() in beschreibung.lower()): yield Completion(wert, start_position=-len(rest), display_meta=beschreibung) def schleife(s: "agent.Sitzung") -> None: konfig.verzeichnisse() tasten = KeyBindings() @tasten.add("s-tab") def _(ereignis) -> None: # noqa: ANN001 s.modus = freigabe.MODI[(freigabe.MODI.index(s.modus) + 1) % 3] ereignis.app.invalidate() @tasten.add("c-c") def _(ereignis) -> None: # noqa: ANN001 # Text in der Zeile: leeren. Leere Zeile: beenden (wie Strg+D). puffer = ereignis.current_buffer if puffer.text: puffer.reset() else: ereignis.app.exit(exception=EOFError()) @tasten.add("escape", "enter") def _(ereignis) -> None: # noqa: ANN001 ereignis.current_buffer.insert_text("\n") ps: PromptSession = PromptSession(history=FileHistory(str(konfig.EINGABE_VERLAUF)), key_bindings=tasten, completer=BefehlVorschlaege(), complete_while_typing=True, # unter tmux steht der Status fest oben (rollt nicht weg) bottom_toolbar=None if _leiste_datei() else _leiste(s), refresh_interval=5) ui = Terminal(s, ps) leiste_datei = _tmux_leiste_starten(s) protokoll.sitzung_beginnen(s.kennung, cwd=str(s.projekt), modell=s.modell_id, modus=s.modus, interaktiv=True, karte=karte.lesen() if not konfig.FESTE_BASIS else "fest") begruessung(s) if not konfig.FESTE_BASIS: _anmelden(True) try: while True: try: zeile = ps.prompt(HTML("<b><ansicyan>›</ansicyan></b> ")).strip() except KeyboardInterrupt: continue except EOFError: break if not zeile: continue if zeile.startswith("/"): try: if not befehl(s, ui, zeile): break except KeyboardInterrupt: ui.antwort_ende() ui.hinweis("Abgebrochen.") continue if zeile.startswith("!"): try: shell(s, ui, zeile[1:].strip()) except KeyboardInterrupt: ui.hinweis("Abgebrochen.") continue if not karte.bereit() or (not konfig.FESTE_BASIS and karte.lesen().get("modell") != s.modell_id): try: if not starten(s): continue except KeyboardInterrupt: con.print("Nichts gemietet.") continue try: s.frage(zeile, ui) except KeyboardInterrupt: ui.antwort_ende() ui.hinweis("Abgebrochen.") except Exception as f: # noqa: BLE001 ui.antwort_ende() protokoll.fehler("auftrag", f, eingabe=zeile) con.print(f"[red]Fehler: {escape(str(f))}[/]") if not konfig.FESTE_BASIS and karte.lesen().get("kennung"): z = karte.lesen() if not anbieter.existiert(z["anbieter"], z["kennung"]): con.print("[red]Die Karte gibt es nicht mehr.[/]") karte.aendern(**karte.LEER) sitzung_sichern(s) finally: sitzung_sichern(s) if not konfig.FESTE_BASIS: _anmelden(False) beenden_fragen() if leiste_datei: leiste_datei.unlink(missing_ok=True) protokoll.ereignis("sitzung_ende", nachrichten=len(s.verlauf), sicherungen=len(s.umgebung.sicherungen))