cs-si__eodag-961

{
  "status": "ok",
  "stdout": "eodag/rest/stac.py:65:STAC_CATALOGS_PREFIX = \"catalogs\"\neodag/rest/stac.py:73:    :param stac_config: STAC configuration from stac.yml conf file\neodag/rest/stac.py:100:        \"\"\"Updates data using given input STAC dict data\neodag/rest/stac.py:102:        :param data: Catalog data (parsed STAC dict)\neodag/rest/stac.py:133:        \"\"\"Parse STAC extension from config and return as dict\neodag/rest/stac.py:137:        :param stac_config: STAC configuration from stac.yml conf file\neodag/rest/stac.py:143:        :returns: STAC extension as dictionnary\neodag/rest/stac.py:159:        :returns: STAC data dictionnary\neodag/rest/stac.py:172:    :param stac_config: STAC configuration from stac.yml conf file\neodag/rest/stac.py:201:        \"\"\"Build STAC items list from EODAG search results\neodag/rest/stac.py:205:        :param catalog: STAC catalog dict used for parsing item metadata\neodag/rest/stac.py:207:        :returns: STAC item dicts list\neodag/rest/stac.py:245:            product_dict[\"assets\"] = product.assets.as_dict()\neodag/rest/stac.py:256:            url_parts = urlparse(str(product_item[\"assets\"][\"downloadLink\"][\"href\"]))\neodag/rest/stac.py:268:            _dc_qs = product_item[\"assets\"][\"downloadLink\"].pop(\"_dc_qs\", None)\neodag/rest/stac.py:274:                product_item[\"assets\"][\"downloadLink\"][\neodag/rest/stac.py:278:            # add origin assets to product assets\neodag/rest/stac.py:279:            origin_assets = product_item[\"assets\"].pop(\"origin_assets\")\neodag/rest/stac.py:280:            if getattr(product, \"assets\", False):\neodag/rest/stac.py:282:                for asset_key, asset_value in origin_assets.items():\neodag/rest/stac.py:283:                    origin_assets[asset_key][\"href\"] = without_arg_url\neodag/rest/stac.py:285:                        origin_assets[asset_key][\neodag/rest/stac.py:289:                        origin_assets[asset_key][\"href\"] += f\"/{asset_key}\"\neodag/rest/stac.py:291:                product_item[\"assets\"] = dict(product_item[\"assets\"], **origin_assets)\neodag/rest/stac.py:340:        \"\"\"Build STAC items from EODAG search results\neodag/rest/stac.py:344:        :param catalog: STAC catalog dict used for parsing item metadata\neodag/rest/stac.py:470:        :param item: STAC item data\neodag/rest/stac.py:501:        \"\"\"Build STAC item from EODAG product\neodag/rest/stac.py:505:        :returns: STAC item\neodag/rest/stac.py:530:        product_dict[\"assets\"] = product.assets.as_dict()\neodag/rest/stac.py:561:    :param stac_config: STAC configuration from stac.yml conf file\neodag/rest/stac.py:616:        \"\"\"Build STAC collections list\neodag/rest/stac.py:620:        :returns: STAC collection dicts list\neodag/rest/stac.py:675:        \"\"\"Build STAC collections\neodag/rest/stac.py:708:        \"\"\"Build STAC collection by its id\neodag/rest/stac.py:733:    :param stac_config: STAC configuration from stac.yml conf file\neodag/rest/stac.py:1258:                            self.url, f\"{STAC_CATALOGS_PREFIX}/{product_type['ID']}\"\neodag/rest/stac.py:1382:        \"\"\"Get nested STAC catalog as data dict\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "eodag/rest/stac.py",
  "content": "    ) -> List[Dict[str, Any]]:\n        \"\"\"Build STAC items list from EODAG search results\n\n        :param search_results: EODAG search results\n        :type search_results: :class:`~eodag.api.search_result.SearchResult`\n        :param catalog: STAC catalog dict used for parsing item metadata\n        :type catalog: dict\n        :returns: STAC item dicts list\n        :rtype: list\n        \"\"\"\n        if len(search_results) <= 0:\n            return []\n\n        item_model = self.__filter_item_model_properties(\n            self.stac_config[\"item\"], str(search_results[0].product_type)\n        )\n        provider_model = deepcopy(self.stac_config[\"provider\"])\n\n        # check if some items need to be converted\n        need_conversion: Dict[str, Any] = {}\n        for k, v in item_model[\"properties\"].items():\n            if isinstance(v, str):\n                conversion, item_model[\"properties\"][k] = get_metadata_path(\n                    item_model[\"properties\"][k]\n                )\n                if conversion is not None:\n                    need_conversion[k] = conversion\n                    # convert str to jsonpath if needed\n                    item_model[\"properties\"][k] = string_to_jsonpath(\n                        k, item_model[\"properties\"][k]\n                    )\n\n        item_list: List[Dict[str, Any]] = []\n        for product in search_results:\n            # parse jsonpath\n            provider_dict = jsonpath_parse_dict_items(\n                provider_model,\n                {\n                    \"provider\": self.eodag_api.providers_config[\n                        product.provider\n                    ].__dict__\n                },\n            )\n\n            product_dict = deepcopy(product.__dict__)\n            product_dict[\"assets\"] = product.assets.as_dict()\n\n            product_item = jsonpath_parse_dict_items(\n                item_model,\n                {\n                    \"product\": product_dict,\n                    \"providers\": [provider_dict],\n                },\n            )\n\n            # parse download link\n            url_parts = urlparse(str(product_item[\"assets\"][\"downloadLink\"][\"href\"]))\n            query_dict = parse_qs(url_parts.query)\n            without_arg_url = (\n                f\"{url_parts.scheme}://{url_parts.netloc}{url_parts.path}\"\n                if url_parts.scheme\n                else f\"{url_parts.netloc}{url_parts.path}\"\n            )\n\n            # add provider to query-args\n            if self.provider:\n                query_dict.update(provider=[self.provider])\n            # add datacube query-string to query-args\n            _dc_qs = product_item[\"assets\"][\"downloadLink\"].pop(\"_dc_qs\", None)\n            if _dc_qs:\n                query_dict.update(_dc_qs=_dc_qs)\n\n            # update download link with up-to-date query-args\n            if query_dict:\n                product_item[\"assets\"][\"downloadLink\"][\n                    \"href\"\n                ] = f\"{without_arg_url}?{urlencode(query_dict, doseq=True)}\"\n\n            # add origin assets to product assets\n            origin_assets = product_item[\"assets\"].pop(\"origin_assets\")\n            if getattr(product, \"assets\", False):\n                # replace origin asset urls with eodag-server ones\n                for asset_key, asset_value in origin_assets.items():\n                    origin_assets[asset_key][\"href\"] = without_arg_url\n                    if query_dict:\n                        origin_assets[asset_key][\n                            \"href\"\n                        ] += f\"/{asset_key}?{urlencode(query_dict, doseq=True)}\"\n                    else:\n                        origin_assets[asset_key][\"href\"] += f\"/{asset_key}\"\n\n                product_item[\"assets\"] = dict(product_item[\"assets\"], **origin_assets)\n\n            # apply conversion if needed\n            for prop_key, prop_val in need_conversion.items():\n                conv_func, conv_args = prop_val\n                # colon `:` in key breaks format() method, hide it\n                formatable_prop_key = prop_key.replace(\":\", \"\")\n                if conv_args is not None:\n                    product_item[\"properties\"][prop_key] = format_metadata(\n                        \"{%s#%s(%s)}\" % (formatable_prop_key, conv_func, conv_args),",
  "start_line": 200,
  "end_line": 300,
  "total_lines": 1387,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "eodag/api/product/_product.py:189:    def as_dict(self) -> Dict[str, Any]:\neodag/api/product/_assets.py:48:    def as_dict(self) -> Dict[str, Any]:\neodag/api/product/_assets.py:79:    def as_dict(self) -> Dict[str, Any]:\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "eodag/api/product/_assets.py",
  "content": "# -*- coding: utf-8 -*-\n# Copyright 2023, CS GROUP - France, https://www.csgroup.eu/\n#\n# This file is part of EODAG project\n#     https://www.github.com/CS-SI/EODAG\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nfrom __future__ import annotations\n\nfrom collections import UserDict\nfrom typing import TYPE_CHECKING, Any, Dict\n\nif TYPE_CHECKING:\n    from eodag.api.product import EOProduct\n\n\nclass AssetsDict(UserDict):\n    \"\"\"A UserDict object listing assets contained in a\n    :class:`~eodag.api.product._product.EOProduct` resulting from a search.\n\n    :param product: Product resulting from a search\n    :type product: :class:`~eodag.api.product._product.EOProduct`\n    :param args: (optional) Arguments used to init the dictionary\n    :type args: Any\n    :param kwargs: (optional) Additional named-arguments used to init the dictionary\n    :type kwargs: Any\n    \"\"\"\n\n    product: EOProduct\n\n    def __init__(self, product: EOProduct, *args: Any, **kwargs: Any) -> None:\n        self.product = product\n        super(AssetsDict, self).__init__(*args, **kwargs)\n\n    def __setitem__(self, key: str, value: Dict[str, Any]) -> None:\n        super().__setitem__(key, Asset(self.product, key, value))\n\n    def as_dict(self) -> Dict[str, Any]:\n        \"\"\"Builds a representation of AssetsDict to enable its serialization\n\n        :returns: The representation of a :class:`~eodag.api.product._assets.AssetsDict`\n                  as a Python dict\n        :rtype: dict\n        \"\"\"\n        return {k: v.as_dict() for k, v in self.data.items()}\n\n\nclass Asset(UserDict):\n    \"\"\"A UserDict object containg one of the assets of a\n    :class:`~eodag.api.product._product.EOProduct` resulting from a search.\n\n    :param product: Product resulting from a search\n    :type product: :class:`~eodag.api.product._product.EOProduct`\n    :param key: asset key\n    :type key: str\n    :param args: (optional) Arguments used to init the dictionary\n    :type args: Any\n    :param kwargs: (optional) Additional named-arguments used to init the dictionary\n    :type kwargs: Any\n    \"\"\"\n\n    product: EOProduct\n\n    def __init__(self, product: EOProduct, key: str, *args: Any, **kwargs: Any) -> None:\n        self.product = product\n        self.key = key\n        super(Asset, self).__init__(*args, **kwargs)\n\n    def as_dict(self) -> Dict[str, Any]:\n        \"\"\"Builds a representation of Asset to enable its serialization\n\n        :returns: The representation of a :class:`~eodag.api.product._assets.Asset` as a\n                  Python dict\n        :rtype: dict\n        \"\"\"\n        return self.data\n\n    def download(self, **kwargs: Any) -> str:\n        \"\"\"Downloads a single asset\n\n        :param kwargs: (optional) Additional named-arguments passed to `plugin.download()`\n        :type kwargs: Any\n        :returns: The absolute path to the downloaded product on the local filesystem\n        :rtype: str\n        \"\"\"\n        return self.product.download(asset=self.key, **kwargs)",
  "start_line": 1,
  "end_line": 96,
  "total_lines": 96,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "eodag/api/product/_product.py",
  "content": "                        float(coords[1]),\n                        float(coords[0]),\n                        float(coords[3]),\n                        float(coords[2]),\n                    )\n        # Best effort to understand provider specific geometry (the default is to\n        # assume an object implementing the Geo Interface: see\n        # https://gist.github.com/2217756)\n        if isinstance(product_geometry, str):\n            try:\n                product_geometry = wkt.loads(product_geometry)\n            except (ShapelyError, GEOSException):\n                try:\n                    product_geometry = wkb.loads(product_geometry)\n                # Also catching TypeError because product_geometry can be a\n                # string and not a bytes string\n                except (ShapelyError, GEOSException, TypeError):\n                    # Giv up!\n                    raise\n        self.geometry = self.search_intersection = geometry.shape(product_geometry)\n        self.search_kwargs = kwargs\n        if self.search_kwargs.get(\"geometry\") is not None:\n            searched_geom = get_geometry_from_various(\n                **{\"geometry\": self.search_kwargs[\"geometry\"]}\n            )\n            try:\n                self.search_intersection = self.geometry.intersection(searched_geom)\n            except (GEOSException, ShapelyError):\n                logger.warning(\n                    \"Unable to intersect the requested extent: %s with the product \"\n                    \"geometry: %s\",\n                    searched_geom,\n                    product_geometry,\n                )\n                self.search_intersection = None\n        self.driver = self.get_driver()\n        self.downloader: Optional[Union[Api, Download]] = None\n        self.downloader_auth: Optional[Authentication] = None\n\n    def as_dict(self) -> Dict[str, Any]:\n        \"\"\"Builds a representation of EOProduct as a dictionary to enable its geojson\n        serialization\n\n        :returns: The representation of a :class:`~eodag.api.product._product.EOProduct` as a\n                  Python dict\n        :rtype: dict\n        \"\"\"\n        search_intersection = None\n        if self.search_intersection is not None:\n            search_intersection = geometry.mapping(self.search_intersection)\n\n        geojson_repr: Dict[str, Any] = {\n            \"type\": \"Feature\",\n            \"geometry\": geometry.mapping(self.geometry),\n            \"id\": self.properties[\"id\"],\n            \"assets\": self.assets.as_dict(),\n            \"properties\": {\n                \"eodag_product_type\": self.product_type,\n                \"eodag_provider\": self.provider,\n                \"eodag_search_intersection\": search_intersection,\n                **{\n                    key: value\n                    for key, value in self.properties.items()\n                    if key not in (\"geometry\", \"id\")\n                },\n            },\n        }\n\n        return geojson_repr\n\n    @classmethod\n    def from_geojson(cls, feature: Dict[str, Any]) -> EOProduct:\n        \"\"\"Builds an :class:`~eodag.api.product._product.EOProduct` object from its\n        representation as geojson\n\n        :param feature: The representation of a :class:`~eodag.api.product._product.EOProduct`\n                        as a Python dict\n        :type feature: dict\n        :returns: An instance of :class:`~eodag.api.product._product.EOProduct`\n        :rtype: :class:`~eodag.api.product._product.EOProduct`\n        \"\"\"\n        properties = feature[\"properties\"]\n        properties[\"geometry\"] = feature[\"geometry\"]\n        properties[\"id\"] = feature[\"id\"]\n        provider = feature[\"properties\"][\"eodag_provider\"]\n        product_type = feature[\"properties\"][\"eodag_product_type\"]\n        obj = cls(provider, properties, productType=product_type)\n        obj.search_intersection = geometry.shape(\n            feature[\"properties\"][\"eodag_search_intersection\"]\n        )\n        obj.assets = AssetsDict(obj, feature.get(\"assets\", {}))\n        return obj\n\n    # Implementation of geo-interface protocol (See\n    # https://gist.github.com/sgillies/2217756)\n    __geo_interface__ = property(as_dict)\n\n    def __repr__(self) -> str:\n        try:\n            return \"{}(id={}, provider={})\".format(\n                self.__class__.__name__, self.properties[\"id\"], self.provider\n            )\n        except KeyError as e:\n            raise MisconfiguredError(\n                f\"Unable to get {e.args[0]} key from EOProduct.properties\"\n            )\n\n    def register_downloader(\n        self, downloader: Union[Api, Download], authenticator: Optional[Authentication]\n    ) -> None:\n        \"\"\"Give to the product the information needed to download itself.\n\n        :param downloader: The download method that it can use\n        :type downloader: Concrete subclass of\n                          :class:`~eodag.plugins.download.base.Download` or\n                          :class:`~eodag.plugins.api.base.Api`\n        :param authenticator: The authentication method needed to perform the download\n        :type authenticator: Concrete subclass of\n                             :class:`~eodag.plugins.authentication.base.Authentication`\n        \"\"\"\n        self.downloader = downloader\n        self.downloader_auth = authenticator\n\n        # resolve locations and properties if needed with downloader configuration\n        location_attrs = (\"location\", \"remote_location\")\n        for location_attr in location_attrs:\n            try:\n                setattr(\n                    self,\n                    location_attr,\n                    urllib.parse.unquote(getattr(self, location_attr))\n                    % vars(self.downloader.config),\n                )\n            except ValueError as e:\n                logger.debug(\n                    f\"Could not resolve product.{location_attr} ({getattr(self, location_attr)})\"\n                    f\" in register_downloader: {str(e)}\"\n                )\n\n        for k, v in self.properties.items():\n            if isinstance(v, str):\n                try:\n                    if \"%\" in v:\n                        parsed = urllib.parse.urlparse(v)\n                        prop = urllib.parse.unquote(parsed.path) % vars(\n                            self.downloader.config\n                        )\n                        parsed = parsed._replace(path=urllib.parse.quote(prop))\n                        self.properties[k] = urllib.parse.urlunparse(parsed)\n                    else:",
  "start_line": 150,
  "end_line": 299,
  "total_lines": 541,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "eodag/rest/stac.py",
  "content": "    def get_stac_item_from_product(self, product: EOProduct) -> Dict[str, Any]:\n        \"\"\"Build STAC item from EODAG product\n\n        :param product: EODAG product\n        :type product: :class:`eodag.api.product._product.EOProduct`\n        :returns: STAC item\n        :rtype: list\n        \"\"\"\n        product_type = str(product.product_type)\n\n        item_model = self.__filter_item_model_properties(\n            self.stac_config[\"item\"], product_type\n        )\n        provider_model = deepcopy(self.stac_config[\"provider\"])\n\n        provider_dict = jsonpath_parse_dict_items(\n            provider_model,\n            {\"provider\": self.eodag_api.providers_config[product.provider].__dict__},\n        )\n\n        catalog = StacCatalog(\n            url=self.url.split(\"/items\")[0],\n            stac_config=self.stac_config,\n            root=self.root,\n            provider=self.provider,\n            eodag_api=self.eodag_api,\n            catalogs=[product_type],\n        )\n\n        product_dict = deepcopy(product.__dict__)\n        product_dict[\"assets\"] = product.assets.as_dict()\n\n        # parse jsonpath\n        product_item = jsonpath_parse_dict_items(\n            item_model,\n            {\n                \"product\": product_dict,\n                \"providers\": provider_dict,\n            },\n        )\n        # parse f-strings\n        format_args = deepcopy(self.stac_config)\n        format_args[\"catalog\"] = dict(\n            catalog.as_dict(), **{\"url\": catalog.url, \"root\": catalog.root}\n        )\n        format_args[\"item\"] = product_item\n        product_item = format_dict_items(product_item, **format_args)\n        product_item[\"bbox\"] = [float(i) for i in product_item[\"bbox\"]]\n\n        # remove empty properties\n        product_item = self.__filter_item_properties_values(product_item)\n\n        self.update_data(product_item)\n        return self.as_dict()\n\n\nclass StacCollection(StacCommon):\n    \"\"\"Stac collection object\n\n    :param url: Requested URL\n    :type url: str\n    :param stac_config: STAC configuration from stac.yml conf file\n    :type stac_config: dict\n    :param provider: (optional) Chosen provider\n    :type provider: str\n    :param eodag_api: EODAG python API instance\n    :type eodag_api: :class:`eodag.api.core.EODataAccessGateway`\n    :param root: (optional) API root\n    :type root: str\n    \"\"\"\n",
  "start_line": 500,
  "end_line": 570,
  "total_lines": 1387,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "eodag/rest/stac.py",
  "content": "            result_item_model[\"stac_extensions\"].remove(\n                self.stac_config[\"stac_extensions\"][\"sar\"]\n            )\n\n        # Filter out unneeded properties\n        extensions_prefixes = [\n            k\n            for k, v in self.stac_config[\"stac_extensions\"].items()\n            if v in result_item_model[\"stac_extensions\"]\n        ]\n        for k, v in item_model[\"properties\"].items():\n            # remove key if extension not in stac_extensions\n            if \":\" in k and k.split(\":\")[0] not in extensions_prefixes:\n                result_item_model[\"properties\"].pop(k, None)\n\n        return result_item_model\n\n    def __filter_item_properties_values(self, item: Dict[str, Any]) -> Dict[str, Any]:\n        \"\"\"Removes empty properties, unused extensions, and add missing extensions\n\n        :param item: STAC item data\n        :type item: dict\n        :returns: Filtered item model\n        :rtype: dict\n        \"\"\"\n        all_extensions_dict: Dict[str, str] = deepcopy(\n            self.stac_config[\"stac_extensions\"]\n        )\n        # parse f-strings with root\n        all_extensions_dict = format_dict_items(\n            all_extensions_dict, **{\"catalog\": {\"root\": self.root}}\n        )\n\n        item[\"stac_extensions\"] = []\n        # dict to list of keys to permit pop() while iterating\n        for k in list(item[\"properties\"]):\n            extension_prefix: str = k.split(\":\")[0]\n\n            if item[\"properties\"][k] is None:\n                item[\"properties\"].pop(k, None)\n            # feed found extensions list\n            elif (\n                extension_prefix in all_extensions_dict.keys()\n                and all_extensions_dict[extension_prefix] not in item[\"stac_extensions\"]\n            ):\n                # append path from item extensions\n                item[\"stac_extensions\"].append(all_extensions_dict[extension_prefix])\n\n        return item\n\n    def get_stac_item_from_product(self, product: EOProduct) -> Dict[str, Any]:",
  "start_line": 450,
  "end_line": 500,
  "total_lines": 1387,
  "is_truncated": false
}

{
  "error": "Invoking `write_file()` failed as the following mandatory input parameters are not present:\nfilepath\nYou could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."
}