comfy-recipes/recipes.py

383 lines
No EOL
12 KiB
Python

import sys
from typing import Dict, List, Any, Optional, Set
import os
import json
import yaml
import jinja2
import jsonschema
import jsonschema.exceptions
class Context:
def __init__(self) -> None:
self.units = Units()
self.ingredients = Ingredients(self)
class Conversion:
def load(self, units: List["Unit"], dct: Dict[str, Any]) -> None:
def find_unit(name: str) -> Optional[Unit]:
for unit in units:
if unit.name == name:
return unit
return None
fromunit = find_unit(dct["from"])
if fromunit is None:
raise RuntimeError(f"unit {dct['from']} doesn't exist")
self.fromunit = fromunit
tounit = find_unit(dct["to"])
if tounit is None:
raise RuntimeError(f"unit {tounit} doesn't exist")
self.tounit = tounit
self.ratio = dct["ratio"]
class Unit:
def load(self, units: List["Unit"], dct: Dict[str, Any]) -> None:
self.name = dct["name"]
unitsx = units[:]
unitsx.append(self)
self.conversions: List[Conversion] = []
if "conversions" in dct:
for convdct in dct["conversions"]:
if "from" in dct["conversions"]:
raise RuntimeError(
f"conversions in units.yaml cannot have a from field, it is automatically assigned from the unit name"
)
convdct["from"] = self.name
conversion = Conversion()
conversion.load(unitsx, convdct)
self.conversions.append(conversion)
self.aliases: List[str] = []
if "aliases" in dct:
for alias in dct["aliases"]:
self.aliases.append(alias)
class Units:
def __init__(self) -> None:
self.units: List[Unit] = []
def load(self, lst: List[Any]) -> List[str]:
for unitdct in lst:
unit = Unit()
unit.load(self.units, unitdct)
self.units.append(unit)
return []
def get(self, name: str) -> Optional[Unit]:
for unit in self.units:
if unit.name == name:
return unit
raise RuntimeError(f"unit {name} not found")
class Ingredient:
def __init__(self, ctx: Context) -> None:
self.ctx = ctx
def load(self, dct: Dict[str, Any]) -> List[str]:
issues = []
self.name = dct["name"]
self.wdid = -1
if "wdid" in dct:
self.wdid = dct["wdid"]
self.pricedb = None
if "pricedb" in dct:
self.pricedb = PriceDBs(self.ctx)
issues += self.pricedb.load(dct["pricedb"])
self.aliases = []
if "aliases" in dct:
for elem in dct["aliases"]:
self.aliases.append(elem)
self.conversions = []
if "conversions" in dct:
for dct in dct["conversions"]:
conversion = Conversion()
conversion.load(self.ctx.units.units, dct)
self.conversions.append(conversion)
return issues
class Ingredients:
def __init__(self, ctx: Context) -> None:
self.ctx = ctx
self.ingredients: List[Ingredient] = []
def load(self, lst: List[Any]) -> List[str]:
issues = []
for ingdct in lst:
ing = Ingredient(self.ctx)
issues += ing.load(ingdct)
self.ingredients.append(ing)
return issues
def get(self, name: str) -> Optional[Ingredient]:
for ing in self.ingredients:
if ing.name == name or name in ing.aliases:
return ing
raise RuntimeError(f"ingredient {name} not found")
class PriceDBs:
def __init__(self, ctx: Context) -> None:
self.ctx = ctx
self.pricedbs: List[PriceDB] = []
def load(self, lst: List[Any]) -> List[str]:
issues = []
for elem in lst:
pricedb = PriceDB(self.ctx)
issues += pricedb.load(elem)
self.pricedbs.append(pricedb)
return issues
class PriceDB:
def __init__(self, ctx: Context) -> None:
self.ctx = ctx
def load(self, dct: Dict[str, Any]) -> List[str]:
issues = []
self.price = dct["price"]
self.amount = 1.0
if "amount" in dct:
self.amount = dct["amount"]
self.unit = None
if "unit" in dct:
try:
self.unit = self.ctx.units.get(dct["unit"])
except RuntimeError as e:
issues.append(str(e))
return issues
class IngredientInstance:
def __init__(
self, ctx: Context, defaultamount: float = 1, defaultunit: Optional[Unit] = None
) -> None:
self.ctx = ctx
self.defaultamount = float(defaultamount)
self.defaultunit: Optional[Unit] = defaultunit
def load(self, dct: Dict[str, Any]) -> List[str]:
issues = []
self.name = dct["name"]
try:
self.ingredient = self.ctx.ingredients.get(self.name)
except RuntimeError as e:
issues.append(str(e))
self.amount = self.defaultamount
if "amount" in dct:
self.amount = dct["amount"]
self.unit = self.defaultunit
if "unit" in dct:
try:
self.unit = self.ctx.units.get(dct["unit"])
except RuntimeError as e:
issues.append(str(e))
self.note = ""
if "note" in dct:
self.note = dct["note"]
self.alternatives = []
if "or" in dct:
for ingdct in dct["or"]:
ingredient = IngredientInstance(
self.ctx, defaultamount=self.amount, defaultunit=self.unit
)
ingredient.load(ingdct)
self.alternatives.append(ingredient)
return issues
class RecipePart:
def __init__(self, ctx: Context) -> None:
self.ctx = ctx
def load(self, dct: Dict[str, Any]) -> List[str]:
issues = []
self.title = dct["title"]
self.ingredients: List[IngredientInstance] = []
for ing in dct["ingredients"]:
ingredient = IngredientInstance(self.ctx)
issues += ingredient.load(ing)
self.ingredients.append(ingredient)
self.steps: List[str] = dct["steps"]
return issues
class Recipe:
def __init__(self, ctx: Context) -> None:
self.ctx = ctx
self.parts: List[RecipePart] = []
self.srcpath = ""
self.outpath = ""
self.title = ""
def load(self, dct: Dict[str, Any]) -> List[str]:
issues: List[str] = []
if "parts" in dct:
self.title = dct["title"]
for partdct in dct["parts"]:
rp = RecipePart(self.ctx)
issues += rp.load(partdct)
self.parts.append(rp)
else:
rp = RecipePart(self.ctx)
issues = rp.load(dct)
self.parts = [rp]
self.title = rp.title
return issues
class Builder:
def __init__(self) -> None:
self.jinjaenv = jinja2.Environment(
loader=jinja2.FileSystemLoader("templates"),
autoescape=jinja2.select_autoescape(),
)
def numprint(input: int) -> str:
out = str(input)
if out.endswith(".0"):
out = out.split(".")[0]
if out == "0.5":
return "1/2"
elif out == "0.25":
return "1/4"
elif out == "0.75":
return "3/4"
return out
self.jinjaenv.filters["numprint"] = numprint
self.ctx = Context()
# list of output files that will be built
self.outfiles: Set[str] = set()
def load_file(self, file: str) -> Any:
print(f"loading {file}")
with open(file, encoding="utf-8") as f:
txt = f.read()
if file.endswith(".json"):
return json.loads(txt)
return yaml.safe_load(txt)
def rendertemplate(
self, templatepath: str, format: str, file: str, dir: str, args: Any
) -> None:
template = self.jinjaenv.get_template(templatepath)
print(f"rendering {file}")
outstr = template.render(args)
os.makedirs(f"{dir}/out/{format}", exist_ok=True)
with open(f"{dir}/out/{format}/{file}", "w", encoding="utf-8") as f:
f.write(outstr)
self.outfiles.add(file)
def load(self, dir: str) -> int:
issues: List[str] = []
unitsdct = self.load_file(dir+"/units.yaml")
unitsschema = self.load_file("schemas/units.json")
jsonschema.validate(instance=unitsdct, schema=unitsschema)
issues += self.ctx.units.load(unitsdct)
if len(issues) != 0:
for issue in issues:
print("ERROR in units.yaml:", issue)
return 1
ingredientsdct = self.load_file(dir+"/ingredients.yaml")
ingredientsschema = self.load_file("schemas/ingredients.json")
jsonschema.validate(instance=ingredientsdct, schema=ingredientsschema)
issues += self.ctx.ingredients.load(ingredientsdct)
if len(issues) != 0:
for issue in issues:
print("ERROR in ingredients.yaml:", issue)
return 1
return 0
def run(self, dir: str) -> int:
issues = []
files = []
for _, _, filesx in os.walk(dir+"/recipes"):
files = filesx
files.sort()
recipes: List[Recipe] = []
recipeschema = self.load_file("schemas/recipe.json")
for file in files:
if not file.endswith(".yaml"):
print(f"unknown extension of {file}")
continue
recipe = Recipe(self.ctx)
recipedct = self.load_file(dir+"/recipes/" + file)
jsonschema.validate(instance=recipedct, schema=recipeschema)
issues += recipe.load(recipedct)
recipe.srcpath = file
recipe.outpath = file[:-5] + ".html"
recipes.append(recipe)
if len(issues) != 0:
for issue in issues:
print("ERROR", issue)
return 1
self.rendertemplate("index.html", "html", "index.html", dir, {"recipes": recipes})
for recipe in recipes:
self.rendertemplate(
"recipe.html", "html", recipe.outpath, dir, {"recipe": recipe}
)
return 0
def finish(self, dir: str) -> int:
files = set()
for _, _, filesx in os.walk(f"{dir}/out/html"):
files = set(filesx)
# files we did not generate, probably left by a previous run, but not valid anymore
extra_files = files - self.outfiles
for file in extra_files:
print(f"removing obsolete {file}")
os.remove(f"{dir}/out/html/{file}")
return 0
def build(self, path: str) -> int:
fcs = [self.load, self.run, self.finish]
for func in fcs:
try:
ret = func(path)
if ret != 0:
return ret
except jsonschema.exceptions.ValidationError as e:
print("ERROR:", e)
return 1
return 0
if __name__ == "__main__":
builder = Builder()
sys.exit(builder.build("recipes"))