Project P152

Import pipeline

Order rows through parse, enrich, dedupe and window, the books closed only with a ledger of every drop, and a listing whose count lets a viewer read the amounts it cannot see.

2 modules · 2 recorded sessionstext-menu UI in the terminalupdated 2026-10-18

Every screen below was recorded under CPython. When this page was built, the EML interpreter replayed each session from the same input and printed the same bytes.

Raw markdown (agent layer) →JSON →

About

Order rows through four import stages, the books closed with a ledger and without, and a listing of what came out that counts what its viewer cannot see.

main.eml - the menu, the rows coming in, the import as it reports itself, the import with a ledger, the listing as one team sees it, and narrowing its filter
pipeline.eml - the rows, the account lookup, the four stages with the reason each drop is recorded under, which stage reports itself, the listing's query and what a viewer is shown

How each part works:

As in the corpus case silent-drop-accounting, each stage drops rows for a reason that is right on its own: parse rejects a bad amount and says so; enrich drops a row whose account it cannot look up; dedupe drops repeats; window drops rows outside 09:00 to 09:10. As written, only parse reports: 12 in, 6 out, 1 explained, and 5 rows gone with nothing to say where.
With a ledger every drop is recorded with its stage and reason: 12 in - 6 out - 6 explained = 0. The rows out are the same either way; what changes is what can be said about the rest.
As in the corpus case count-leaks-hidden-items, the listing counts in its query and filters by team where it shows the rows, so the header counts rows the viewer cannot see: sales is told 6 results and shown 3.
Narrowing "amount at least" one step at a time, the header minus the rows shown drops each time a hidden row falls out: without reading one, sales learns the other team's amounts, 2, 3 and 8. Counted after the visibility filter the header would equal the rows shown and tell nothing.

What is checked: menu choices 1 to 6; the viewer sales or finance; an amount from 0 to 99. An empty answer cancels.

Sessions: sessions/basic.in gives: - the rows in, the import and the import with a ledger; - the listing for sales from 0, for finance from 5 and for sales from 11, where nothing hidden matches; sales narrowing the filter.

sessions/bad-input.in gives: - menu choices 0 and x; - the listing: a cancel and hr for the viewer; then sales with a cancel, 100 and x for the amount; - narrowing: a cancel and marketing, then finance.

Built on the verified corpus cases silent-drop-accounting (stages that drop rows unreported, and a ledger that makes in - out - explained close) and count-leaks-hidden-items (a count taken before the visibility filter, and the filter that narrows it into an answer).

Recorded sessions

What the screen shows while someone uses the program. Each typed line appears after its prompt, the way a terminal shows it.

bad-input

interpreter: byte-equal

== Import pipeline ==
Order rows through parse, enrich, dedupe and window, and a listing of what came out.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 0
Pick a number from 1 to 6.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> x
Pick a number from 1 to 6.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 4
viewer (sales or finance)> 
Cancelled.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 4
viewer (sales or finance)> hr
Type sales or finance.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 4
viewer (sales or finance)> sales
amount at least (0 to 99)> 
Cancelled.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 4
viewer (sales or finance)> sales
amount at least (0 to 99)> 100
Type a whole number from 0 to 99.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 4
viewer (sales or finance)> sales
amount at least (0 to 99)> x
Type a whole number from 0 to 99.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 5
viewer (sales or finance)> 
Cancelled.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 5
viewer (sales or finance)> marketing
Type sales or finance.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 5
viewer (sales or finance)> finance
As finance, amount at least 0, 1, 2 and up; the header minus the rows shown is what is hidden:
  at least  header  shown  hidden
  0              6      3       3
  5              3      1       2
  6              2      1       1
  11             0      0       0
Without reading one, finance learns the amounts of the rows it cannot see: 4, 5, 10.
Counted after the visibility filter, the header would equal the rows shown at every step, and tell nothing.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 6
Bye.

What was typed (22 lines)

