Blame for training/detector_data.py

"""Data for the bold-line intersection detector (CenterNet style) on v5 canvases.

Ground truth: bold 5 mm lines lie at U = ou + 5i, V = ov + 5j; with a = (U - ou) / 5, b = (V - ov) / 5
from the dense uv map, an intersection is found in every 2 x 2 pixel cell by one Newton step. Nodes
inside the grid area are kept; a border lying on a bold line keeps its nodes, at other borders nodes
closer than MARGIN_TOL_MM are dropped.
Windows: mostly >= 44 of 49 patches on the sheet; EDGE_WINDOW_PROB of them show the sheet border.
Occlusions: 0-2 blobs (blot, blur, noise, glare); the targets stay unchanged.
Targets: 28 x 28 cells of 8 px: heat (Gaussian around each node), offset inside the cell, node flag.
"""

import math

import cv2
import numpy as np

import canvas
import crops

CROP_SIZE = crops.CROP_SIZE
CELL = 8
CELLS = CROP_SIZE // CELL
HEAT_SIGMA = 0.7
EDGE_WINDOW_PROB = 0.3
EDGE_MIN_VALID = 20
MARGIN_TOL_MM = 0.3
OCCLUSION_COUNT_P = (0.2, 0.5, 0.3)
OCCLUSION_AREA_PATCHES = (0.6, 2.2)
OCCLUSION_KINDS = ("blot", "blur", "noise", "glare")
PLAIN_KEEP_EVERY = 3                     # train: every 3rd "plain_1mm" sheet (answer: no nodes)
KIND_CODES = {"red": 0, "dots": 1, "plain_1mm": 2}
CH_HEAT, CH_OFF_X, CH_OFF_Y, CH_POS = range(4)


def canvas_points(uv_mm, valid, texture_info):
    """Bold-line intersections in a canvas. -> [K, 2] x, y (canvas px)."""
    grid = texture_info["grid"]
    if grid["kind"] == "plain_1mm":
        return np.zeros((0, 2))
    ou, ov = grid["bold_line_mm"]
    a = (uv_mm[..., 0].astype(np.float64) - ou) / 5.0
    b = (uv_mm[..., 1].astype(np.float64) - ov) / 5.0
    ok = valid[:-1, :-1] & valid[:-1, 1:] & valid[1:, :-1] & valid[1:, 1:]
    ac = 0.25 * (a[:-1, :-1] + a[:-1, 1:] + a[1:, :-1] + a[1:, 1:])
    bc = 0.25 * (b[:-1, :-1] + b[:-1, 1:] + b[1:, :-1] + b[1:, 1:])
    ax = 0.5 * (a[:-1, 1:] - a[:-1, :-1] + a[1:, 1:] - a[1:, :-1])
    ay = 0.5 * (a[1:, :-1] - a[:-1, :-1] + a[1:, 1:] - a[:-1, 1:])
    bx = 0.5 * (b[:-1, 1:] - b[:-1, :-1] + b[1:, 1:] - b[1:, :-1])
    by = 0.5 * (b[1:, :-1] - b[:-1, :-1] + b[1:, 1:] - b[:-1, 1:])
    det = ax * by - ay * bx
    i, j = np.round(ac), np.round(bc)
    da, db = i - ac, j - bc
    with np.errstate(divide="ignore", invalid="ignore"):
        dx = (by * da - ay * db) / det
        dy = (-bx * da + ax * db) / det
    hit = ok & (np.abs(det) > 1e-9) & (np.abs(dx) <= 0.5) & (np.abs(dy) <= 0.5)
    ys, xs = np.nonzero(hit)
    if len(ys) == 0:
        return np.zeros((0, 2))
    px = xs + 0.5 + dx[ys, xs]
    py = ys + 0.5 + dy[ys, xs]
    ii, jj = i[ys, xs], j[ys, xs]
    u, v = ou + 5.0 * ii, ov + 5.0 * jj
    x0, y0, x1, y1 = grid["area_mm"]
    bold = grid["edges_bold"]
    width_mm, height_mm = texture_info["size_mm"]
    t = MARGIN_TOL_MM
    lo_u = max(x0 - t if bold[0] else x0 + t, t)
    lo_v = max(y0 - t if bold[1] else y0 + t, t)
    hi_u = min(x1 + t if bold[2] else x1 - t, width_mm - t)
    hi_v = min(y1 + t if bold[3] else y1 - t, height_mm - t)
    inside = (u > lo_u) & (u < hi_u) & (v > lo_v) & (v < hi_v)
    keys = np.stack([ii, jj], axis=1)[inside]
    if len(keys) == 0:
        return np.zeros((0, 2))
    # a node on the border of two cells is found twice: average
    _, group = np.unique(keys, axis=0, return_inverse=True)
    group = group.reshape(-1)
    sums = np.zeros((group.max() + 1, 2))
    np.add.at(sums, group, np.stack([px, py], axis=1)[inside])
    return sums / np.bincount(group)[:, None]


