#!/usr/bin/env python3
"""CS06c: standard-library specification checker, not a DBMS benchmark.

python3 foundations-batch-bitmap-checker.py
python3 foundations-batch-bitmap-checker.py --batch-sizes 1,5 --bitmap-budget 4

All checks remain enabled under python -O. A nonzero exit is a failed check or
invalid request, never a partial successful query. Memory units are pedagogical.
"""
from __future__ import annotations

import argparse
from collections import Counter
from dataclasses import dataclass
import itertools
import json
import sqlite3
from typing import Callable


class ContractError(ValueError):
    pass


class StaleBatch(ContractError):
    pass


class CapacityExceeded(ContractError):
    pass


class ExecutionFailure(RuntimeError):
    pass


def check(condition: bool, message: str) -> None:
    if not condition:
        raise AssertionError(message)


def integer(value: object, name: str, minimum: int = 0) -> int:
    if type(value) is not int or value < minimum:
        raise ContractError(f"{name} must be an integer >= {minimum}")
    return value


@dataclass(frozen=True)
class Row:
    x: int | None
    y: int | None
    v: int

    def __post_init__(self) -> None:
        for name in ("x", "y"):
            value = getattr(self, name)
            if value is not None and (type(value) is not int or value not in (0, 1)):
                raise ContractError(f"{name} must be 0, 1, or None")
        if type(self.v) is not int:
            raise ContractError("v must be an exact integer")


ROWS = tuple(Row(x, y, v) for x, y, v in zip(
    [1, 1, 0, 1, 0, 1, 0, 1, 1, 0, 1, 0],
    [1, 1, 1, 0, 1, 1, None, 1, 0, 0, 1, 1],
    [10, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110]))


def truth_eq_one(value: int | None) -> bool | None:
    return None if value is None else value == 1


def sql_and(a: bool | None, b: bool | None) -> bool | None:
    if a is False or b is False:
        return False
    if a is None or b is None:
        return None
    return True


def sql_or(a: bool | None, b: bool | None) -> bool | None:
    if a is True or b is True:
        return True
    if a is None or b is None:
        return None
    return False


def qualifies(row: Row, op: str = "and") -> bool:
    if op not in ("and", "or"):
        raise ContractError("unknown predicate")
    combine = sql_and if op == "and" else sql_or
    return combine(truth_eq_one(row.x), truth_eq_one(row.y)) is True


def scalar(rows: tuple[Row, ...], op: str = "and") -> list[tuple[int, int]]:
    return [(i, row.v) for i, row in enumerate(rows) if qualifies(row, op)]


def sql_sum(values: list[int]) -> int | None:
    # Internal zero state is finalized to NULL for empty SQL SUM.
    return sum(values) if values else None


@dataclass(frozen=True)
class Result:
    tag: str
    batch: BorrowedBatch | None = None
    cause: str | None = None


@dataclass(frozen=True)
class BorrowedBatch:
    owner: BatchSource
    epoch: int
    n: int
    selection: tuple[int, ...]

    def ensure_live(self) -> None:
        if self.owner.epoch != self.epoch or self.owner.state != "READY":
            raise StaleBatch("batch expired on next pull, failure, or close")

    def get(self, field: str, lane: int) -> int | None:
        self.ensure_live()
        integer(lane, "lane")
        if lane >= self.n:
            raise ContractError("read from invalid tail lane")
        if field not in self.owner.columns:
            raise ContractError("unknown column")
        if not self.owner.validity[field][lane]:
            return None
        return self.owner.columns[field][lane]

    def occurrence(self, lane: int) -> int:
        self.ensure_live()
        integer(lane, "lane")
        if lane >= self.n:
            raise ContractError("invalid occurrence lane")
        return self.owner.ids[lane]

    def select(self, lanes: tuple[int, ...]) -> BorrowedBatch:
        self.ensure_live()
        if type(lanes) is not tuple or any(type(i) is not int for i in lanes):
            raise ContractError("selection must be an integer tuple")
        if any(a >= b for a, b in zip(lanes, lanes[1:])):
            raise ContractError("selection must be strictly increasing")
        allowed = set(self.selection)
        if any(i not in allowed for i in lanes):
            raise ContractError("filter selection must be a subset of active lanes")
        return BorrowedBatch(self.owner, self.epoch, self.n, lanes)

    def snapshot(self, field: str = "v") -> list[tuple[int, int | None]]:
        self.ensure_live()
        return [(self.occurrence(i), self.get(field, i)) for i in self.selection]

    def map_active(self, field: str, function: Callable) -> list:
        # Evaluation is inside the active-lane loop, not before a blend.
        self.ensure_live()
        return [function(self.get(field, i)) for i in self.selection]