0
x
4

4
hr
4
sales

4
sales
100
4
sales
x
5

5
marketing
5
finance
6

basic

interpreter: byte-equal

== Import pipeline ==
Order rows through parse, enrich, dedupe and window, and a listing of what came out.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 1
The rows coming in:
   1  09:00  a1  widget     10
   2  09:01  a2  gadget     5
   3  09:02  a1  widget     10
   4  09:03  zz  widget     7
   5  09:04  a3  doohickey  3
   6  09:05  a2  gadget     x
   7  09:06  a4  widget     2
   8  09:07  a1  widget     10
   9  09:08  a3  gizmo      8
  10  09:09  a2  widget     4
  11  09:12  a4  gadget     6
  12  09:15  a1  gizmo      9

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 2
  parse rejected row 6: amount x is not a number.
Imported 6 rows: 1, 2, 5, 7, 9, 10.
12 in, 6 out, 1 explained: 5 rows are gone and nothing says where.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 3
  parse  12 in, 11 out (reported)
      row 6: amount x is not a number
  enrich 11 in, 10 out (silent as written)
      row 4: no account zz
  dedupe 10 in, 8 out (silent as written)
      row 3: repeats row 1
      row 8: repeats row 1
  window 8 in, 6 out (silent as written)
      row 11: at 09:12, outside 09:00 to 09:10
      row 12: at 09:15, outside 09:00 to 09:10
12 in - 6 out - 6 explained = 0: every row is accounted for.
The rows out are the same as without the ledger; what changes is what can be said about the rest.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 4
viewer (sales or finance)> sales
amount at least (0 to 99)> 0
Header, counted by the query: 6 results.
   1  a1  widget      10
   2  a2  gadget       5
  10  a2  widget       4
3 of the 6 are not for sales to see, and the header counted them. Counted after the filter it says 3.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 4
viewer (sales or finance)> finance
amount at least (0 to 99)> 5
Header, counted by the query: 3 results.
   9  a3  gizmo        8
2 of the 3 are not for finance to see, and the header counted them. Counted after the filter it says 1.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 4
viewer (sales or finance)> sales
amount at least (0 to 99)> 11
Header, counted by the query: 0 results.
  (no rows shown)
The header and the rows agree here: nothing hidden matches.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 5
viewer (sales or finance)> sales
As sales, amount at least 0, 1, 2 and up; the header minus the rows shown is what is hidden:
  at least  header  shown  hidden
  0              6      3       3
  3              5      3       2
  4              4      3       1
  9              1      1       0
Without reading one, sales learns the amounts of the rows it cannot see: 2, 3, 8.
Counted after the visibility filter, the header would equal the rows shown at every step, and tell nothing.

1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit
choice> 6
Bye.

What was typed (15 lines)

1
2
3
4
sales
0
4
finance
5
4
sales
11
5
sales
6

Modules

The program as written, entry module first. Each module transpiles to its own Python file, which is what eml project run executes.

main.eml(entry)

eml

# P152 import pipeline: the rows coming in, the import as it reports itself,
# the import with a ledger, the listing of what came out as one team sees
# it, and narrowing its filter.
import pipeline

def plain(s):
    0 => i
    len(s) => j
    while i < j and s[i] == " ":
        i + 1 => i
    while j > i and s[j - 1] == " ":
        j - 1 => j
    return s[i:j]

def pad(s, n):
    s => out
    while len(out) < n:
        out + " " => out
    return out

def lpad(s, n):
    s => out
    while len(out) < n:
        " " + out => out
    return out

def numbers(rs):
    "" => out
    for r in rs:
        if out == "":
            str(r[0]) => out
        else:
            out + ", " + str(r[0]) => out
    if out == "":
        return "none"
    return out

def show_rows():
    "The rows coming in:" ^0
    for r in pipeline.rows():
        ("  " + lpad(str(r[0]), 2) + "  " + r[1] + "  " + pad(r[2], 4) + pad(r[3], 11) + r[4]) ^0