def points_to_crop(points, window):
    x0, y0, s = window
    x = (points[:, 0] + 0.5 - x0) / s - 0.5
    y = (points[:, 1] + 0.5 - y0) / s - 0.5
    keep = (x >= -0.5) & (x < CROP_SIZE - 0.5) & (y >= -0.5) & (y < CROP_SIZE - 0.5)
    return np.stack([x, y], axis=1)[keep]


def _blob_mask(rng, size):
    """Soft rotated ellipse with a wobbly border, 0..1."""
    area = rng.uniform(*OCCLUSION_AREA_PATCHES) * crops.PATCH_SIZE ** 2
    aspect = rng.uniform(1.0, 2.5)
    ra = math.sqrt(area * aspect / math.pi)
    rb = ra / aspect
    cx, cy = rng.uniform(0.1, 0.9) * size, rng.uniform(0.1, 0.9) * size
    angle = rng.uniform(0, math.pi)
    soft = rng.uniform(0.8, 3.0)
    reach = int(math.ceil(1.4 * ra + 3 * soft + 2))
    xa, xb = max(0, int(cx) - reach), min(size, int(cx) + reach + 1)
    ya, yb = max(0, int(cy) - reach), min(size, int(cy) + reach + 1)
    yy, xx = np.mgrid[ya:yb, xa:xb].astype(np.float32)
    xr = (xx - cx) * math.cos(angle) + (yy - cy) * math.sin(angle)
    yr = -(xx - cx) * math.sin(angle) + (yy - cy) * math.cos(angle)
    radius = np.sqrt((xr / ra) ** 2 + (yr / rb) ** 2)
    coarse = rng.normal(0, 1, (6, 6)).astype(np.float32)
    wobble = cv2.resize(coarse, (xb - xa, yb - ya), interpolation=cv2.INTER_CUBIC) * rng.uniform(0.1, 0.3)
    mask = np.zeros((size, size), np.float32)
    mask[ya:yb, xa:xb] = (radius + wobble < 1.0).astype(np.float32)
    return cv2.GaussianBlur(mask, (0, 0), soft)


def _smooth_field(rng, size, cells, amplitude):
    coarse = rng.normal(0, 1, (cells, cells)).astype(np.float32)
    return cv2.resize(coarse, (size, size), interpolation=cv2.INTER_CUBIC) * amplitude


def occlude(rng, crop_bgr):
    """0-2 occlusion blobs."""
    size = crop_bgr.shape[0]
    count = int(rng.choice(len(OCCLUSION_COUNT_P), p=OCCLUSION_COUNT_P))
    image = crop_bgr.astype(np.float32)
    for _ in range(count):
        alpha = _blob_mask(rng, size)
        kind = OCCLUSION_KINDS[int(rng.integers(len(OCCLUSION_KINDS)))]
        if kind == "blot":
            color = np.array([[150, 60, 25], [30, 30, 35], [45, 85, 140], [60, 60, 170],
                              [20, 50, 90]][int(rng.integers(5))], np.float32)
            color = color * rng.uniform(0.7, 1.3)
            layer = color[None, None, :] * (1.0 + _smooth_field(rng, size, 12, 0.15))[..., None]
            layer += rng.normal(0, 6, (size, size, 1)).astype(np.float32)
            strength = rng.uniform(0.8, 1.0)
        elif kind == "blur":
            layer = cv2.GaussianBlur(image, (0, 0), rng.uniform(5.0, 12.0))
            strength = 1.0
        elif kind == "noise":
            sigma = rng.uniform(45.0, 100.0)
            grain = rng.normal(0, sigma, (size, size, 3 if rng.random() < 0.5 else 1)).astype(np.float32)
            layer = cv2.GaussianBlur(image, (0, 0), 1.5) + grain
            strength = 1.0
        else:                                   # glare: a saturated core fading out
            level = rng.uniform(235, 255)
            tint = np.array([1.0, rng.uniform(0.97, 1.0), rng.uniform(0.95, 1.0)], np.float32)
            layer = np.full_like(image, level) * tint
            alpha = np.clip(cv2.GaussianBlur(alpha, (0, 0), rng.uniform(3.0, 8.0)) * 1.4, 0, 1)
            strength = rng.uniform(0.9, 1.0)
        a = (alpha * strength)[..., None]
        image = image * (1 - a) + layer * a
    return np.clip(image, 0, 255).astype(np.uint8)


def make_targets(points_crop):
    """-> [28, 28, 4] float32: heat, offset x, offset y, node flag."""
    target = np.zeros((CELLS, CELLS, 4), np.float32)
    if len(points_crop) == 0:
        return target
    pc = (points_crop + 0.5) / CELL
    cell = np.clip(np.floor(pc).astype(int), 0, CELLS - 1)
    offset = np.clip(pc - cell, 0.0, 1.0 - 1e-6)
    grid = np.arange(CELLS)
    heat = np.zeros((CELLS, CELLS), np.float32)
    for (cx, cy) in cell:
        heat = np.maximum(heat, np.exp(-((grid[None, :] - cx) ** 2 + (grid[:, None] - cy) ** 2) / (2 * HEAT_SIGMA ** 2)))
    for (cx, cy), (ox, oy) in zip(cell, offset):
        heat[cy, cx] = 1.0
        target[cy, cx, CH_OFF_X] = ox
        target[cy, cx, CH_OFF_Y] = oy
        target[cy, cx, CH_POS] = 1.0
    target[..., CH_HEAT] = heat
    return target


