|
1 | 1 | import enum |
2 | | -from typing import Any, Dict, Optional, Union |
| 2 | +from typing import Any, Dict, Literal, Optional, Union |
3 | 3 |
|
4 | | -from pydantic import BaseModel, model_validator |
| 4 | +from pydantic import BaseModel, Field, model_validator |
5 | 5 |
|
6 | 6 | #: A precise expectation of what mappings looks like in json. |
7 | 7 | #: (dict where keys are always of type `str`). |
@@ -913,3 +913,97 @@ class TaxonomyType(str, enum.Enum): |
913 | 913 | origin = "origin" |
914 | 914 | language = "language" |
915 | 915 | other_nutritional_substance = "other_nutritional_substance" |
| 916 | + |
| 917 | + |
| 918 | +class NutritionV3NutrientAggregated(BaseModel): |
| 919 | + value: float |
| 920 | + unit: str = Field(description="Normalized unit", examples=["g", "kJ", "kcal", "%"]) |
| 921 | + modifier: str | None = None |
| 922 | + source: str = Field(examples=["packaging", "manufacturer", "estimate", "usda"]) |
| 923 | + source_per: Literal["serving", "100g", "100ml"] |
| 924 | + source_index: int |
| 925 | + |
| 926 | + |
| 927 | +class NutritionV3NutrientInput(BaseModel): |
| 928 | + unit: str |
| 929 | + value: float | None = None |
| 930 | + value_computed: float | None = None |
| 931 | + value_string: str | None = None |
| 932 | + modifier: str | None = None |
| 933 | + |
| 934 | + |
| 935 | +class NutritionV3AggregatedSet(BaseModel): |
| 936 | + preparation: Literal["as_sold", "prepared"] |
| 937 | + per: Literal["100g", "100ml"] |
| 938 | + nutrients: dict[str, NutritionV3NutrientAggregated] = Field(default_factory=dict) |
| 939 | + |
| 940 | + |
| 941 | +class NutritionV3InputSet(BaseModel): |
| 942 | + preparation: Literal["as_sold", "prepared"] |
| 943 | + per: Literal["serving", "100g", "100ml"] |
| 944 | + per_quantity: int |
| 945 | + per_unit: Literal["g", "ml"] |
| 946 | + source: str = Field( |
| 947 | + examples=["packaging", "manufacturer", "database-usda", "estimate"] |
| 948 | + ) |
| 949 | + source_description: str | None = None |
| 950 | + last_updated_t: int | None = Field( |
| 951 | + None, description="timestamp of the last update of the input set" |
| 952 | + ) |
| 953 | + nutrients: dict[str, NutritionV3NutrientInput] = Field( |
| 954 | + default_factory=dict, description="Mapping from nutrient name to its value" |
| 955 | + ) |
| 956 | + unspecified_nutrients: list[str] = Field( |
| 957 | + default_factory=list, |
| 958 | + description="List of unspecified nutrients (ex: not provided on the packaging)", |
| 959 | + ) |
| 960 | + |
| 961 | + |
| 962 | +class NutritionV3(BaseModel): |
| 963 | + """New `nutrition` field of products, which come with schema_version 1003.""" |
| 964 | + |
| 965 | + aggregated_set: NutritionV3AggregatedSet | None = None |
| 966 | + input_sets: list[NutritionV3InputSet] = Field(default_factory=list) |
| 967 | + |
| 968 | + def filter_input_sets( |
| 969 | + self, |
| 970 | + per: str | None = None, |
| 971 | + preparation: str | None = None, |
| 972 | + per_quantity: int | None = None, |
| 973 | + per_unit: str | None = None, |
| 974 | + source: str | None = None, |
| 975 | + ) -> list["NutritionV3InputSet"]: |
| 976 | + results = self.input_sets |
| 977 | + if per is not None: |
| 978 | + results = [s for s in results if s.per == per] |
| 979 | + if preparation is not None: |
| 980 | + results = [s for s in results if s.preparation == preparation] |
| 981 | + if per_quantity is not None: |
| 982 | + results = [s for s in results if s.per_quantity == per_quantity] |
| 983 | + if per_unit is not None: |
| 984 | + results = [s for s in results if s.per_unit == per_unit] |
| 985 | + if source is not None: |
| 986 | + results = [s for s in results if s.source == source] |
| 987 | + return results |
| 988 | + |
| 989 | + def get_input_nutrient( |
| 990 | + self, |
| 991 | + nutrient: str, |
| 992 | + per: str | None = None, |
| 993 | + preparation: str | None = None, |
| 994 | + per_quantity: int | None = None, |
| 995 | + per_unit: str | None = None, |
| 996 | + source: str | None = None, |
| 997 | + ) -> NutritionV3NutrientInput | None: |
| 998 | + filtered_sets = self.filter_input_sets( |
| 999 | + per=per, |
| 1000 | + preparation=preparation, |
| 1001 | + per_quantity=per_quantity, |
| 1002 | + per_unit=per_unit, |
| 1003 | + source=source, |
| 1004 | + ) |
| 1005 | + if not filtered_sets: |
| 1006 | + return None |
| 1007 | + |
| 1008 | + input_set = filtered_sets[0] |
| 1009 | + return input_set.nutrients.get(nutrient) |
0 commit comments