class BatchSource:
    def __init__(self, rows: tuple[Row, ...], capacity: int,
                 chunks: tuple[int, ...] | None = None,
                 fail_at: int | None = None) -> None:
        self.capacity = integer(capacity, "capacity", 1)
        self.rows = tuple(rows)
        if any(not isinstance(row, Row) for row in self.rows):
            raise ContractError("rows must contain Row values")
        if chunks is None:
            chunks = tuple(min(capacity, len(rows) - b)
                           for b in range(0, len(rows), capacity))
        if any(type(n) is not int or not 1 <= n <= capacity for n in chunks):
            raise ContractError("source chunks must be positive and <= capacity")
        if sum(chunks) != len(rows):
            raise ContractError("chunks must cover every input occurrence once")
        if fail_at is not None:
            integer(fail_at, "fail_at")
        self.chunks, self.fail_at = chunks, fail_at
        self.columns = {name: [999999] * capacity for name in ("x", "y", "v")}
        self.validity = {name: [False] * capacity for name in self.columns}
        self.ids = [-1] * capacity
        self.epoch = self.cursor = self.chunk_index = 0
        self.state = "READY"
        self.data_calls = self.terminal_calls = 0
        self.cause = None

    def next_batch(self) -> Result:
        if self.state == "CLOSED":
            raise ContractError("pull after close")
        if self.state == "FAILED":
            return Result("ERROR", cause=self.cause)
        if self.state == "EXHAUSTED":
            return Result("END")
        self.epoch += 1  # even a failed/terminal pull invalidates the old view
        if self.fail_at is not None and self.cursor >= self.fail_at:
            self.state, self.cause = "FAILED", "InjectedReadError"
            return Result("ERROR", cause=self.cause)
        if self.cursor == len(self.rows):
            self.state = "EXHAUSTED"
            self.terminal_calls += 1
            return Result("END")
        n = self.chunks[self.chunk_index]
        for j in range(n):
            row = self.rows[self.cursor + j]
            self.ids[j] = self.cursor + j
            for name in self.columns:
                value = getattr(row, name)
                self.validity[name][j] = value is not None
                # Invalid NULL payload deliberately remains stale.
                if value is not None:
                    self.columns[name][j] = value
        # Invalid tail deliberately stays stale too: n is the authority.
        self.cursor += n
        self.chunk_index += 1
        self.data_calls += 1
        return Result("DATA", BorrowedBatch(self, self.epoch, n, tuple(range(n))))

    def close(self) -> None:
        if self.state != "CLOSED":
            self.epoch += 1
            self.state = "CLOSED"


def filter_batch(batch: BorrowedBatch, op: str = "and") -> BorrowedBatch:
    combine = {"and": sql_and, "or": sql_or}.get(op)
    if combine is None:
        raise ContractError("unknown predicate")
    lanes = tuple(j for j in batch.selection if combine(
        truth_eq_one(batch.get("x", j)), truth_eq_one(batch.get("y", j))) is True)
    return batch.select(lanes)


def run_batch(rows: tuple[Row, ...], capacity: int,
              chunks: tuple[int, ...] | None = None,
              fail_at: int | None = None, op: str = "and") -> tuple[list, dict]:
    source = BatchSource(rows, capacity, chunks, fail_at)
    output, empty_batches = [], 0
    try:
        while True:
            result = source.next_batch()
            if result.tag == "END":
                break
            if result.tag == "ERROR":
                raise ExecutionFailure(result.cause)
            batch = filter_batch(result.batch, op)
            empty_batches += not batch.selection
            output.extend(batch.snapshot())  # copy before next pull
        return output, {"data_dispatches": source.data_calls,
                        "terminal_calls": source.terminal_calls,
                        "empty_data_batches": empty_batches,
                        "predicate_rows": len(rows)}
    finally:
        source.close()


