Skip to content

Inference

Three modules with no type annotations, and the three .ppy files ppy convert wrote from them. Each exercises one thing the inference has to get right. examples/verify_conversions.py regenerates each converted file to prove it is what the converter produces.

Run it

ppy convert pipeline.py --dry-run
python  pipeline.py
ppy run pipeline.ppy

What it prints

ppy convert pipeline.py --dry-run

74 lines
# ---- ./pipeline.ppy ----
import math
from collections.abc import Sequence

import ppy


class Summary:
    def __init__(self, mean: float, deviation: float, count: int) -> None:
        self.mean: float = mean
        self.deviation: float = deviation
        self.count: int = count

    def scaled(self, factor: float) -> 'Summary':
        return Summary(self.mean * factor, self.deviation * factor, self.count)

    def shifted(self, offset: float) -> 'Summary':
        return Summary(self.mean + offset, self.deviation, self.count)

    @ppy.pure
    def describe(self) -> str:
        return f"n={self.count} mean={self.mean:.3f} sd={self.deviation:.3f}"


@ppy.pure
def clamp(value: float, low: float, high: float) -> float:
    if value < low:
        return low
    if value > high:
        return high
    return value


@ppy.pure
def normalize(value: float, mean: float, spread: float) -> float:
    return clamp((value - mean) / spread, -3.0, 3.0)


def summarize(readings: Sequence[float]) -> Summary | None:
    count: int = len(readings)
    if not count:
        return None
    total: float = 0.0
    for i in range(count):
        total += readings[i]
    mean: float = total / count
    spread: float = 0.0
    for i in range(count):
        spread += (readings[i] - mean) * (readings[i] - mean)
    return Summary(mean, math.sqrt(spread / count), count)


def standardized(readings: Sequence[float], summary: Summary) -> list[float]:
    out: list[float] = []
    for reading in readings:
        out.append(normalize(reading, summary.mean, summary.deviation))
    return out


def report(readings: Sequence[float]) -> str:
    summary = summarize(readings)
    if summary is None:
        return "empty"
    values: list[float] = standardized(readings, summary)
    return summary.describe() + " first=" + str(round(values[0], 4))


samples: list[float] = [2.0, 4.0, 4.0, 4.0, 5.0, 5.0, 7.0, 9.0]
print(report(samples))
print(report([]))

overall: Summary | None = summarize(samples)
if overall is not None:
    print(overall.scaled(2.0).describe())

python pipeline.py, ppy run pipeline.ppy

n=8 mean=5.000 sd=2.000 first=-1.5
empty
n=8 mean=10.000 sd=4.000

stats.py: types flow along the call graph

Nothing is annotated. mean, variance, standardize, and report all get parameter and return types from one module-level samples list of floats:

  • the call site types values,
  • the arithmetic types the return,
  • the next call carries it on.

The parameters are Sequence[float] rather than list[float] because every body only reads.

shapes.py: fields from __init__, and None as a real case

@ppy.pure
def widest(rects: Sequence[Rect]) -> Rect | None:
    best: Rect | None = None
    for rect in rects:
        if best is None or rect.width > best.width:
            best = rect
    return best

Rect.width and Rect.height are typed from what __init__ assigns. widest is defined before Rect, so its annotation is quoted as a forward reference. It returns Rect | None because one path returns best's initial value, and label accepts that union and narrows it.

pipeline.py: everything at once

pipeline.py combines the cases:

  • a call chain several functions deep,
  • instance fields from __init__,
  • a class used before it is defined,
  • a list whose element type comes from what is appended,
  • shifted, which nothing calls, typed from the arithmetic in its body,
  • @ppy.pure where the checker proved it.

The converter moved Summary above summarize because the move is provably safe. With --hoist-classes=off it would have kept the quoted forward reference.

Limitations

The converter will not rename, split a function that does several jobs, or restructure an algorithm. When a buffer promotion is blocked it names the blocker:

remark[R3003]: `raw` could be a borrowed buffer, but `raw` is sliced, which
               copies; indexing it element by element instead would let the
               memory be borrowed

Where the code comes from

Generated, not hand-written: pipeline.ppy, shapes.ppy, and stats.ppy are exactly what ppy convert writes from the .py beside each, and examples/verify_conversions.py checks that on every run.

Read on: Conversion and inference ยท Inventory

