151 lines
4.9 KiB
Python
151 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from glob import glob
|
|
from os import path
|
|
import itertools
|
|
from copy import deepcopy
|
|
from typing import Iterable, List, Union, Tuple
|
|
from tatsu import parse as tatsu_parse
|
|
from tatsu.ast import AST
|
|
from tatsu.exceptions import SemanticError
|
|
from .models import Location, Choice, Option, Dish
|
|
from utils import first
|
|
|
|
|
|
# TODO Use proper way to get resources, see https://stackoverflow.com/a/10935674
|
|
with open(path.join(path.dirname(__file__), "hlds.tatsu")) as fh:
|
|
GRAMMAR = fh.read()
|
|
|
|
|
|
def filter_instance(cls, iterable):
|
|
return [item for item in iterable if isinstance(item, cls)]
|
|
|
|
|
|
class ChoiceReference:
|
|
def __init__(self, identifier, price):
|
|
self.identifier = identifier
|
|
self.price = price
|
|
|
|
|
|
# pylint: disable=no-self-use
|
|
class HldsSemanticActions:
|
|
def location(self, ast) -> Location:
|
|
choices = {
|
|
choice.id: choice for choice in filter_instance(Choice, ast["items_"])
|
|
}
|
|
dishes: Iterable[Dish] = filter_instance(Dish, ast["items_"])
|
|
for dish in dishes:
|
|
for i, choice in enumerate(dish.choices):
|
|
if not isinstance(choice[1], Choice):
|
|
choiceId, choiceRef = choice
|
|
assert isinstance(choiceRef, ChoiceReference)
|
|
# We must replace the ChoiceReference with the Choice it refers to. A deep copy
|
|
# allows us to modify the individual Options of the Choice.
|
|
choiceObject = deepcopy(choices[choiceRef.identifier])
|
|
|
|
for option in choiceObject.options:
|
|
option.price += choiceRef.price
|
|
|
|
dish.choices[i] = (choiceId, choiceObject)
|
|
|
|
# Move the base price to the first single_choice if the dish doesn't have a fixed price
|
|
first_single_choice = first(
|
|
c[1] for c in dish.choices if c[0] == "single_choice"
|
|
)
|
|
price_range = dish.price_range()
|
|
if dish.price and price_range[0] != price_range[1] and first_single_choice:
|
|
for option in first_single_choice.options:
|
|
option.price += dish.price
|
|
dish.price = 0
|
|
|
|
attributes = {att["key"]: att["value"] for att in ast["attributes"]}
|
|
|
|
return Location(
|
|
ast["id"],
|
|
name=ast["name"],
|
|
dishes=dishes,
|
|
osm=attributes.get("osm"),
|
|
address=attributes.get("address"),
|
|
telephone=attributes.get("phone"),
|
|
website=attributes.get("website"),
|
|
)
|
|
|
|
def dish_block(self, ast) -> Dish:
|
|
return Dish(
|
|
ast["id"],
|
|
name=ast["name"],
|
|
description=ast["description"],
|
|
price=ast["price"] or 0,
|
|
tags=ast["tags"] or [],
|
|
choices=ast["choices"],
|
|
)
|
|
|
|
def choice_block(self, ast) -> Choice:
|
|
if ast["price"] or ast["tags"]:
|
|
raise SemanticError(
|
|
"Choice block definitions cannot have price or tags, put them on each of its options instead"
|
|
)
|
|
|
|
return Choice(
|
|
ast["id"],
|
|
name=ast["name"],
|
|
description=ast["description"],
|
|
options=ast["entries"],
|
|
)
|
|
|
|
def indent_choice_block(self, ast) -> Tuple[str, Union[Choice, AST]]:
|
|
if ast["kind"] == "declaration":
|
|
return (ast["type"], self.choice_block(ast))
|
|
else:
|
|
if ast["type"] == "single_choice" and ast["price"]:
|
|
raise SemanticError(
|
|
"Single_choice choices can't have a price, because it would always be triggered"
|
|
)
|
|
return (ast["type"], ChoiceReference(ast["id"], ast["price"] or 0))
|
|
|
|
def indent_choice_entry(self, ast) -> Option:
|
|
return Option(
|
|
ast["id"],
|
|
name=ast["name"],
|
|
description=ast["description"],
|
|
price=ast["price"] or 0,
|
|
tags=ast["tags"] or [],
|
|
)
|
|
|
|
noindent_choice_entry = indent_choice_entry
|
|
|
|
def price(self, ast) -> int:
|
|
return 100 * int(ast["value_unit"]) + (
|
|
0
|
|
if not ast["value_cents"]
|
|
else 10 * int(ast["value_cents"])
|
|
if len(ast["value_cents"]) == 1
|
|
else int(ast["value_cents"])
|
|
)
|
|
|
|
def _default(self, ast):
|
|
return ast
|
|
|
|
|
|
SEMANTICS = HldsSemanticActions()
|
|
|
|
|
|
def parse(menu: str) -> List[Location]:
|
|
parsed = tatsu_parse(GRAMMAR, menu, semantics=SEMANTICS)
|
|
return parsed
|
|
|
|
|
|
def parse_file(filename: str) -> List[Location]:
|
|
with open(filename, "r") as file_handle:
|
|
return parse(file_handle.read())
|
|
|
|
|
|
def parse_files(files: Iterable[str]) -> List[Location]:
|
|
menus = map(parse_file, files)
|
|
return list(itertools.chain.from_iterable(menus))
|
|
|
|
|
|
def parse_all_directory(directory: str) -> List[Location]:
|
|
# TODO Use proper way to get resources, see https://stackoverflow.com/a/10935674
|
|
files = glob(path.join(directory, "**.hlds"), recursive=True)
|
|
return parse_files(files)
|