@dataclass(frozen=True)
class Page:
    slots: frozenset[int] = frozenset()
    lossy: bool = False
    recheck: bool = False

    def __post_init__(self) -> None:
        if type(self.slots) is not frozenset or any(type(s) is not int or s < 0 for s in self.slots):
            raise ContractError("slots must be a frozenset of nonnegative integers")
        if type(self.lossy) is not bool or type(self.recheck) is not bool:
            raise ContractError("page flags must be booleans")
        if self.lossy and (self.slots or not self.recheck):
            raise ContractError("lossy page has no exact slots and must recheck")

    @property
    def empty(self) -> bool:
        return not self.lossy and not self.slots


def combine_page(a: Page, b: Page, op: str) -> Page:
    if op == "and":
        if a.empty or b.empty:
            return Page()
        if a.lossy and b.lossy:
            return Page(lossy=True, recheck=True)
        if a.lossy:
            return Page(b.slots, recheck=True)
        if b.lossy:
            return Page(a.slots, recheck=True)
        slots = a.slots & b.slots
        return Page(slots, recheck=(a.recheck or b.recheck) if slots else False)
    if op == "or":
        if a.empty:
            return b
        if b.empty:
            return a
        if a.lossy or b.lossy:
            return Page(lossy=True, recheck=True)
        return Page(a.slots | b.slots, recheck=a.recheck or b.recheck)
    raise ContractError("only AND/OR candidate operations are supported")


class Bitmap:
    def __init__(self, total: int, page_size: int = 3,
                 pages: dict[int, Page] | None = None) -> None:
        self.total = integer(total, "total")
        self.page_size = integer(page_size, "page_size", 1)
        self.pages = {}
        for h, page in (pages or {}).items():
            integer(h, "page id")
            if h * page_size >= total or not isinstance(page, Page):
                raise ContractError("invalid page")
            if any(s >= min(page_size, total - h * page_size) for s in page.slots):
                raise ContractError("slot outside legal page tail")
            if not page.empty:
                self.pages[h] = page

    @classmethod
    def from_tids(cls, total: int, tids, page_size: int = 3,
                  recheck: bool = False) -> Bitmap:
        integer(total, "total")
        integer(page_size, "page_size", 1)
        pages: dict[int, set[int]] = {}
        for tid in tids:
            integer(tid, "TID")
            if tid >= total:
                raise ContractError("TID outside fixed snapshot")
            h, s = divmod(tid, page_size)
            pages.setdefault(h, set()).add(s)
        return cls(total, page_size, {h: Page(frozenset(s), recheck=recheck) for h, s in pages.items()})

    def candidates(self) -> set[int]:
        out = set()
        for h, page in self.pages.items():
            slots = range(min(self.page_size, self.total - h * self.page_size)) if page.lossy else page.slots
            out.update(h * self.page_size + s for s in slots)
        return out

    def combine(self, other: Bitmap, op: str) -> Bitmap:
        if (self.total, self.page_size) != (other.total, other.page_size):
            raise ContractError("bitmap snapshot universe/layout mismatch")
        if op not in ("and", "or"):
            raise ContractError("unsupported combination")
        return Bitmap(self.total, self.page_size, {
            h: combine_page(self.pages.get(h, Page()), other.pages.get(h, Page()), op)
            for h in self.pages.keys() | other.pages.keys()})

    def lossify(self, page_id: int) -> Bitmap:
        integer(page_id, "page id")
        pages = dict(self.pages)
        if page_id not in pages:
            raise ContractError("cannot lossify an absent page")
        pages[page_id] = Page(lossy=True, recheck=True)
        return Bitmap(self.total, self.page_size, pages)

    def units(self) -> int:
        return sum(1 if page.lossy else 3 for page in self.pages.values())

    def fit(self, budget: int, first: tuple[int, ...] = (1,)) -> Bitmap:
        integer(budget, "budget")
        if len(self.pages) > budget:
            raise CapacityExceeded("even all lossy pages exceed the bitmap budget")
        result = Bitmap(self.total, self.page_size, self.pages)
        order = list(dict.fromkeys(first + tuple(sorted(self.pages))))
        for h in order:
            if result.units() <= budget:
                break
            if h in result.pages and not result.pages[h].lossy:
                result = result.lossify(h)
        check(result.units() <= budget, "fit must honor budget")
        return result


