Forge Heat Cycles

1"""Forge Heat Cycle Registry API built with Pixeltable.
2
3Log a forge heat with forge id, alloy, peak temperature, hold time and quench.
4Pixeltable derives a heat band, risk score, combined heat tag and priority lane
5via a UDF chain, and B-tree indexes on forge, alloy, status and band back the
6shop-floor lookup queries. The client demo exercises parallel writers.
7
8Layout: UDFs live in udfs.py; this file holds table models, @pxt.query functions and routes.
9
10Run locally:
11    pxt schema update app.py forge
12    pxt service run app.py forge
13"""
14import pixeltable as pxt
15import pixeltable.functions as pxtf
16from pixeltable.serving import FastAPIRouter
17
18from udfs import heat_band, heat_tag, priority_lane, risk_score
19
20# ---- tables ----
21TableModel = pxt.model_base()
22
23
24class Heats(TableModel, name='heats', has_default_idxs=False):
25    heat_id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
26    forge_id: pxt.String
27    alloy: pxt.String
28    status: pxt.String  # queued / firing / quench / cool / done / hold / scrapped
29    peak_c: pxt.Float
30    hold_min: pxt.Float
31    quench: pxt.String  # air / oil / water / brine
32    operator: pxt.String | None
33    note: pxt.String | None
34
35    band = heat_band(peak_c)
36    risk = risk_score(peak_c, hold_min, quench)
37    tag = heat_tag(band, risk, alloy)
38    lane = priority_lane(risk, status)
39
40    __indexes__ = [
41        pxt.BtreeIndex(forge_id),
42        pxt.BtreeIndex(alloy),
43        pxt.BtreeIndex(status),
44        pxt.BtreeIndex(band),
45    ]
46
47
48# ---- queries ----
49@pxt.query
50def by_forge(forge_id: str):
51    """Heats on one forge (forge_id index), highest risk first."""
52    return (
53        Heats.where(Heats.forge_id == forge_id)
54        .select(Heats.heat_id, Heats.alloy, Heats.status, Heats.peak_c, Heats.band, Heats.risk, Heats.tag, Heats.lane)
55        .order_by(Heats.risk, asc=False)
56    )
57
58
59@pxt.query
60def by_band_status(band: str, status: str):
61    """Heats in a heat band and status (band + status indexes)."""
62    return (
63        Heats.where((Heats.band == band) & (Heats.status == status))
64        .select(Heats.heat_id, Heats.forge_id, Heats.alloy, Heats.peak_c, Heats.risk, Heats.tag, Heats.lane)
65        .order_by(Heats.peak_c, asc=False)
66    )
67
68
69@pxt.query
70def by_alloy(alloy: str):
71    """All heats for an alloy (alloy index)."""
72    return (
73        Heats.where(Heats.alloy == alloy)
74        .select(Heats.heat_id, Heats.forge_id, Heats.status, Heats.band, Heats.risk, Heats.tag)
75        .order_by(Heats.forge_id)
76    )
77
78
79@pxt.query
80def hot_lane():
81    """Shop board: heats currently in the hot priority lane."""
82    return (
83        Heats.where(Heats.lane == 'hot')
84        .select(Heats.heat_id, Heats.forge_id, Heats.alloy, Heats.status, Heats.risk, Heats.tag)
85        .order_by(Heats.risk, asc=False)
86    )
87
88
89# ---- routes ----
90forge_api = FastAPIRouter(name='forge_api')
91
92forge_api.add_insert_route(
93    Heats,
94    path='/heats',
95    inputs=[Heats.forge_id, Heats.alloy, Heats.status, Heats.peak_c, Heats.hold_min, Heats.quench, Heats.operator, Heats.note],
96    outputs=[Heats.heat_id, Heats.band, Heats.risk, Heats.tag, Heats.lane],
97)
98forge_api.add_update_route(
99    Heats,
100    path='/heats/status',
101    inputs=[Heats.status, Heats.note],
102    outputs=[Heats.heat_id, Heats.status, Heats.lane, Heats.tag],
103)
104forge_api.add_compute_route(
105    Heats,
106    path='/score',
107    inputs=[Heats.peak_c, Heats.hold_min, Heats.quench, Heats.alloy, Heats.status],
108    outputs=[Heats.band, Heats.risk, Heats.tag, Heats.lane],
109)
110forge_api.add_delete_route(Heats, path='/heats/delete')
111forge_api.add_query_route(path='/heats/by-forge', query=by_forge, method='get')
112forge_api.add_query_route(path='/heats/by-band', query=by_band_status, method='get')
113forge_api.add_query_route(path='/heats/by-alloy', query=by_alloy, method='get')
114forge_api.add_query_route(path='/heats/hot-lane', query=hot_lane, method='get')