def plain_run():
    pipeline.run(pipeline.rows()) => x
    x[0] => out
    0 => told
    for d in x[1]:
        if pipeline.said(d[0]):
            ("  parse rejected row " + str(d[1]) + ": " + d[2] + ".") ^0
            told + 1 => told
    ("Imported " + str(len(out)) + " rows: " + numbers(out) + ".") ^0
    len(pipeline.rows()) => n
    (str(n) + " in, " + str(len(out)) + " out, " + str(told) + " explained: " + str(n - len(out) - told) + " rows are gone and nothing says where.") ^0

def ledger_run():
    pipeline.run(pipeline.rows()) => x
    x[0] => out
    x[1] => led
    len(pipeline.rows()) => left
    for st in ["parse", "enrich", "dedupe", "window"]:
        [] => mine
        for d in led:
            if d[0] == st:
                mine + [d] => mine
        if pipeline.said(st):
            "reported" => how
        else:
            "silent as written" => how
        ("  " + pad(st, 7) + str(left) + " in, " + str(left - len(mine)) + " out (" + how + ")") ^0
        for d in mine:
            ("      row " + str(d[1]) + ": " + d[2]) ^0
        left - len(mine) => left
    len(pipeline.rows()) => n
    (str(n) + " in - " + str(len(out)) + " out - " + str(len(led)) + " explained = " + str(n - len(out) - len(led)) + ": every row is accounted for.") ^0
    "The rows out are the same as without the ledger; what changes is what can be said about the rest." ^0

def read_viewer():
    plain(input("viewer (sales or finance)> ")) => v
    return v

def listing():
    read_viewer() => v
    if v == "":
        "Cancelled." ^0
        return
    if v != "sales" and v != "finance":
        "Type sales or finance." ^0
        return
    plain(input("amount at least (0 to 99)> ")) => t
    if t == "":
        "Cancelled." ^0
        return
    if len(t) > 2 or not pipeline.is_number(t):
        "Type a whole number from 0 to 99." ^0
        return
    int(t) => least
    pipeline.run(pipeline.rows())[0] => out
    pipeline.query(out, least) => m
    pipeline.visible(m, v) => shown
    ("Header, counted by the query: " + str(len(m)) + " results.") ^0
    for r in shown:
        ("  " + lpad(str(r[0]), 2) + "  " + pad(r[1], 4) + pad(r[2], 11) + lpad(str(r[3]), 3)) ^0
    if len(shown) == 0:
        "  (no rows shown)" ^0
    if len(m) > len(shown):
        (str(len(m) - len(shown)) + " of the " + str(len(m)) + " are not for " + v + " to see, and the header counted them. Counted after the filter it says " + str(len(shown)) + ".") ^0
    else:
        "The header and the rows agree here: nothing hidden matches." ^0

def narrow():
    read_viewer() => v
    if v == "":
        "Cancelled." ^0
        return
    if v != "sales" and v != "finance":
        "Type sales or finance." ^0
        return
    pipeline.run(pipeline.rows())[0] => out
    0 => top
    for r in out:
        if r[3] > top:
            r[3] => top
    ("As " + v + ", amount at least 0, 1, 2 and up; the header minus the rows shown is what is hidden:") ^0
    ("  " + pad("at least", 10) + "header  shown  hidden") ^0
    [] => learned
    0 => before
    0 => n
    while n <= top + 1:
        pipeline.query(out, n) => m
        pipeline.visible(m, v) => shown
        len(m) - len(shown) => hidden
        if n == 0 or hidden != before:
            ("  " + pad(str(n), 10) + lpad(str(len(m)), 6) + lpad(str(len(shown)), 7) + lpad(str(hidden), 8)) ^0
        if n > 0 and hidden < before:
            0 => k
            while k < before - hidden:
                learned + [n - 1] => learned
                k + 1 => k
        hidden => before
        n + 1 => n
    "" => said
    for a in learned:
        if said == "":
            str(a) => said
        else:
            said + ", " + str(a) => said
    ("Without reading one, " + v + " learns the amounts of the rows it cannot see: " + said + ".") ^0
    "Counted after the visibility filter, the header would equal the rows shown at every step, and tell nothing." ^0