def train_selection(directory, indices):
    """No "dots" sheets (their dots are invisible at 3.5-7.5 px/mm), every PLAIN_KEEP_EVERY-th "plain_1mm" one."""
    kind = {i: canvas.read_texture_info(directory, i)["grid"]["kind"] for i in indices}
    keep_plain = set([i for i in indices if kind[i] == "plain_1mm"][::PLAIN_KEEP_EVERY])
    return [i for i in indices if kind[i] == "red" or i in keep_plain]


def make_sample(rng, directory, index, cache):
    """-> (crop BGR uint8, target [28, 28, 4], grid kind code)."""
    image, uv_mm, valid = canvas.read_sample(directory, index)
    texture = canvas.read_texture_info(directory, index)
    if rng.random() < EDGE_WINDOW_PROB:
        window = crops.choose_window(rng, uv_mm, valid, image.shape[0], EDGE_MIN_VALID, crops.MIN_VALID_PATCHES)
    else:
        window = crops.choose_window(rng, uv_mm, valid, image.shape[0])
    crop = occlude(rng, crops.crop_image(image, window))
    if index not in cache:
        cache[index] = canvas_points(uv_mm, valid, texture)
    return crop, make_targets(points_to_crop(cache[index], window)), KIND_CODES[texture["grid"]["kind"]]


class PointDataset:
    """train: new random window and occlusions every call; otherwise one seeded window per index."""

    def __init__(self, directory, indices, train=True, fixed_seed=777):
        self.directory, self.indices, self.train, self.fixed_seed = directory, list(indices), train, fixed_seed
        self.rng = None
        self.cache = {}

    def __len__(self):
        return len(self.indices)

    def __getitem__(self, position):
        import torch
        index = self.indices[position]
        if self.train:
            if self.rng is None:
                self.rng = np.random.default_rng(torch.initial_seed() % (2 ** 63))
            rng = self.rng
        else:
            rng = np.random.default_rng([self.fixed_seed, index])
        crop, target, kind = make_sample(rng, self.directory, index, self.cache)
        rgb = cv2.cvtColor(crop, cv2.COLOR_BGR2RGB)
        return torch.from_numpy(np.ascontiguousarray(rgb.transpose(2, 0, 1))), torch.from_numpy(target), torch.tensor(kind)


def decode(heat_prob, offsets, threshold):
    """3 x 3 local maxima above threshold -> [K, 2] crop px."""
    padded = np.pad(heat_prob, 1, constant_values=-1.0)
    neighbours = np.max(np.stack([padded[dy:dy + CELLS, dx:dx + CELLS] for dy in range(3) for dx in range(3)]), axis=0)
    ys, xs = np.nonzero((heat_prob >= neighbours) & (heat_prob >= threshold))
    return np.stack([(xs + offsets[ys, xs, 0]) * CELL - 0.5, (ys + offsets[ys, xs, 1]) * CELL - 0.5], axis=1)


def target_points(target):
    ys, xs = np.nonzero(target[..., CH_POS] > 0.5)
    return np.stack([(xs + target[ys, xs, CH_OFF_X]) * CELL - 0.5, (ys + target[ys, xs, CH_OFF_Y]) * CELL - 0.5], axis=1)


def match_count(predicted, truth, radius):
    """Greedy one-to-one matching by distance -> number of matched pairs."""
    if len(predicted) == 0 or len(truth) == 0:
        return 0
    distance = np.hypot(predicted[:, None, 0] - truth[None, :, 0], predicted[:, None, 1] - truth[None, :, 1])
    pi, ti = np.nonzero(distance <= radius)
    used_p, used_t = set(), set()
    for k in np.argsort(distance[pi, ti]):
        if pi[k] not in used_p and ti[k] not in used_t:
            used_p.add(pi[k])
            used_t.add(ti[k])
    return len(used_p)


def f1_scores(heat_prob, offsets, targets, kinds, thresholds, radius):
    """Precision / recall / F1 on "red" crops per threshold and the best F1."""
    result, best = {}, -1.0
    for threshold in thresholds:
        tp = fp = fn = 0
        for n in np.nonzero(kinds == KIND_CODES["red"])[0]:
            predicted = decode(heat_prob[n], offsets[n], threshold)
            truth = target_points(targets[n])
            m = match_count(predicted, truth, radius)
            tp, fp, fn = tp + m, fp + len(predicted) - m, fn + len(truth) - m
        precision, recall = tp / max(tp + fp, 1), tp / max(tp + fn, 1)
        f1 = 2 * precision * recall / max(precision + recall, 1e-12)
        result["t{:02d}".format(int(round(threshold * 100)))] = (precision, recall, f1)
        best = max(best, f1)
    result["best_f1"] = best
    return result