adapter updates

soleprint/station/tools/dataconvert/README.md

@@ -46,6 +46,8 @@ Every key is optional.
 
 - **out_dir, max_rows, schema** are the run settings: the same as `--out-dir`,
   `--max-rows` and `--no-schema`, and a flag on the command line wins over the file.
+  There is no default output directory: without `--out-dir` or an `out_dir` in the
+  config, the run stops before writing anything.
   `out_dir` is relative to the folder the config is in, so `"sample"` beside the data
   means `<source>/sample`. Inputs whose configs name different `out_dir`s each get
   their own `SCHEMA.md`.
@@ -87,6 +89,7 @@ what pandas read: a starting point, not DDL. `--no-schema` skips the file.
 | `dataconvert.py` | command line |
 | `config.py` | `dataconvert.json`: layouts, sheet naming and run settings |
 | `readers.py` | files, directories, ZIPs and globs into DataFrames |
+| `progress.py` | progress lines: which file is being read, and how far a long table has got |
 | `sqlgen.py` | DataFrames into INSERT statements, with the row cap |
 | `output.py` | file naming and writing |
 | `schema.py` | `SCHEMA.md` |

soleprint/station/tools/dataconvert/config.py

@@ -101,17 +101,16 @@ class Config:
         self.out_dir = out_dir
 
     @property
-    def detects(self):
-        return self.forced is None and (bool(self.rules) or self.fallback is not None)
+    def probe_rows(self):
+        """How many rows the layouts' marker cells need to be read."""
+        return max((rule.row for rule in self.rules), default=0)
 
-    def layout_for(self, raw_df):
-        """The layout for a sheet read without a header, or None for the default read."""
-        if self.forced is not None:
-            return None if self.forced.is_default else self.forced
+    def match(self, probe_df):
+        """The first layout whose marker cell is in these top rows, or None."""
         for rule in self.rules:
-            if rule.matches(raw_df):
+            if rule.matches(probe_df):
                 return rule.layout
-        return self.fallback
+        return None
 
     def settings(self):
         """What the file set, for the line the run prints."""

soleprint/station/tools/dataconvert/dataconvert.py

@@ -28,6 +28,7 @@ import argparse
 from pathlib import Path
 
 from output import write_tables
+from progress import say
 import config as cfg
 from readers import expand_inputs, iter_sources
 from schema import SchemaReport
@@ -44,7 +45,7 @@ def main():
     parser = argparse.ArgumentParser(description="Convert data sources into schema-agnostic SQL seed files.")
     parser.add_argument("--input", nargs="+", required=True, help="Input file(s), directory, wildcard pattern(s), or ZIP archive(s)")
     parser.add_argument("--out-dir", default=None,
-                        help="Directory where individual .sql files will be written (default: the config's out_dir, else ./seed)")
+                        help="Directory where individual .sql files will be written (default: the config's out_dir; one of the two is required)")
     parser.add_argument("--max-rows", type=positive_int, default=None,
                         help="Write at most N rows per table; the header and SCHEMA.md note the full row count and size (default: the config's max_rows, else all)")
     parser.add_argument("--no-schema", action="store_true", help="Do not write SCHEMA.md, whatever the config says")
@@ -81,6 +82,17 @@ def main():
                 parser.error(str(e))
         plan.append((in_path, loaded[found] if found is not None else cfg.Config(forced=forced)))
 
+    # No silent default directory. A run whose config was not found would
+    # otherwise write, unlaid-out, into ./seed wherever it was started, and look
+    # like it worked. Checked for every input before anything is written.
+    if not args.out_dir:
+        for in_path, config in plan:
+            if config.out_dir is not None:
+                continue
+            where = (f"{config.source} sets no out_dir" if config.source is not None
+                     else f"no {cfg.FILENAME} found in its folder")
+            parser.error(f"no output directory for {in_path} ({where}): pass --out-dir, or set out_dir in the config")
+
     # One report per output directory: inputs whose configs name different
     # out_dirs each get their own SCHEMA.md, beside their own .sql files.
     reports = {}
@@ -89,14 +101,13 @@ def main():
         if config.source is not None and config.source not in announced:
             announced.add(config.source)
             said = config.settings()
-            print(f"[dataconvert] config: {config.source}" + (f" ({said})" if said else ""))
+            say(f"config: {config.source}" + (f" ({said})" if said else ""))
 
-        # The command line wins, then the config, then the built-in default.
-        out_dir = Path(args.out_dir) if args.out_dir else (config.out_dir or Path("seed"))
+        # The command line wins, then the config.
+        out_dir = Path(args.out_dir) if args.out_dir else config.out_dir
         max_rows = args.max_rows if args.max_rows is not None else config.max_rows
         schema = not args.no_schema and config.schema is not False
 