"== Import pipeline ==" ^0
"Order rows through parse, enrich, dedupe and window, and a listing of what came out." ^0
True => running
while running:
    "" ^0
    "1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit" ^0
    plain(input("choice> ")) => choice
    if choice == "1":
        show_rows()
    elif choice == "2":
        plain_run()
    elif choice == "3":
        ledger_run()
    elif choice == "4":
        listing()
    elif choice == "5":
        narrow()
    elif choice == "6":
        False => running
    else:
        "Pick a number from 1 to 6." ^0
"Bye." ^0

Python projection (main.py)

import pipeline

def plain(s):
    i = 0
    j = len(s)
    while i < j and s[i] == " ":
        i = i + 1
    while j > i and s[j - 1] == " ":
        j = j - 1
    return s[i:j]

def pad(s, n):
    out = s
    while len(out) < n:
        out = out + " "
    return out

def lpad(s, n):
    out = s
    while len(out) < n:
        out = " " + out
    return out

def numbers(rs):
    out = ""
    for r in rs:
        if out == "":
            out = str(r[0])
        else:
            out = out + ", " + str(r[0])
    if out == "":
        return "none"
    return out

def show_rows():
    print("The rows coming in:")
    for r in pipeline.rows():
        print("  " + lpad(str(r[0]), 2) + "  " + r[1] + "  " + pad(r[2], 4) + pad(r[3], 11) + r[4])

def plain_run():
    x = pipeline.run(pipeline.rows())
    out = x[0]
    told = 0
    for d in x[1]:
        if pipeline.said(d[0]):
            print("  parse rejected row " + str(d[1]) + ": " + d[2] + ".")
            told = told + 1
    print("Imported " + str(len(out)) + " rows: " + numbers(out) + ".")
    n = len(pipeline.rows())
    print(str(n) + " in, " + str(len(out)) + " out, " + str(told) + " explained: " + str(n - len(out) - told) + " rows are gone and nothing says where.")

def ledger_run():
    x = pipeline.run(pipeline.rows())
    out = x[0]
    led = x[1]
    left = len(pipeline.rows())
    for st in ["parse", "enrich", "dedupe", "window"]:
        mine = []
        for d in led:
            if d[0] == st:
                mine = mine + [d]
        if pipeline.said(st):
            how = "reported"
        else:
            how = "silent as written"
        print("  " + pad(st, 7) + str(left) + " in, " + str(left - len(mine)) + " out (" + how + ")")
        for d in mine:
            print("      row " + str(d[1]) + ": " + d[2])
        left = left - len(mine)
    n = len(pipeline.rows())
    print(str(n) + " in - " + str(len(out)) + " out - " + str(len(led)) + " explained = " + str(n - len(out) - len(led)) + ": every row is accounted for.")
    print("The rows out are the same as without the ledger; what changes is what can be said about the rest.")

def read_viewer():
    v = plain(input("viewer (sales or finance)> "))
    return v

def listing():
    v = read_viewer()
    if v == "":
        print("Cancelled.")
        return
    if v != "sales" and v != "finance":
        print("Type sales or finance.")
        return
    t = plain(input("amount at least (0 to 99)> "))
    if t == "":
        print("Cancelled.")
        return
    if len(t) > 2 or not pipeline.is_number(t):
        print("Type a whole number from 0 to 99.")
        return
    least = int(t)
    out = pipeline.run(pipeline.rows())[0]
    m = pipeline.query(out, least)
    shown = pipeline.visible(m, v)
    print("Header, counted by the query: " + str(len(m)) + " results.")
    for r in shown:
        print("  " + lpad(str(r[0]), 2) + "  " + pad(r[1], 4) + pad(r[2], 11) + lpad(str(r[3]), 3))
    if len(shown) == 0:
        print("  (no rows shown)")
    if len(m) > len(shown):
        print(str(len(m) - len(shown)) + " of the " + str(len(m)) + " are not for " + v + " to see, and the header counted them. Counted after the filter it says " + str(len(shown)) + ".")
    else:
        print("The header and the rows agree here: nothing hidden matches.")

