Vineyard Estate

1"""Vineyard Estate API built with Pixeltable.
2
3    pxt schema update app.py estate
4    pxt service run app.py estate
5"""
6import pixeltable as pxt
7import pixeltable.functions as pxtf
8from pixeltable.serving import FastAPIRouter
9
10from udfs import ripeness, score_label, vine_age, yield_band, yield_per_acre
11
12# ---- tables ----
13TableModel = pxt.model_base()
14
15
16class Blocks(TableModel, name='blocks'):
17    block_id = pxt.Column(type=pxt.String, primary_key=True)
18    name: pxt.String
19    varietal: pxt.String
20    acres: pxt.Float
21    organic: pxt.Bool
22    region: pxt.String
23    irrigation: pxt.String | None        # added later; nullable, so the update is in place
24
25
26class Vines(TableModel, name='vines'):
27    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
28    block_id: pxt.String
29    clone: pxt.String
30    rootstock: pxt.String
31    planted_year: pxt.Int
32
33    age = vine_age(planted_year)
34
35
36class Harvests(TableModel, name='harvests'):
37    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
38    block_id: pxt.String
39    picked_on: pxt.String
40    tons: pxt.Float
41    acres: pxt.Float
42    brix: pxt.Float
43
44    tpa = yield_per_acre(tons, acres)
45    band = yield_band(tons, acres)
46    ripe = ripeness(brix)
47
48
49class Tastings(TableModel, name='tastings'):
50    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
51    block_id: pxt.String
52    vintage: pxt.Int
53    score: pxt.Int
54    notes: pxt.String | None
55
56    label = score_label(score)
57
58
59# ---- queries ----
60@pxt.query
61def open_blocks(min_acres: float):
62    """Organic blocks of at least `min_acres`, largest first."""
63    return Blocks.where((Blocks.acres >= min_acres) & (Blocks.organic == True)).select(  # noqa: E712
64        Blocks.block_id, Blocks.name, Blocks.varietal, Blocks.acres, Blocks.irrigation
65    ).order_by(Blocks.acres, asc=False)
66
67
68@pxt.query
69def block_harvests(block_id: str):
70    """Harvest history for one block, oldest first."""
71    return Harvests.where(Harvests.block_id == block_id).select(
72        Harvests.picked_on, Harvests.tons, Harvests.tpa, Harvests.band, Harvests.brix, Harvests.ripe
73    ).order_by(Harvests.picked_on)
74
75
76# ---- routes ----
77estate_api = FastAPIRouter(name='estate_api')
78estate_api.add_insert_route(Blocks, path='/blocks',
79                            inputs=[Blocks.block_id, Blocks.name, Blocks.varietal, Blocks.acres, Blocks.organic,
80                                    Blocks.region, Blocks.irrigation],
81                            outputs=[Blocks.block_id])
82estate_api.add_insert_route(Vines, path='/vines',
83                            inputs=[Vines.block_id, Vines.clone, Vines.rootstock, Vines.planted_year],
84                            outputs=[Vines.id, Vines.age])
85estate_api.add_insert_route(Harvests, path='/harvests',
86                            inputs=[Harvests.block_id, Harvests.picked_on, Harvests.tons, Harvests.acres, Harvests.brix],
87                            outputs=[Harvests.id, Harvests.tpa, Harvests.band, Harvests.ripe])
88estate_api.add_insert_route(Tastings, path='/tastings',
89                            inputs=[Tastings.block_id, Tastings.vintage, Tastings.score, Tastings.notes],
90                            outputs=[Tastings.id, Tastings.label])
91estate_api.add_update_route(Blocks, path='/blocks/irrigation', inputs=[Blocks.irrigation],
92                            outputs=[Blocks.block_id, Blocks.irrigation])
93estate_api.add_compute_route(Harvests, path='/yield-band', inputs=[Harvests.tons, Harvests.acres],
94                             outputs=[Harvests.tpa, Harvests.band])
95estate_api.add_query_route(path='/blocks/open', query=open_blocks, method='get')
96estate_api.add_query_route(path='/blocks/harvests', query=block_harvests, method='get')