def build_path(rows: tuple[Row, ...], column: str, fail_at: int | None = None) -> Bitmap:
    # Complete construction precedes publication. A failure returns no bitmap.
    if column not in ("x", "y"):
        raise ContractError("unknown index path")
    tids = []
    for i, row in enumerate(rows):
        if i == fail_at:
            raise ExecutionFailure("InjectedIndexReadError")
        if truth_eq_one(getattr(row, column)) is True:
            tids.append(i)
    return Bitmap.from_tids(len(rows), tids)


def heap_recheck(bitmap: Bitmap, rows: tuple[Row, ...], op: str = "and",
                 visible: set[int] | None = None,
                 fail_page: int | None = None) -> tuple[list, dict]:
    if len(rows) != bitmap.total:
        raise ContractError("rows/bitmap snapshot mismatch")
    if op not in ("and", "or"):
        raise ContractError("unknown predicate")
    visible = set(range(len(rows))) if visible is None else set(visible)
    if any(type(i) is not int or not 0 <= i < len(rows) for i in visible):
        raise ContractError("visible TID outside snapshot")
    out, predicate_checks = [], 0
    for i in sorted(bitmap.candidates()):
        if i // bitmap.page_size == fail_page:
            raise ExecutionFailure("InjectedHeapReadError")
        if i in visible:
            predicate_checks += 1
            if qualifies(rows[i], op):
                out.append((i, rows[i].v))
    return out, {"heap_pages": len(bitmap.pages), "candidate_visits": len(bitmap.candidates()),
                 "full_predicate_checks": predicate_checks}


def expect_error(kind, function: Callable) -> None:
    try:
        function()
    except kind:
        return
    raise AssertionError(f"expected {kind.__name__}")


def all_cuts(n: int):
    if n == 0:
        yield ()
        return
    for mask in range(1 << (n - 1)):
        cuts = [0] + [j for j in range(1, n) if mask & (1 << (j - 1))] + [n]
        yield tuple(b - a for a, b in zip(cuts, cuts[1:]))