-        out_dir.mkdir(parents=True, exist_ok=True)
         key = out_dir.resolve()
         if key not in reports:
             reports[key] = (out_dir, SchemaReport(), set())
@@ -107,11 +118,12 @@ def main():
             write_tables(dfs, out_dir, source_name, max_rows, report if schema else None, config.bare_sheet_prefixes)
 
     for out_dir, report, caps in reports.values():
-        if report.entries:
+        if report.entries and out_dir.is_dir():
             # The sampling note names the cap only when every input here used
             # the same one; otherwise each table's row still says what it kept.
             cap = next(iter(caps)) if len(caps) == 1 else None
-            print(f"[dataconvert] Generated: {report.write(out_dir, cap)}")
+            say(f"writing SCHEMA.md for {len(report.tables())} tables")
+            say(f"Generated: {report.write(out_dir, cap)}")
 
 
 if __name__ == "__main__":

soleprint/station/tools/dataconvert/output.py

@@ -4,6 +4,7 @@ Output: one .sql file per table or sheet, named after it.
 
 from pathlib import Path
 
+from progress import say
 from sqlgen import render_table, sanitize_identifier
 
 
@@ -29,6 +30,9 @@ def write_tables(dfs: dict, out_dir: Path, source_name: str, max_rows=None, repo
         filename = table_filename(raw_name, source_name, len(dfs), bare_prefixes)
         sql, total, full_bytes, exact = render_table(df, table, max_rows)
         out_file = out_dir / filename
+        # Created on the first file, so a run that finds nothing to convert
+        # leaves no empty directory behind.
+        out_dir.mkdir(parents=True, exist_ok=True)
 
         # Several sources can feed the same table; they accumulate in one file.
         mode = "a" if out_file.exists() else "w"
@@ -40,4 +44,4 @@ def write_tables(dfs: dict, out_dir: Path, source_name: str, max_rows=None, repo
             report.add(source_name, table, filename, df, total, full_bytes, exact, shown)
 
         suffix = "" if exact else f" ({shown} of {total} rows)"
-        print(f"[dataconvert] Generated: {out_file}{suffix}")
+        say(f"Generated: {out_file}{suffix}")

soleprint/station/tools/dataconvert/progress.py

@@ -0,0 +1,39 @@
+"""
+Progress lines: enough to see that a long run is alive and where it is, not so
+many that printing them slows it down.
+
+Every line is flushed, so progress shows up as it happens even when the output
+is piped or tee'd into a log.
+"""
+
+import time
+
+PREFIX = "[dataconvert]"
+
+
+def say(message):
+    print(f"{PREFIX} {message}", flush=True)
+
+
+def seconds(start):
+    return f"{time.monotonic() - start:.0f}s"
+
+
+class Ticker:
+    """How far a long loop has got, at most once every `every` seconds."""
+
+    # How often the loop asks. Checking the clock on every row would itself be
+    # a measurable cost on a million-row table.
+    CHECK = 2000
+
+    def __init__(self, label, total, every=10.0):
+        self.label = label
+        self.total = total
+        self.every = every
+        self.started = self.last = time.monotonic()
+
+    def tick(self, done):
+        now = time.monotonic()
+        if now - self.last >= self.every:
+            self.last = now
+            say(f"  {self.label}: {done} of {self.total} rows ({now - self.started:.0f}s)")

soleprint/station/tools/dataconvert/readers.py

@@ -8,11 +8,15 @@ one per spreadsheet or CSV. Nothing here knows about SQL or output files.
 import glob
 import os
 import tempfile
+import time
 import zipfile
 from pathlib import Path
 
 import pandas as pd
 
+from progress import say, seconds
+from sqlgen import human_bytes
+
 SUPPORTED = {".csv", ".xlsx", ".xls", ".ods"}
 
 
@@ -23,7 +27,7 @@ def expand_inputs(inputs):
         if any(c in in_str for c in ["*", "?", "["]):
             matched = glob.glob(in_str, recursive=True)
             if not matched:
-                print(f"[Warning] No files matched wildcard pattern: '{in_str}'")
+                print(f"[Warning] No files matched wildcard pattern: '{in_str}'", flush=True)
             paths.extend(Path(m) for m in sorted(matched))
         else:
             paths.append(Path(in_str))
@@ -34,16 +38,22 @@ def read_sheet(read, config):
     """
     One sheet or CSV, laid out as the config says.
 
-    With nothing to detect, a plain read, exactly as before. Otherwise the sheet
-    is read once without a header, checked against the layouts, and when one
-    matches the header and data rows are sliced out of that same read.
+    With no layout to apply, a plain read, exactly as before. The layouts'
+    marker cells are checked on only the top few rows, so a big sheet that
+    matches none of them is read once, not twice. When a layout applies, the
+    sheet is read without a header and the header and data rows are sliced out.
     """
-    if not config.detects and config.forced is None:
-        return read(), None
-    raw = read(header=None)
-    layout = config.layout_for(raw)
+    layout = None
+    if config.forced is not None:
+        layout = None if config.forced.is_default else config.forced
+    else:
+        if config.rules:
+            layout = config.match(read(header=None, nrows=config.probe_rows))
+        if layout is None:
+            layout = config.fallback
     if layout is None:
         return read(), None
+    raw = read(header=None)
     data = raw.iloc[layout.data_row - 1:].copy()
     data.columns = [str(c).strip() for c in raw.iloc[layout.header_row - 1]]
     return data, layout
@@ -54,26 +64,37 @@ def load_dataframes_from_file(file_path: Path, config) -> dict:
     ext = file_path.suffix.lower()
     dfs = {}
 
+    # Said before the read, not after: a big workbook can take minutes inside
+    # pandas, and this line is what says which file that is.
+    say(f"reading {file_path.name} ({human_bytes(file_path.stat().st_size)})")
     try:
         if ext == ".csv":
-            dfs[file_path.stem], layout = read_sheet(lambda **kw: pd.read_csv(file_path, **kw), config)
-            note_layout(file_path.name, None, layout)
+            start = time.monotonic()
+            # low_memory=False: by default pandas types a big CSV chunk by
+            # chunk, so one column can come out as numbers in one chunk and
+            # text in the next, quoted differently row to row in the SQL. Read
+            # whole, each column gets one type. Costs memory on huge files.
+            df, layout = read_sheet(lambda **kw: pd.read_csv(file_path, low_memory=False, **kw), config)
+            dfs[file_path.stem] = df
+            say(f"  {len(df)} rows{layout_note(layout)} ({seconds(start)})")
         elif ext in [".xlsx", ".xls", ".ods"]:
             xls = pd.ExcelFile(file_path, engine="odf") if ext == ".ods" else pd.ExcelFile(file_path)
             for sheet in xls.sheet_names:
-                dfs[sheet], layout = read_sheet(
+                start = time.monotonic()
+                df, layout = read_sheet(
                     lambda sheet=sheet, **kw: pd.read_excel(xls, sheet_name=sheet, **kw), config)
-                note_layout(file_path.name, sheet, layout)
+                dfs[sheet] = df
+                say(f"  sheet {sheet}: {len(df)} rows{layout_note(layout)} ({seconds(start)})")
     except Exception as e:
-        print(f"[Warning] Could not read '{file_path.name}': {e}")
+        print(f"[Warning] Could not read '{file_path.name}': {e}", flush=True)
 
     return dfs
 
 
-def note_layout(file_name, sheet, layout):
-    if layout is not None:
-        where = f"{file_name} [{sheet}]" if sheet else file_name
-        print(f"[dataconvert] {where}: layout '{layout.name}', header row {layout.header_row}, data from row {layout.data_row}")
+def layout_note(layout):
+    if layout is None:
+        return ""
+    return f", layout '{layout.name}' (header row {layout.header_row}, data from row {layout.data_row})"
 
 
 def iter_sources(path: Path, config):

soleprint/station/tools/dataconvert/sqlgen.py

@@ -10,6 +10,8 @@ import re
 
 import pandas as pd
 
+from progress import Ticker
+
 
 def sanitize_identifier(identifier: str) -> str:
     """Sanitize names for SQL tables, columns, and filenames."""
@@ -45,11 +47,14 @@ def render_table(df: pd.DataFrame, table_name: str, max_rows=None):
     shown = df if max_rows is None else df.head(max_rows)
 
     rows = []
+    ticker = Ticker(table_name, len(shown))
     # iterrows, not itertuples: it hands values over the way the original tool
     # did, and seed files people already load depend on exactly that quoting.
-    for _, row in shown.iterrows():
+    for i, (_, row) in enumerate(shown.iterrows(), 1):
         vals = ", ".join(sql_value(v) for v in row)
         rows.append(f"INSERT INTO {table_ref} ({cols}) VALUES ({vals}) ON CONFLICT DO NOTHING;\n")
+        if i % Ticker.CHECK == 0:
+            ticker.tick(i)
 
     head = f"-- Generated seed data for table: {table_ref}\n"
     begin, commit = "BEGIN;\n\n", "\nCOMMIT;\n"