#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ csv2trac.py -- erzeugt aus einer Zellen-CSV eine Trac-Wiki-Tabelle. Aufruf: ./csv2trac.py zellen-gross-rietz.csv ./csv2trac.py zellen.csv -o tabelle.trac ./csv2trac.py zellen.csv --config meine-spalten.json ./csv2trac.py zellen.csv --dump-config > meine-spalten.json Eingabe ist die CSV, die die Android-Scanner-App zu den gescannten Zell-QR-Codes ausgibt. Ausgabe ist eine fertige Trac-Wiki-Tabelle plus Vorspannzeile mit den ueber alle Zellen konstanten Angaben. Anpassung an ein geaendertes CSV-Format --------------------------------------- Die Zuordnung CSV-Spalte -> logische Rolle steht in DEFAULT_CONFIG["columns"] und laesst sich per JSON-Datei (--config) teilweise oder ganz ueberschreiben. Die JSON-Datei wird flach in die Vorgabe hineingemischt, d.h. es genuegt, nur die abweichenden Schluessel anzugeben, z.B.: { "columns": { "code": "qr_code", "date": "datum" } } Wenn die App irgendwann andere Spaltenkoepfe oder zusaetzliche Felder liefert, reicht normalerweise ein Eintrag in "columns" -- am Code ist dann nichts zu aendern. Fehlt eine konfigurierte Spalte in der CSV, bricht das Script mit Klartextmeldung ab, statt stillschweigend Leerfelder zu schreiben. --dump-config gibt die komplette Vorgabe-Konfiguration als Startpunkt fuer eine eigene JSON-Datei aus. Messwerte und Bemerkungen ------------------------- Zellbezeichnung und Messwerte stehen gemeinsam im Kommentarfeld der CSV, getrennt durch labelSeparator (";"). Innenwiderstand und Spannung werden per Regex herausgeschnitten und normiert; alles, was danach uebrig bleibt, wandert als Freitext in die Bemerkungsspalte, mehrere ";"-Abschnitte mit remarkJoin ("[[br]]") verbunden. Ein nicht interpretierbarer Messwert wird als "?" dargestellt, der zugehoerige Text geht vollstaendig in die Bemerkungen -- es geht also keine Information verloren, auch wenn die Erfassung mal von der gewohnten Schreibweise abweicht. Warnungen --------- Auffaelligkeiten (nicht erkannte Messwerte, unlesbares Datum, im Vorspann nicht konstante Felder) meldet das Script auf stderr, nicht in der Tabelle. Beim Umleiten der Ausgabe mit -o bleibt die Tabelle also sauber, die Warnungen erscheinen weiterhin im Terminal. Wenn in einer erzeugten Tabelle Fragezeichen auftauchen, steht die Ursache dort. """ import argparse import copy import csv import json import re import sys from datetime import datetime DEFAULT_CONFIG = { # ---- Eingabe ------------------------------------------------------- "inputEncoding": "utf-8-sig", "inputDelimiter": ",", # Logische Rolle -> Spaltenkopf in der CSV. # Rolle auf null setzen, wenn die Spalte in der CSV fehlt. "columns": { "label": "note", "code": "original_qr_content", "date": "production_date", "daySeq": "number_of_cell_produced_that_day", "manufacturer": "manufacturers", "productType": "product_types", "cellType": "cell_types", "modelCode": "model_codes", "capacityAh": "capacity_ah", "energyWh": "energy_wh", "city": "production_city", "productLine": "factory_product_line", "taskOrder": "factory_task_order", }, # ---- Aufbereitung -------------------------------------------------- # Die Zellbezeichnung steht vor diesem Trennzeichen, die Messwerte dahinter. "labelSeparator": ";", # Quelle der Messwerte: "labelRemainder" = Rest hinter labelSeparator, # sonst der Name einer CSV-Spalte. "measurementSource": "labelRemainder", "dateInputFormats": ["%d/%m/%Y", "%Y-%m-%d", "%d.%m.%Y", "%d/%b/%Y"], "dateOutputFormat": "%Y-%m-%d", "decimalPoint": ".", "resistanceDecimals": 2, "voltageDecimals": 3, # Rollen, deren Rohwert als Zahl geglaettet wird (628.0 -> 628) "numericRoles": ["capacityAh", "energyWh"], # Zeilen nach der Zahl am Ende der Bezeichnung sortieren "sortByTrailingNumber": True, # Begriffe, die ausserhalb von Backticks mit "!" gegen WikiWord- # Verlinkung geschuetzt werden. Leer, da Trac weder LiFePO4 noch MB56 # als WikiWord auffasst; bei Bedarf hier eintragen. "escapeTerms": [], # Verbinder zwischen mehreren Freitext-Abschnitten der Bemerkungsspalte "remarkJoin": "[[br]]", # ---- Ausgabe ------------------------------------------------------- "emitPreamble": True, "preambleTemplate": ( "Alle Zellen: {manufacturer}, {cellType} Prismazelle {modelCode}, " "{capacityAh} Ah / {energyWh} Wh, Werk {city}, " "Fertigungslinie {productLine}, Fertigungsauftrag {taskOrder}." ), # Rollen, die im Vorspann stehen; sie muessen ueber alle Zeilen # konstant sein, sonst gibt es eine Warnung auf stderr. "preambleRoles": [ "manufacturer", "cellType", "modelCode", "capacityAh", "energyWh", "city", "productLine", "taskOrder", ], "headerRow": ( "||= !Batterie/Zelle =||=`Code` [[br]] dekodiert " "=||= Innenwiderstand, Spannung =||= Bemerkungen =||" ), "cellTemplates": [ "{label}", "`{code}`[[br]]Production Date: {date}" "[[br]]lfd. Nr. am Produktionstag: {daySeq}", "{resistance} mR, {voltage} V", "{remark}", ], # Vorbelegung der Bemerkungen-Spalte, falls im Freitext nichts # ueber die Messwerte hinaus steht "defaultRemark": "", } RESISTANCE_RE = re.compile(r"([0-9]+(?:[.,][0-9]+)?)\s*m\s*(?:ohm|\u03a9|R)\b", re.I) VOLTAGE_RE = re.compile(r"([0-9]+(?:[.,][0-9]+)?)\s*V\b", re.I) TRAILING_NUMBER_RE = re.compile(r"([0-9]+)\s*$") def loadConfig(path): config = copy.deepcopy(DEFAULT_CONFIG) if not path: return config with open(path, encoding="utf-8") as stream: overrides = json.load(stream) for key, value in overrides.items(): if isinstance(value, dict) and isinstance(config.get(key), dict): config[key].update(value) else: config[key] = value return config def readRows(path, config): with open(path, encoding=config["inputEncoding"], newline="") as stream: reader = csv.DictReader(stream, delimiter=config["inputDelimiter"]) headers = reader.fieldnames or [] wanted = [name for name in config["columns"].values() if name] missing = sorted({name for name in wanted if name not in headers}) if missing: raise SystemExit("Fehlende Spalten in der CSV: " + ", ".join(missing)) return [row for row in reader if any((value or "").strip() for value in row.values())] def fieldValue(row, config, role): column = config["columns"].get(role) if not column: return "" return (row.get(column) or "").strip() def formatNumber(rawValue, decimals, decimalPoint): text = str(rawValue).strip().replace(",", ".") try: number = float(text) except ValueError: return str(rawValue).strip() text = ("%g" % number) if decimals is None else ("%.*f" % (decimals, number)) return text.replace(".", decimalPoint) if decimalPoint != "." else text def formatDate(rawValue, config): text = rawValue.strip() if not text: return "" for pattern in config["dateInputFormats"]: try: return datetime.strptime(text, pattern).strftime(config["dateOutputFormat"]) except ValueError: continue warn("Datum nicht interpretierbar, uebernehme unveraendert: %r" % text) return text def parseMeasurements(text, config, label): """Liefert (Innenwiderstand, Spannung, uebriger Freitext). Alles, was nicht als Messwert erkannt wird, bleibt als Freitext uebrig und landet spaeter in der Bemerkungsspalte -- auch dann, wenn ein Messwert gar nicht interpretierbar war. """ resistanceMatch = RESISTANCE_RE.search(text) voltageMatch = VOLTAGE_RE.search(text) if not resistanceMatch: warn("%s: kein Innenwiderstand in %r gefunden, Text geht in die " "Bemerkungen" % (label or "?", text)) if not voltageMatch: warn("%s: keine Spannung in %r gefunden, Text geht in die " "Bemerkungen" % (label or "?", text)) resistance = formatNumber(resistanceMatch.group(1), config["resistanceDecimals"], config["decimalPoint"]) if resistanceMatch else "?" voltage = formatNumber(voltageMatch.group(1), config["voltageDecimals"], config["decimalPoint"]) if voltageMatch else "?" remainder = text for match in sorted([m for m in (resistanceMatch, voltageMatch) if m], key=lambda m: m.start(), reverse=True): remainder = remainder[:match.start()] + " " + remainder[match.end():] return resistance, voltage, remainder def tidyRemark(text, config): """Raeumt den uebrig gebliebenen Freitext fuer die Bemerkungsspalte auf.""" parts = [] for part in text.split(";"): part = re.sub(r"\s+", " ", part) part = part.strip(" \t,;") if part: parts.append(part) return config["remarkJoin"].join(parts) def escapeWikiTerms(text, terms): if not terms: return text segments = text.split("`") for index in range(0, len(segments), 2): # nur ausserhalb Backticks for term in terms: segments[index] = re.sub(r"(? 1 else "" source = config["measurementSource"] measurementText = remainder if source == "labelRemainder" \ else (row.get(source) or "").strip() values["date"] = formatDate(values.get("date", ""), config) values["resistance"], values["voltage"], leftover = parseMeasurements( measurementText, config, values["label"]) values["remark"] = tidyRemark(leftover, config) or config["defaultRemark"] for role in config["numericRoles"]: if role in values: values[role] = formatNumber(values[role], None, config["decimalPoint"]) return values def render(template, values, terms): try: return escapeWikiTerms(template.format(**values), terms) except KeyError as error: raise SystemExit("Unbekannter Platzhalter im Template: %s" % error) def sortKey(values): match = TRAILING_NUMBER_RE.search(values["label"]) return (TRAILING_NUMBER_RE.sub("", values["label"]).strip(), int(match.group(1)) if match else 0) def checkPreambleConstants(allValues, config): for role in config["preambleRoles"]: distinct = {values.get(role, "") for values in allValues} if len(distinct) > 1: warn("Rolle %r ist nicht konstant (%s) -- Vorspann zeigt den " "Wert der ersten Zeile." % (role, ", ".join(sorted(distinct)))) def warn(message): print("Warnung: " + message, file=sys.stderr) def buildTable(rows, config): allValues = [buildValues(row, config) for row in rows] if config["sortByTrailingNumber"]: allValues.sort(key=sortKey) lines = [] if config["emitPreamble"]: checkPreambleConstants(allValues, config) lines.append(render(config["preambleTemplate"], allValues[0], config["escapeTerms"])) lines.append("") lines.append(config["headerRow"]) for values in allValues: cells = [render(template, values, config["escapeTerms"]) for template in config["cellTemplates"]] lines.append("||" + "||".join(" %s " % cell if cell else " " for cell in cells) + "||") return "\n".join(lines) + "\n" def main(): parser = argparse.ArgumentParser( description="Erzeugt eine Trac-Wiki-Tabelle aus einer Zellen-CSV.") parser.add_argument("csvFile", nargs="?", help="Eingabe-CSV") parser.add_argument("-o", "--output", help="Ausgabedatei (Vorgabe: stdout)") parser.add_argument("-c", "--config", help="JSON-Datei mit Abweichungen " "von der Vorgabe-Konfiguration") parser.add_argument("--dump-config", action="store_true", help="Vorgabe-Konfiguration als JSON ausgeben und beenden") arguments = parser.parse_args() if arguments.dump_config: json.dump(DEFAULT_CONFIG, sys.stdout, indent=2, ensure_ascii=False) sys.stdout.write("\n") return if not arguments.csvFile: parser.error("Eingabe-CSV fehlt (oder --dump-config verwenden)") config = loadConfig(arguments.config) rows = readRows(arguments.csvFile, config) if not rows: raise SystemExit("Die CSV enthaelt keine Datenzeilen.") table = buildTable(rows, config) if arguments.output: with open(arguments.output, "w", encoding="utf-8") as stream: stream.write(table) else: sys.stdout.write(table) if __name__ == "__main__": main()