---
title: "Forge Heat Cycles"
description: "Forge heat-cycle registry API: B-tree indexes, chained risk UDFs, and parallel shop-floor writers"
url: "https://www.pixeltable.com/recipes/forge-heats"
source: "https://github.com/pierrebrunelle/pixeltable-forge-heats"
---

# Forge Heat Cycles

Forge heat-cycle registry API: B-tree indexes, chained risk UDFs, and parallel shop-floor writers

One `app.py` with 1 table, 4 computed columns, 8 HTTP routes. Needs `pixeltable[serve]>=0.7.16`.

## Tables

### heats (table)

| Column | Kind | Type or expression |
| --- | --- | --- |
| heat_id | primary key | `pxtf.uuid.uuid7()` |
| forge_id | stored | String |
| alloy | stored | String |
| status | stored | String |
| peak_c | stored | Float |
| hold_min | stored | Float |
| quench | stored | String |
| operator | stored | String (optional) |
| note | stored | String (optional) |
| band | computed on insert | `heat_band(peak_c)` |
| risk | computed on insert | `risk_score(peak_c, hold_min, quench)` |
| tag | computed on insert | `heat_tag(band, risk, alloy)` |
| lane | computed on insert | `priority_lane(risk, status)` |

- B-tree index on `forge_id`
- B-tree index on `alloy`
- B-tree index on `status`
- B-tree index on `band`

## Routes (service `forge_api`)

- `POST /heats`: inserts into `heats` (takes forge_id, alloy, status, peak_c, hold_min, quench, operator, note; returns heat_id, band, risk, tag, lane)
- `POST /heats/status`: updates `heats` (takes status, note; returns heat_id, status, lane, tag)
- `POST /score`: computes `heats` columns without storing (takes peak_c, hold_min, quench, alloy, status; returns band, risk, tag, lane)
- `POST /heats/delete`: deletes from `heats`
- `GET /heats/by-forge`: runs the query `by_forge` (takes forge_id)
- `GET /heats/by-band`: runs the query `by_band_status` (takes band, status)
- `GET /heats/by-alloy`: runs the query `by_alloy` (takes alloy)
- `GET /heats/hot-lane`: runs the query `hot_lane`

## Run it locally

```sh
git clone https://github.com/pierrebrunelle/pixeltable-forge-heats
cd pixeltable-forge-heats
uv sync
uv run pxt schema update app.py forge
uv run python seed.py forge
uv run pxt service run app.py forge --port 8000
```

## Run it on Pixeltable Cloud

```sh
uv run pxt new    # a free database, no account
# put the pxt:// address it prints in pixeltable.toml
uv run pxt db update pxt://<org>:main -f
uv run pxt schema update app.py pxt://<org>:main -f
uv run pxt service update app.py pxt://<org>:main -f
```

## app.py

At commit `d4529667` (https://github.com/pierrebrunelle/pixeltable-forge-heats/blob/d452966710e90f38ba6eba61bfd338d9ad8f1dad/app.py):

```python
"""Forge Heat Cycle Registry API built with Pixeltable.

Log a forge heat with forge id, alloy, peak temperature, hold time and quench.
Pixeltable derives a heat band, risk score, combined heat tag and priority lane
via a UDF chain, and B-tree indexes on forge, alloy, status and band back the
shop-floor lookup queries. The client demo exercises parallel writers.

Layout: UDFs live in udfs.py; this file holds table models, @pxt.query functions and routes.

Run locally:
    pxt schema update app.py forge
    pxt service run app.py forge
"""
import pixeltable as pxt
import pixeltable.functions as pxtf
from pixeltable.serving import FastAPIRouter

from udfs import heat_band, heat_tag, priority_lane, risk_score

# ---- tables ----
TableModel = pxt.model_base()


class Heats(TableModel, name='heats', has_default_idxs=False):
    heat_id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
    forge_id: pxt.String
    alloy: pxt.String
    status: pxt.String  # queued / firing / quench / cool / done / hold / scrapped
    peak_c: pxt.Float
    hold_min: pxt.Float
    quench: pxt.String  # air / oil / water / brine
    operator: pxt.String | None
    note: pxt.String | None

    band = heat_band(peak_c)
    risk = risk_score(peak_c, hold_min, quench)
    tag = heat_tag(band, risk, alloy)
    lane = priority_lane(risk, status)

    __indexes__ = [
        pxt.BtreeIndex(forge_id),
        pxt.BtreeIndex(alloy),
        pxt.BtreeIndex(status),
        pxt.BtreeIndex(band),
    ]


# ---- queries ----
@pxt.query
def by_forge(forge_id: str):
    """Heats on one forge (forge_id index), highest risk first."""
    return (
        Heats.where(Heats.forge_id == forge_id)
        .select(Heats.heat_id, Heats.alloy, Heats.status, Heats.peak_c, Heats.band, Heats.risk, Heats.tag, Heats.lane)
        .order_by(Heats.risk, asc=False)
    )


@pxt.query
def by_band_status(band: str, status: str):
    """Heats in a heat band and status (band + status indexes)."""
    return (
        Heats.where((Heats.band == band) & (Heats.status == status))
        .select(Heats.heat_id, Heats.forge_id, Heats.alloy, Heats.peak_c, Heats.risk, Heats.tag, Heats.lane)
        .order_by(Heats.peak_c, asc=False)
    )


@pxt.query
def by_alloy(alloy: str):
    """All heats for an alloy (alloy index)."""
    return (
        Heats.where(Heats.alloy == alloy)
        .select(Heats.heat_id, Heats.forge_id, Heats.status, Heats.band, Heats.risk, Heats.tag)
        .order_by(Heats.forge_id)
    )


@pxt.query
def hot_lane():
    """Shop board: heats currently in the hot priority lane."""
    return (
        Heats.where(Heats.lane == 'hot')
        .select(Heats.heat_id, Heats.forge_id, Heats.alloy, Heats.status, Heats.risk, Heats.tag)
        .order_by(Heats.risk, asc=False)
    )


# ---- routes ----
forge_api = FastAPIRouter(name='forge_api')

forge_api.add_insert_route(
    Heats,
    path='/heats',
    inputs=[Heats.forge_id, Heats.alloy, Heats.status, Heats.peak_c, Heats.hold_min, Heats.quench, Heats.operator, Heats.note],
    outputs=[Heats.heat_id, Heats.band, Heats.risk, Heats.tag, Heats.lane],
)
forge_api.add_update_route(
    Heats,
    path='/heats/status',
    inputs=[Heats.status, Heats.note],
    outputs=[Heats.heat_id, Heats.status, Heats.lane, Heats.tag],
)
forge_api.add_compute_route(
    Heats,
    path='/score',
    inputs=[Heats.peak_c, Heats.hold_min, Heats.quench, Heats.alloy, Heats.status],
    outputs=[Heats.band, Heats.risk, Heats.tag, Heats.lane],
)
forge_api.add_delete_route(Heats, path='/heats/delete')
forge_api.add_query_route(path='/heats/by-forge', query=by_forge, method='get')
forge_api.add_query_route(path='/heats/by-band', query=by_band_status, method='get')
forge_api.add_query_route(path='/heats/by-alloy', query=by_alloy, method='get')
forge_api.add_query_route(path='/heats/hot-lane', query=hot_lane, method='get')
```