23_inference/pipeline.ppy

import math
from collections.abc import Sequence

import ppy


class Summary:
    def __init__(self, mean: float, deviation: float, count: int) -> None:
        self.mean: float = mean
        self.deviation: float = deviation
        self.count: int = count

    def scaled(self, factor: float) -> 'Summary':
        return Summary(self.mean * factor, self.deviation * factor, self.count)

    def shifted(self, offset: float) -> 'Summary':
        return Summary(self.mean + offset, self.deviation, self.count)

    @ppy.pure
    def describe(self) -> str:
        return f"n={self.count} mean={self.mean:.3f} sd={self.deviation:.3f}"


@ppy.pure
def clamp(value: float, low: float, high: float) -> float:
    if value < low:
        return low
    if value > high:
        return high
    return value


@ppy.pure
def normalize(value: float, mean: float, spread: float) -> float:
    return clamp((value - mean) / spread, -3.0, 3.0)


def summarize(readings: Sequence[float]) -> Summary | None:
    count: int = len(readings)
    if not count:
        return None
    total: float = 0.0
    for i in range(count):
        total += readings[i]
    mean: float = total / count
    spread: float = 0.0
    for i in range(count):
        spread += (readings[i] - mean) * (readings[i] - mean)
    return Summary(mean, math.sqrt(spread / count), count)


def standardized(readings: Sequence[float], summary: Summary) -> list[float]:
    out: list[float] = []
    for reading in readings:
        out.append(normalize(reading, summary.mean, summary.deviation))
    return out


def report(readings: Sequence[float]) -> str:
    summary = summarize(readings)
    if summary is None:
        return "empty"
    values: list[float] = standardized(readings, summary)
    return summary.describe() + " first=" + str(round(values[0], 4))


samples: list[float] = [2.0, 4.0, 4.0, 4.0, 5.0, 5.0, 7.0, 9.0]
print(report(samples))
print(report([]))

overall: Summary | None = summarize(samples)
if overall is not None:
    print(overall.scaled(2.0).describe())

23_inference/shapes.ppy

from collections.abc import Sequence

import ppy


class Rect:
    def __init__(self, width: float, height: float) -> None:
        self.width: float = width
        self.height: float = height

    @ppy.pure
    def area(self) -> float:
        return self.width * self.height

    def scaled(self, factor: float) -> 'Rect':
        return Rect(self.width * factor, self.height * factor)


@ppy.pure
def widest(rects: Sequence[Rect]) -> Rect | None:
    best: Rect | None = None
    for rect in rects:
        if best is None or rect.width > best.width:
            best = rect
    return best


@ppy.pure
def total_area(rects: Sequence[Rect]) -> float:
    total: float = 0.0
    for rect in rects:
        total += rect.area()
    return total


@ppy.pure
def label(rect: Rect | None, prefix: str) -> str:
    if rect is None:
        return prefix + ":none"
    return f"{prefix}:{rect.width}x{rect.height}"


def build(count: int) -> list[Rect]:
    out: list[Rect] = []
    for i in range(count):
        out.append(Rect(float(i), float(i) * 2.0))
    return out


boxes: list[Rect] = build(4)
print(total_area(boxes))
print(label(widest(boxes), "max"))
print(label(widest([]), "none") if widest([]) is None else 0.0)

23_inference/stats.ppy

from collections.abc import Sequence

import ppy


@ppy.pure
def mean(values: Sequence[float]) -> float:
    total: float = 0.0
    for value in values:
        total += value
    return total / len(values)


@ppy.pure
def variance(values: Sequence[float]) -> float:
    m: float = mean(values)
    total: float = 0.0
    for value in values:
        total += (value - m) * (value - m)
    return total / len(values)


def standardize(values: Sequence[float]) -> list[float]:
    m: float = mean(values)
    spread: float = variance(values) ** 0.5
    out: list[float] = []
    for value in values:
        out.append((value - m) / spread)
    return out


@ppy.pure
def report(values: Sequence[float], label: str) -> str:
    return f"{label}: mean={mean(values):.3f} var={variance(values):.3f}"


samples: list[float] = [1.0, 2.0, 3.0, 4.0, 5.0]
print(report(samples, "samples"), [round(x, 4) for x in standardize(samples)])

Source: examples/23_inference.