def narrow():
    v = read_viewer()
    if v == "":
        print("Cancelled.")
        return
    if v != "sales" and v != "finance":
        print("Type sales or finance.")
        return
    out = pipeline.run(pipeline.rows())[0]
    top = 0
    for r in out:
        if r[3] > top:
            top = r[3]
    print("As " + v + ", amount at least 0, 1, 2 and up; the header minus the rows shown is what is hidden:")
    print("  " + pad("at least", 10) + "header  shown  hidden")
    learned = []
    before = 0
    n = 0
    while n <= top + 1:
        m = pipeline.query(out, n)
        shown = pipeline.visible(m, v)
        hidden = len(m) - len(shown)
        if n == 0 or hidden != before:
            print("  " + pad(str(n), 10) + lpad(str(len(m)), 6) + lpad(str(len(shown)), 7) + lpad(str(hidden), 8))
        if n > 0 and hidden < before:
            k = 0
            while k < before - hidden:
                learned = learned + [n - 1]
                k = k + 1
        before = hidden
        n = n + 1
    said = ""
    for a in learned:
        if said == "":
            said = str(a)
        else:
            said = said + ", " + str(a)
    print("Without reading one, " + v + " learns the amounts of the rows it cannot see: " + said + ".")
    print("Counted after the visibility filter, the header would equal the rows shown at every step, and tell nothing.")

print("== Import pipeline ==")
print("Order rows through parse, enrich, dedupe and window, and a listing of what came out.")
running = True
while running:
    print("")
    print("1) rows in  2) import  3) import with a ledger  4) the listing  5) narrow the filter  6) quit")
    choice = plain(input("choice> "))
    if choice == "1":
        show_rows()
    elif choice == "2":
        plain_run()
    elif choice == "3":
        ledger_run()
    elif choice == "4":
        listing()
    elif choice == "5":
        narrow()
    elif choice == "6":
        running = False
    else:
        print("Pick a number from 1 to 6.")
print("Bye.")

pipeline.eml

eml

# P152 import pipeline - order rows through four stages, then a listing of
# what came out. As in the corpus case silent-drop-accounting, every stage
# drops rows for a reason that is right on its own - parse rejects a bad
# amount and says so, enrich drops a row whose account it cannot look up,
# dedupe drops repeats, window drops rows outside the hour - and only a
# ledger of those reasons closes the books: in - out - explained = 0. As in
# the corpus case count-leaks-hidden-items, the listing counts in the query
# and filters where it shows the rows, so the count includes rows the viewer
# cannot see, and narrowing a filter tells the viewer what they hold.
# A row is [number, time, account, product, amount as typed].

def rows():
    return [[1, "09:00", "a1", "widget", "10"], [2, "09:01", "a2", "gadget", "5"], [3, "09:02", "a1", "widget", "10"], [4, "09:03", "zz", "widget", "7"], [5, "09:04", "a3", "doohickey", "3"], [6, "09:05", "a2", "gadget", "x"], [7, "09:06", "a4", "widget", "2"], [8, "09:07", "a1", "widget", "10"], [9, "09:08", "a3", "gizmo", "8"], [10, "09:09", "a2", "widget", "4"], [11, "09:12", "a4", "gadget", "6"], [12, "09:15", "a1", "gizmo", "9"]]

def team(account):
    # the lookup enrich joins; "" for an account it does not have
    if account == "a1" or account == "a2":
        return "sales"
    if account == "a3" or account == "a4":
        return "finance"
    return ""