def selftest() -> dict:
    truth = scalar(ROWS)
    with sqlite3.connect(":memory:") as connection:
        connection.execute("CREATE TABLE r(i INTEGER, x INTEGER, y INTEGER, v INTEGER)")
        connection.executemany("INSERT INTO r VALUES (?,?,?,?)", [(i, r.x, r.y, r.v) for i, r in enumerate(ROWS)])
        for op in ("and", "or"):
            actual = connection.execute(f"SELECT i,v FROM r WHERE x=1 {op} y=1 ORDER BY i").fetchall()
            check(actual == scalar(ROWS, op), "independent SQLite fixed-data oracle")
        check(connection.execute("SELECT SUM(v) FROM r WHERE 0").fetchone() == (None,), "SQL empty aggregate oracle")
    check(truth == [(0, 10), (1, 10), (5, 50), (7, 70), (10, 100)], "fixed endpoint")
    cuts = 0
    for chunks in all_cuts(12):
        out, _ = run_batch(ROWS, max(chunks), chunks)
        check(out == truth, "every continuous cut must preserve occurrences")
        cuts += 1
    values = tuple(Row(x, y, v) for x in (0, 1, None) for y in (0, 1, None) for v in (0, 7))
    batch_cases = 0
    for n in range(4):
        for rows in itertools.product(values, repeat=n):
            for k in (1, 2, 4):
                for op in ("and", "or"):
                    out, _ = run_batch(rows, k, op=op)
                    check(out == scalar(rows, op), "small batch scalar equivalence")
                    batch_cases += 1
    check(run_batch(ROWS, 1)[1]["empty_data_batches"] == 7, "empty DATA not EOF")
    check(sql_sum([]) is None, "empty SQL SUM must finalize to NULL")
    check(Counter(v for _, v in truth)[10] == 2, "bag copies")
    for transition in ("next", "error", "end", "close"):
        empty_source = BatchSource((Row(0, 0, 7),), 1,
                                   fail_at=1 if transition == "error" else None)
        empty = filter_batch(empty_source.next_batch().batch)
        check(empty.snapshot() == [], "live empty batch")
        if transition == "close":
            empty_source.close()
        else:
            empty_source.next_batch()
        expect_error(StaleBatch, empty.snapshot)
        empty_source.close()
    source = BatchSource(ROWS, 5)
    first = source.next_batch().batch
    saved = first.snapshot()
    source.next_batch()
    expect_error(StaleBatch, first.snapshot)
    tail = source.next_batch().batch
    check(tail.n == 2 and filter_batch(tail).snapshot() == [(10, 100)], "short tail")
    expect_error(ContractError, lambda: tail.get("v", 2))
    source.close()
    source.close()
    expect_error(StaleBatch, tail.snapshot)
    expect_error(ContractError, source.next_batch)
    check(saved[:2] == [(0, 10), (1, 10)], "owning snapshot survives")
    failed = BatchSource(ROWS, 4, fail_at=4)
    old = failed.next_batch().batch
    check(failed.next_batch().tag == "ERROR" and failed.next_batch().tag == "ERROR", "sticky failure")
    expect_error(StaleBatch, old.snapshot)
    failed.close()
    expect_error(ExecutionFailure, lambda: run_batch(ROWS, 4, fail_at=4))
    trap = BatchSource((Row(1, 1, 1), Row(0, 0, 0)), 2)
    b = trap.next_batch().batch.select((0,))
    check(b.map_active("v", lambda d: 10 // d) == [10], "active-only trap avoidance")
    expect_error(ZeroDivisionError, lambda: [10 // trap.columns["v"][j] for j in range(2)])
    trap.close()
    a, b = build_path(ROWS, "x"), build_path(ROWS, "y")
    exact = a.combine(b, "and")
    check(exact.candidates() == {0, 1, 5, 7, 10}, "exact AND")
    union = a.combine(b, "or")
    check(len(union.candidates()) == 10, "OR unique TIDs")
    out, _ = heap_recheck(union, ROWS, "or")
    check(sql_sum([v for _, v in out]) == 520, "OR sum")
    check(len(a.candidates()) + len(b.candidates()) == 15, "bad concatenation witness")
    check(sum(ROWS[i].v for i in a.candidates()) + sum(ROWS[i].v for i in b.candidates()) == 760, "bad concatenation sum")
    lossy = exact.fit(10)
    check(lossy.candidates() == {0, 1, 3, 4, 5, 7, 10}, "budget-driven page1 lossification")
    check(lossy.units() == 10 and heap_recheck(lossy, ROWS)[0] == truth, "lossy restores exact answer")
    check(exact.fit(4).candidates() == set(range(12)), "all lossy visits all slots")
    expect_error(CapacityExceeded, lambda: exact.fit(3))
    for budget in range(4, 13):
        candidate = exact.fit(budget)
        check(exact.candidates() <= candidate.candidates(), "lossification never loses candidates")
        check(heap_recheck(candidate, ROWS)[0] == truth, "budget migration")
    mixed = a.lossify(1).combine(b, "and")
    check(mixed.pages[1].slots == frozenset({1, 2}) and mixed.pages[1].recheck,
          "exact positions can still require recheck")
    check(heap_recheck(mixed, ROWS)[0] == truth, "mixed input recheck")
    check(2 + len(exact.pages) == 6 and len(Bitmap.from_tids(12, [0, 1]).pages) + 2 == 3,
          "separate index and heap I/O")
    # Every one-page true set and candidate superset, including exact positions
    # with recheck, plus lossy encoding. This is independent of fixed rows.
    universe = frozenset(range(3))
    sets = [frozenset(i for i in range(3) if mask & (1 << i)) for mask in range(8)]
    combinations = 0
    for ta, tb in itertools.product(sets, repeat=2):
        reps_a = [Page(c, recheck=(c != ta)) for c in sets if ta <= c] + [Page(lossy=True, recheck=True)]
        reps_b = [Page(c, recheck=(c != tb)) for c in sets if tb <= c] + [Page(lossy=True, recheck=True)]
        for pa, pb, op in itertools.product(reps_a, reps_b, ("and", "or")):
            result = combine_page(pa, pb, op)
            ca, cb = universe if pa.lossy else pa.slots, universe if pb.lossy else pb.slots
            target = ta & tb if op == "and" else ta | tb
            expected = ca & cb if op == "and" else ca | cb
            actual = universe if result.lossy else result.slots
            check(actual == expected and target <= actual, "page algebra and no false negatives")
            check(actual & target == target, "full recheck equality")
            if not result.recheck:
                check(actual <= target, "false recheck flag is a predicate certificate")
            combinations += 1
    visibility_cases = 0
    for mask in range(1 << 12):
        visible = {i for i in range(12) if mask & (1 << i)}
        for op in ("and", "or"):
            bitmap = a.combine(b, op).fit(4)
            got, _ = heap_recheck(bitmap, ROWS, op, visible)
            expected = [(i, v) for i, v in scalar(ROWS, op) if i in visible]
            check(got == expected, "snapshot visibility distinct from index evidence")
            visibility_cases += 1
    expect_error(ExecutionFailure, lambda: build_path(ROWS, "x", fail_at=4))
    expect_error(ExecutionFailure, lambda: heap_recheck(exact, ROWS, fail_page=1))
    for bad in (True, -1, 0, 1.5):
        expect_error(ContractError, lambda bad=bad: BatchSource(ROWS, bad))
    for bad in (True, -1, 1.5):
        expect_error(ContractError, lambda bad=bad: exact.fit(bad))
    expect_error(ContractError, lambda: Bitmap.from_tids(12, [12]))
    expect_error(ContractError, lambda: Bitmap.from_tids(12, [True]))
    expect_error(ContractError, lambda: a.combine(Bitmap(13), "and"))
    expect_error(ContractError, lambda: a.combine(b, "not"))
    expect_error(ContractError, lambda: Bitmap(4, 3, {1: Page(frozenset({1}))}))
    descending_payload = [120 - 10 * i for i, _ in truth]
    check(descending_payload != sorted(descending_payload), "heap order not ORDER BY")
    return {"status": "passed", "continuous_partitions": cuts,
            "small_batch_cases": batch_cases, "page_algebra_cases": combinations,
            "visibility_cases": visibility_cases, "checks_enabled_under_optimized_python": True, "sqlite_fixed_query_oracles": 3}


def positive_csv(text: str) -> list[int]:
    try:
        result = [int(x) for x in text.split(",")]
    except ValueError as error:
        raise argparse.ArgumentTypeError("comma-separated positive integers required") from error
    if not result or any(x <= 0 for x in result):
        raise argparse.ArgumentTypeError("positive batch sizes required")
    return result


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--batch-sizes", type=positive_csv, default=[1, 2, 4, 5, 12])
    parser.add_argument("--bitmap-budget", type=int, default=10)
    args = parser.parse_args()
    try:
        a, b = build_path(ROWS, "x"), build_path(ROWS, "y")
        exact = a.combine(b, "and")
        limited = exact.fit(args.bitmap_budget)
        output, costs = heap_recheck(limited, ROWS)
        tests = selftest()
        batches = {}
        for k in args.batch_sizes:
            answer, stats = run_batch(ROWS, k)
            check(answer == output, "requested batch agrees with bitmap result")
            batches[str(k)] = stats
        report = {"tests": tests, "occurrences": [i for i, _ in output],
                  "values": [v for _, v in output], "sql_sum": sql_sum([v for _, v in output]),
                  "batch_runs": batches, "bitmap_budget": args.bitmap_budget,
                  "bitmap_units": limited.units(), "candidates": sorted(limited.candidates()),
                  "bitmap_costs": {"index_page_reads_assumed": 2, **costs,
                                   "total_input_page_reads": 2 + costs["heap_pages"],
                                   "full_heap_scan_pages": 4},
                  "logical_bytes_model": {"field_width_assumed": 8, "predicate_field_bytes": 24 * 8,
                                          "selected_v_bytes": 5 * 8, "copied_id_v_bytes": 5 * 16},
                  "limits": ["no wall-clock performance claim", "no actual DBMS/MVCC execution",
                             "budget constrains one returned bitmap, not peak build/execution memory",
                             "A/B/exact/limited and temporary copies are simultaneously live and separately chargeable",
                             "logical budget units exclude fixed control area and Python overhead",
                             "finite tests complement, not replace, the article proofs"]}
        print(json.dumps(report, ensure_ascii=False, indent=2))
    except (ContractError, ExecutionFailure) as error:
        parser.exit(2, f"{type(error).__name__}: {error}\n")


if __name__ == "__main__":
    main()