fix: use openfoodfacts-python package for taxonomy processing

pyproject.toml

@@ -78,6 +78,7 @@ lark = "~1.1.4"
 h5py = "~3.8.0"
 opencv-contrib-python = "~4.7.0.72"
 toml = "~0.10.2"
+openfoodfacts = "~0.1.5"
 
 [tool.poetry.dependencies.sentry-sdk]
 version = "~1.14.0"

robotoff/insights/importer.py

@@ -863,7 +863,7 @@ def add_optional_fields(cls, insight: ProductInsight, product: Optional[Product]
         campaigns = []
         if (
             insight.value_tag in taxonomy
-            and "agribalyse_food_code" in taxonomy[insight.value_tag].additional_data
+            and "agribalyse_food_code" in taxonomy[insight.value_tag].properties
         ):
             # This category is linked to an agribalyse category, add it as a
             # campaign tag

robotoff/prediction/category/neural/keras_category_classifier_3_0/preprocessing.py

@@ -8,7 +8,7 @@
 import numpy as np
 
 from robotoff import settings
-from robotoff.taxonomy import Taxonomy, fetch_taxonomy
+from robotoff.taxonomy import Taxonomy
 from robotoff.types import JSONType
 from robotoff.utils.text import KeywordProcessor
 
@@ -35,11 +35,7 @@
 
 @functools.cache
 def get_ingredient_taxonomy():
-    return fetch_taxonomy(
-        "",
-        V3_MODEL_DATA_DIR / "ingredients.full.json.gz",
-        offline=True,
-    )
+    return Taxonomy.from_path(V3_MODEL_DATA_DIR / "ingredients.full.json.gz")
 
 
 @functools.cache

robotoff/taxonomy.py

@@ -1,255 +1,18 @@
 import collections
-import pathlib
-from enum import Enum
-from typing import Any, Iterable, Optional, Union
+from typing import Optional
 
 import cachetools
-import requests
+from openfoodfacts.taxonomy import Taxonomy
+from openfoodfacts.taxonomy import get_taxonomy as _get_taxonomy
+from openfoodfacts.types import TaxonomyType
 
 from robotoff import settings
-from robotoff.types import JSONType
-from robotoff.utils import get_logger, http_session, load_json
+from robotoff.utils import get_logger
 from robotoff.utils.text import get_tag
 
-try:
-    import networkx
-except ImportError:
-    networkx = None
-
 logger = get_logger(__name__)
 
 
-class TaxonomyType(Enum):
-    category = 1
-    ingredient = 2
-    label = 3
-    brand = 4
-    packaging_shape = 5
-    packaging_material = 6
-    packaging_recycling = 7
-
-
-class TaxonomyNode:
-    __slots__ = ("id", "names", "parents", "children", "synonyms", "additional_data")
-
-    def __init__(
-        self,
-        identifier: str,
-        names: dict[str, str],
-        synonyms: Optional[dict[str, list[str]]],
-        additional_data: Optional[dict[str, Any]] = None,
-    ):
-        self.id: str = identifier
-        self.names: dict[str, str] = names
-        self.parents: list["TaxonomyNode"] = []
-        self.children: list["TaxonomyNode"] = []
-        self.additional_data = additional_data or {}
-
-        if synonyms:
-            self.synonyms = synonyms
-        else:
-            self.synonyms = {}
-
-    def is_child_of(self, item: "TaxonomyNode") -> bool:
-        """Return True if `item` is a child of `self` in the taxonomy."""
-        if not self.parents:
-            return False
-
-        if item in self.parents:
-            return True
-
-        for parent in self.parents:
-            is_parent = parent.is_child_of(item)
-
-            if is_parent:
-                return True
-
-        return False
-
-    def is_parent_of(self, item: "TaxonomyNode") -> bool:
-        return item.is_child_of(self)
-
-    def is_parent_of_any(self, candidates: Iterable["TaxonomyNode"]) -> bool:
-        for candidate in candidates:
-            if candidate.is_child_of(self):
-                return True
-
-        return False
-
-    def get_parents_hierarchy(self) -> list["TaxonomyNode"]:
-        """Return the list of all parent nodes (direct and indirect)."""
-        all_parents = []
-        seen: set[str] = set()
-
-        if not self.parents:
-            return []
-
-        for self_parent in self.parents:
-            if self_parent.id not in seen:
-                all_parents.append(self_parent)
-                seen.add(self_parent.id)
-
-            for parent_parent in self_parent.get_parents_hierarchy():
-                if parent_parent.id not in seen:
-                    all_parents.append(parent_parent)
-                    seen.add(parent_parent.id)
-
-        return all_parents
-
-    def get_localized_name(self, lang: str) -> str:
-        if lang in self.names:
-            return self.names[lang]
-
-        if "xx" in self.names:
-            # Return international name if it exists
-            return self.names["xx"]
-
-        return self.id
-
-    def get_synonyms(self, lang: str) -> list[str]:
-        return self.synonyms.get(lang, [])
-
-    def add_parents(self, parents: Iterable["TaxonomyNode"]):
-        for parent in parents:
-            if parent not in self.parents:
-                self.parents.append(parent)
-                parent.children.append(self)
-
-    def to_dict(self) -> JSONType:
-        return {"name": self.names, "parents": [p.id for p in self.parents]}
-
-    def __repr__(self):
-        return "<TaxonomyNode %s>" % self.id
-
-
-class Taxonomy:
-    def __init__(self) -> None:
-        self.nodes: dict[str, TaxonomyNode] = {}
-
-    def add(self, key: str, node: TaxonomyNode) -> None:
-        self.nodes[key] = node
-
-    def __contains__(self, item: str):
-        return item in self.nodes
-
-    def __getitem__(self, item: str):
-        return self.nodes.get(item)
-
-    def __len__(self):
-        return len(self.nodes)
-
-    def iter_nodes(self) -> Iterable[TaxonomyNode]:
-        """Iterate over the nodes of the taxonomy."""
-        return iter(self.nodes.values())
-
-    def keys(self):
-        return self.nodes.keys()
-
-    def find_deepest_nodes(self, nodes: list[TaxonomyNode]) -> list[TaxonomyNode]:
-        """Given a list of nodes, returns the list of nodes where all the
-        parents within the list have been removed.
-
-        For example, for a taxonomy, 'fish' -> 'salmon' -> 'smoked-salmon':
-
-        ['fish', 'salmon'] -> ['salmon'] ['fish', 'smoked-salmon'] ->
-        [smoked-salmon']
-        """
-        excluded: set[str] = set()
-
-        for node in nodes:
-            for second_node in (
-                n for n in nodes if n.id not in excluded and n.id != node.id
-            ):
-                if node.is_child_of(second_node):
-                    excluded.add(second_node.id)
-
-        return [node for node in nodes if node.id not in excluded]
-
-    def is_parent_of_any(
-        self, item: str, candidates: Iterable[str], raises: bool = True
-    ) -> bool:
-        """Return True if `item` is parent of any candidate, False otherwise.
-
-        If the item is not in the taxonomy and raises is False, return False.
-
-        :param item: The item to compare
-        :param candidates: A list of candidates
-        :param raises: if True, raises a ValueError if item is not in the
-        taxonomy, defaults to True.
-        """
-        node: TaxonomyNode = self[item]
-
-        if node is None:
-            if raises:
-                raise ValueError(f"unknown id in taxonomy: {node}")
-            else:
-                return False
-
-        to_check_nodes: set[TaxonomyNode] = set()
-
-        for candidate in candidates:
-            candidate_node = self[candidate]
-
-            if candidate_node is not None:
-                to_check_nodes.add(candidate_node)
-
-        return node.is_parent_of_any(to_check_nodes)
-
-    def get_localized_name(self, key: str, lang: str) -> str:
-        if key not in self.nodes:
-            return key
-
-        return self.nodes[key].get_localized_name(lang)
-
-    def to_dict(self) -> JSONType:
-        export = {}
-
-        for key, node in self.nodes.items():
-            export[key] = node.to_dict()
-
-        return export
-
-    @classmethod
-    def from_dict(cls, data: JSONType) -> "Taxonomy":
-        taxonomy = Taxonomy()
-
-        for key, key_data in data.items():
-            if key not in taxonomy:
-                node = TaxonomyNode(
-                    identifier=key,
-                    names=key_data.get("name", {}),
-                    synonyms=key_data.get("synonyms", None),
-                    additional_data={
-                        k: v
-                        for k, v in key_data.items()
-                        if k not in {"parents", "name", "synonyms", "children"}
-                    },
-                )
-                taxonomy.add(key, node)
-
-        for key, key_data in data.items():
-            node = taxonomy[key]
-            parents = [taxonomy[ref] for ref in key_data.get("parents", [])]
-            node.add_parents(parents)
-
-        return taxonomy
-
-    @classmethod
-    def from_json(cls, file_path: Union[str, pathlib.Path]):
-        return cls.from_dict(load_json(file_path, compressed=True))  # type: ignore
-
-    def to_graph(self):
-        """Generate a networkx.DiGraph from the taxonomy."""
-        graph = networkx.DiGraph()
-        graph.add_nodes_from((x.id for x in self.iter_nodes()))
-
-        for node in self.iter_nodes():
-            for child in node.children:
-                graph.add_edge(node.id, child.id)
-
-        return graph
-
-
 def generate_category_hierarchy(
     taxonomy: Taxonomy, category_to_index: dict[str, int], root: int
 ):
@@ -280,42 +43,28 @@ def generate_category_hierarchy(
     return categories_hierarchy_list
 
 
-def fetch_taxonomy(
-    url: str, fallback_path: Optional[str] = None, offline: bool = False
-) -> Taxonomy:
-    if offline and fallback_path:
-        return Taxonomy.from_json(fallback_path)
-
-    try:
-        r = http_session.get(url, timeout=120)  # might take some time
-        if r.status_code >= 300:
-            raise requests.HTTPError(
-                "Taxonomy download at %s returned status code %s", url, r.status_code
-            )
-        data = r.json()
-    except Exception as e:
-        logger.exception(f"{type(e)} exception while fetching taxonomy at %s", url)
-        if fallback_path:
-            return Taxonomy.from_json(fallback_path)
-        else:
-            raise e
-
-    return Taxonomy.from_dict(data)
-
-
 @cachetools.cached(cache=cachetools.TTLCache(maxsize=100, ttl=12 * 60 * 60))  # 12h
 def get_taxonomy(taxonomy_type: str, offline: bool = False) -> Taxonomy:
-    """Returned the requested Taxonomy."""
-    logger.info("Loading taxonomy %s...", taxonomy_type)
+    """Return the taxonomy of type `taxonomy_type`.
+
+    The taxonomy is cached in memory and locally on disk. Every 12h, we check
+    if a new version is available and download if True.
 
-    if taxonomy_type not in settings.TAXONOMY_URLS:
-        raise ValueError(f"unknown taxonomy type: {taxonomy_type}")
+    A local static version can also be fetched (for unit tests for example)
+    with `offline=True`.
+
+    :param taxonomy_type: the taxonomy type
+    :param offline: if True, return a local static version of the taxonomy,
+      defaults to False. It's not available for all taxonomy types.
+    :return: the Taxonomy
+    """
+    if offline:
+        return Taxonomy.from_path(str(settings.TAXONOMY_PATHS[taxonomy_type]))
 
-    url = settings.TAXONOMY_URLS[taxonomy_type]
-    return fetch_taxonomy(
-        url,
-        fallback_path=str(settings.TAXONOMY_PATHS.get(taxonomy_type, "")) or None,
-        offline=offline,
+    return _get_taxonomy(
+        TaxonomyType[taxonomy_type],
+        force_download=False,
+        cache_dir=settings.DATA_DIR / "taxonomies",
     )
 
 

scripts/category_dataset.py

@@ -17,11 +17,12 @@
 import os
 from typing import Iterator, Optional
 
+from openfoodfacts.taxonomy import Taxonomy, TaxonomyNode
 from sklearn.model_selection import train_test_split
 
 from robotoff import settings
 from robotoff.products import ProductDataset, ProductStream
-from robotoff.taxonomy import Taxonomy, TaxonomyNode, get_taxonomy
+from robotoff.taxonomy import get_taxonomy
 from robotoff.types import JSONType
 from robotoff.utils import dump_jsonl, get_logger
 

tests/conftest.py

@@ -46,4 +46,4 @@ def peewee_db(peewee_db_create):
 
 @pytest.fixture(scope="session")
 def category_taxonomy():
-    return Taxonomy.from_json(settings.TAXONOMY_PATHS["category"])
+    return Taxonomy.from_path(settings.TAXONOMY_PATHS["category"])