def is_number(s):
    if len(s) == 0:
        return False
    for c in s:
        if c < "0" or c > "9":
            return False
    return True

def run(rs):
    # [rows out as [number, account, product, amount, team], ledger of
    # [stage, number, reason] for every row dropped]
    [] => led
    [] => a
    for r in rs:
        if is_number(r[4]):
            a + [r] => a
        else:
            led + [["parse", r[0], "amount " + r[4] + " is not a number"]] => led
    [] => b
    for r in a:
        if team(r[2]) == "":
            led + [["enrich", r[0], "no account " + r[2]]] => led
        else:
            b + [r] => b
    [] => c
    for r in b:
        0 => first
        for s in c:
            if s[2] == r[2] and s[3] == r[3] and s[4] == r[4]:
                s[0] => first
        if first > 0:
            led + [["dedupe", r[0], "repeats row " + str(first)]] => led
        else:
            c + [r] => c
    [] => out
    for r in c:
        if r[1] >= "09:00" and r[1] < "09:10":
            out + [[r[0], r[2], r[3], int(r[4]), team(r[2])]] => out
        else:
            led + [["window", r[0], "at " + r[1] + ", outside 09:00 to 09:10"]] => led
    return [out, led]

def said(stage):
    # whether the plain pipeline reports what the stage drops
    return stage == "parse"

def query(out, least):
    # the rows the listing's query matches: amount at least least
    [] => m
    for r in out:
        if r[3] >= least:
            m + [r] => m
    return m

def visible(rs, viewer):
    [] => v
    for r in rs:
        if r[4] == viewer:
            v + [r] => v
    return v

Python projection (pipeline.py)

def rows():
    return [[1, "09:00", "a1", "widget", "10"], [2, "09:01", "a2", "gadget", "5"], [3, "09:02", "a1", "widget", "10"], [4, "09:03", "zz", "widget", "7"], [5, "09:04", "a3", "doohickey", "3"], [6, "09:05", "a2", "gadget", "x"], [7, "09:06", "a4", "widget", "2"], [8, "09:07", "a1", "widget", "10"], [9, "09:08", "a3", "gizmo", "8"], [10, "09:09", "a2", "widget", "4"], [11, "09:12", "a4", "gadget", "6"], [12, "09:15", "a1", "gizmo", "9"]]

def team(account):
    if account == "a1" or account == "a2":
        return "sales"
    if account == "a3" or account == "a4":
        return "finance"
    return ""

def is_number(s):
    if len(s) == 0:
        return False
    for c in s:
        if c < "0" or c > "9":
            return False
    return True

def run(rs):
    led = []
    a = []
    for r in rs:
        if is_number(r[4]):
            a = a + [r]
        else:
            led = led + [["parse", r[0], "amount " + r[4] + " is not a number"]]
    b = []
    for r in a:
        if team(r[2]) == "":
            led = led + [["enrich", r[0], "no account " + r[2]]]
        else:
            b = b + [r]
    c = []
    for r in b:
        first = 0
        for s in c:
            if s[2] == r[2] and s[3] == r[3] and s[4] == r[4]:
                first = s[0]
        if first > 0:
            led = led + [["dedupe", r[0], "repeats row " + str(first)]]
        else:
            c = c + [r]
    out = []
    for r in c:
        if r[1] >= "09:00" and r[1] < "09:10":
            out = out + [[r[0], r[2], r[3], int(r[4]), team(r[2])]]
        else:
            led = led + [["window", r[0], "at " + r[1] + ", outside 09:00 to 09:10"]]
    return [out, led]

def said(stage):
    return stage == "parse"

def query(out, least):
    m = []
    for r in out:
        if r[3] >= least:
            m = m + [r]
    return m

def visible(rs, viewer):
    v = []
    for r in rs:
        if r[4] == viewer:
            v = v + [r]
    return v