Web3

Add description here about this class

ipfs_stac.client

Web3

__init__(local_gateway='localhost', api_port=5001, gateway_port=8080, stac_endpoint='https://stac.easierdata.info')

Web3 client constructor.

Parameters:

Name	Type	Description	Default
local_gateway	str	Local gateway endpoint without port. Defaults to "localhost".	'localhost'
api_port	int	API port. Defaults to 5001.	5001
gateway_port	int	Gateway port. Defaults to 8080.	8080
stac_endpoint	str	STAC browser endpoint. Defaults to "https://stac.easierdata.info".	'https://stac.easierdata.info'

getAssetFromItem(item, asset_name, fetch_data=False)

Returns asset object from item.

Parameters:

Name	Type	Description	Default
item	Item	STAC catalog item.	required
asset_name	str	Name of asset to return.	required
fetch_data	bool	Fetch data from CID. Defaults to False.	False

Returns:

Type	Description
Union[Asset, None]	Union[Asset, None]: Asset object.

getAssetNames(stac_obj)

Get a list of unique asset names from a STAC object.

Parameters:

Name	Type	Description	Default
stac_obj	Union[CollectionClient, ItemCollection, Item]	STAC object to get asset names from.	required

Returns:

Type	Description
Union[List[str], None]	Union[List[str], None]: A sorted list of unique asset names.

getAssetsFromItem(item, assets)

Returns array of asset objects from item.

Parameters:

Name	Type	Description	Default
item	Item	STAC catalog item.	required
assets	List[str]	Names of asset to return (strings).	required

Returns:

Type	Description
Union[List[Asset], None]	Union[List[Asset], None]: List of asset objects.

getCSVDataframeFromCID(cid)

Parse CSV CID to pandas dataframe.

Parameters:

Name	Type	Description	Default
cid	str	CID to retrieve.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: Parsed DataFrame.

getCollections()

Returns list of collections from STAC endpoint.

Returns:

Type	Description
Sequence[Collection]	Sequence[Collection]: List of collections.

getFromCID(cid)

Retrieves raw data from CID.

Parameters:

Name	Type	Description	Default
cid	str	CID to retrieve.	required

Returns:

Type	Description
Union[bytes, None]	Union[bytes, None]: Retrieved data or None if not found.

overwrite_config(path=None)

Overwrite configuration file with configuration in memory.

Parameters:

Name	Type	Description	Default
path	Optional[Path]	Path to configuration file (optional).	None

pinned_list(pin_type='recursive', names=False)

Fetch pinned CIDs from local node.

Parameters:

Name	Type	Description	Default
pin_type	str	The type of pinned keys to list. Can be "direct", "indirect", "recursive", or "all". Defaults to "recursive".	'recursive'
names	bool	Include pin names in the output. Defaults to False.	False

Returns:

Type	Description
Union[List[str], None]	Union[List[str], None]: List of pinned CIDs. If names is True, returns list of json objects.

searchSTAC(**kwargs)

Search STAC catalog for items using the search method from pystac-client.

Note: No request is sent to the API until a method is called to iterate through the resulting STAC Items, either ItemSearch.item_collections, ItemSearch.items, or ItemSearch.items_as_dicts.

Parameters:

Name	Type	Description	Default
kwargs		Keyword arguments for the search method.	{}

Returns:

Name	Type	Description
ItemCollection	ItemCollection	List of pystac.Item objects.

searchSTACByBox(bbox, collections)

Search STAC catalog by bounding box and return array of items.

Parameters:

Name	Type	Description	Default
bbox	List[float]	Array of coordinates for bounding box.	required
collections	List[str]	Array of collection names.	required

Returns:

Name	Type	Description
ItemCollection	ItemCollection	Collection of items.

searchSTACByBoxIndex(bbox, collections, index)

Search STAC catalog by bounding box and return singular item.

Parameters:

Name	Type	Description	Default
bbox	List[float]	Array of coordinates for bounding box.	required
collections	List[str]	Array of collection names.	required
index	int	Index of item to return.	required

Returns:

Name	Type	Description
Item	Item	STAC item.

shutdown_process()

Shutdown the IPFS daemon process.

startDaemon()

Start the IPFS daemon process.

Raises:

Type	Description
Exception	If the IPFS daemon fails to start.

uploadToIPFS(content, file_name=None, pin_content=False, mfs_path=None, chunker=None)

Uploads a file or bytes data to IPFS.

Parameters:

Name	Type	Description	Default
content	Union[str, Path, bytes]	The path to the file or bytes data to be uploaded.	required
file_name	Optional[str]	The name of the file. Defaults to None.	None
pin_content	bool	Pin locally to protect added files from garbage collection. Defaults to False.	False
mfs_path	Optional[str]	Add reference to Files API (MFS) at the provided path. Defaults to None.	None
chunker	Optional[str]	Chunking algorithm, size-[bytes], rabin-[min]-[avg]-[max] or buzhash. Defaults to None.	None

Raises:

Type	Description
ValueError	If neither file_path nor bytes_data is provided.
ValueError	If bytes_data is not of type bytes.
FileNotFoundError	If the file path provided does not exist.

Returns:

Name	Type	Description
str	None	The CID (Content Identifier) of the uploaded content.

writeCID(cid, filePath)

Write CID contents to local file system (WIP).

Parameters:

Name	Type	Description	Default
cid	str	CID to retrieve.	required
filePath	Union[str, Path]	Directory to write contents to.	required

Source Code

ipfs_stac.client.Web3

Source code in ipfs_stac/client.py

class Web3:
    def __init__(
        self,
        local_gateway: str = "localhost",
        api_port: int = 5001,
        gateway_port: int = 8080,
        stac_endpoint: str = "https://stac.easierdata.info",
    ) -> None:
        """
        Web3 client constructor.

        Args:
            local_gateway (str): Local gateway endpoint without port. Defaults to "localhost".
            api_port (int): API port. Defaults to 5001.
            gateway_port (int): Gateway port. Defaults to 8080.
            stac_endpoint (str): STAC browser endpoint. Defaults to "https://stac.easierdata.info".
        """
        self.local_gateway = local_gateway
        self.stac_endpoint = stac_endpoint
        self.daemon_status = None
        self.client: Client = Client.open(self.stac_endpoint)
        self.collections: List[str] = self._get_collections_ids()
        self.config = None

        self.api_port = api_port
        self.gateway_port = gateway_port

        if self.local_gateway:
            self.startDaemon()

        # When local gateway is `localhost``, ipfsspec does not play well with it.
        # This is a workaround to set the environment variable to the local gateway as `127.0.0.1`
        if self.local_gateway == "localhost":
            os.environ[ENV_VAR_NAME] = f"http://127.0.0.1:{self.gateway_port}"

        # Load configuration at instantiation
        # config_path = os.path.join(os.path.dirname(__file__), "config.json")
        # with open(config_path, "r") as f:
        #     self.config = json.load(f)

    def overwrite_config(self, path: Optional[Path] = None) -> None:
        """
        Overwrite configuration file with configuration in memory.

        Args:
            path (Optional[Path]): Path to configuration file (optional).
        """

        # Get user's home directory
        home = Path.home()
        if path:
            config_path = path
        else:
            config_path = Path(home, ".ipfs", "config")

        with Path.open(config_path, "w", encoding="utf-8") as f:
            json.dump(self.config, f)

    def _get_collections_ids(self) -> List[str]:
        """
        Get the collection ids from the STAC endpoint.

        Returns:
            List[str]: List of collection ids.
        """
        return [collection.id for collection in self.client.get_collections()]

    def getCollections(self) -> Sequence[Collection]:
        """
        Returns list of collections from STAC endpoint.

        Returns:
            Sequence[Collection]: List of collections.
        """
        return list(self.client.get_collections())

    def _is_process_running(self) -> bool:
        """
        Check if IPFS daemon process is running.

        Returns:
            bool: True if process is running, False otherwise.
        """
        process_name = "ipfs"
        for proc in psutil.process_iter(["name"]):
            try:
                if process_name.lower() in proc.info["name"].lower():
                    return True
            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
                pass
        return False

    def shutdown_process(self) -> None:
        """
        Shutdown the IPFS daemon process.
        """
        if self.daemon_status:
            print("Shutdown process is running...")

            self.daemon_status.terminate()
            self.daemon_status.wait(timeout=3)

            # If process is still running, kill it
            if self._is_process_running():
                self.daemon_status.kill()

            # If process is still running, exit with error
            elif self._is_process_running():
                import sys

                sys.exit("Failed to shutdown IPFS daemon")

            self.daemon_status = None

    def startDaemon(self) -> None:
        """
        Start the IPFS daemon process.

        Raises:
            Exception: If the IPFS daemon fails to start.
        """
        try:
            if not self._is_process_running():
                self.daemon_status = subprocess.Popen(["ipfs", "daemon"])
                atexit.register(self.shutdown_process)

            heartbeat_response = requests.post(
                f"http://{self.local_gateway}:{self.api_port}/api/v0/id",
                timeout=10,
            )
            if heartbeat_response.status_code != 200:
                warnings.warn(
                    "IPFS Daemon is running but still can't connect. Check your IPFS configuration."
                )
        except Exception as exc:
            print(f"Error starting IPFS daemon: {exc}")
            raise Exception("Failed to start IPFS daemon")

    def getFromCID(self, cid: str) -> Union[bytes, None]:
        """
        Retrieves raw data from CID.

        Args:
            cid (str): CID to retrieve.

        Returns:
            Union[bytes, None]: Retrieved data or None if not found.
        """
        content_cid = None
        try:
            content_cid = fetchCID(cid)
        except FileNotFoundError as e:
            print(f"Could not file with CID: {cid}. Are you sure it exists?")
            raise e
        return content_cid

    def searchSTACByBox(
        self, bbox: List[float], collections: List[str]
    ) -> ItemCollection:
        """
        Search STAC catalog by bounding box and return array of items.

        Args:
            bbox (List[float]): Array of coordinates for bounding box.
            collections (List[str]): Array of collection names.

        Returns:
            ItemCollection: Collection of items.
        """
        catalog = Client.open(self.stac_endpoint)
        search_results = catalog.search(
            collections=collections,
            bbox=bbox,
        )

        return search_results.item_collection()

    def searchSTAC(self, **kwargs) -> ItemCollection:
        """
        Search STAC catalog for items using the search method from pystac-client.

        Note: No request is sent to the API until a method is called to iterate
        through the resulting STAC Items, either `ItemSearch.item_collections`,
        `ItemSearch.items`, or `ItemSearch.items_as_dicts`.

        Args:
            kwargs: Keyword arguments for the search method.

        Returns:
            ItemCollection: List of pystac.Item objects.
        """
        try:
            search_results = self.client.search(**kwargs)
            return search_results.item_collection()

        except Exception as e:
            # Print the error message and the keyword argument that caused the error.
            if isinstance(e, TypeError):
                print(f"Error: {e}")
                print(f"Search method docstring: {self.client.search.__doc__}")
            else:
                print(f"Error: {e}")
            return ItemCollection([])

    def searchSTACByBoxIndex(
        self, bbox: List[float], collections: List[str], index: int
    ) -> Item:
        """
        Search STAC catalog by bounding box and return singular item.

        Args:
            bbox (List[float]): Array of coordinates for bounding box.
            collections (List[str]): Array of collection names.
            index (int): Index of item to return.

        Returns:
            Item: STAC item.
        """
        # Validate Bounding box coordinates before trying to search
        if (
            not isinstance(bbox, list)
            or len(bbox) != 4
            or not all(isinstance(coord, float) for coord in bbox)
        ):
            raise ValueError("bbox must be a list of four float numbers")

        catalog = Client.open(self.stac_endpoint)
        search_results = catalog.search(
            collections=collections,
            bbox=bbox,
        )

        return search_results.item_collection()[index]

    def getAssetNames(
        self, stac_obj: Union[CollectionClient, ItemCollection, Item]
    ) -> Union[List[str], None]:
        """
        Get a list of unique asset names from a STAC object.

        Args:
            stac_obj (Union[CollectionClient, ItemCollection, Item]): STAC object to get asset names from.

        Returns:
            Union[List[str], None]: A sorted list of unique asset names.
        """

        def get_asset_names_from_items(items: List[Item]) -> List[str]:
            """
            Get asset names from list of items.

            Args:
                items (List[Item]): List of STAC item objects.

            Returns:
                List[str]: A sorted list of unique asset names.
            """
            asset_names = set()
            for item in items:
                names = list(item.get_assets().keys())
                asset_names.update(names)
            return sorted(asset_names)

        if not stac_obj:
            raise ValueError(
                "STAC Object (CollectionClient, ItemCollection, Item) must be provided"
            )

        if isinstance(stac_obj, CollectionClient):
            try:
                items = stac_obj.get_items()
                return get_asset_names_from_items(list(items))
            except Exception as e:
                print(f"Error with getting asset names: {e}")
        elif isinstance(stac_obj, ItemCollection):
            try:
                items = stac_obj.items
                return get_asset_names_from_items(list(items))
            except Exception as e:
                print(f"Error with getting asset names: {e}")
        elif isinstance(stac_obj, Item):
            try:
                return sorted(stac_obj.get_assets().keys())
            except Exception as e:
                print(f"Error with getting asset names: {e}")
        else:
            raise ValueError(
                "STAC Object must be a Collection, Item, or ItemCollection"
            )

    def getAssetFromItem(
        self, item: Item, asset_name: str, fetch_data: bool = False
    ) -> Union["Asset", None]:
        """
        Returns asset object from item.

        Args:
            item (Item): STAC catalog item.
            asset_name (str): Name of asset to return.
            fetch_data (bool, optional): Fetch data from CID. Defaults to False.

        Returns:
            Union[Asset, None]: Asset object.
        """
        try:
            item_dict = item.to_dict()
            cid = item_dict["assets"][f"{asset_name}"]["alternate"]["IPFS"][
                "href"
            ].split("/")[-1]
            return Asset(
                cid,
                self.local_gateway,
                self.api_port,
                fetch_data=fetch_data,
                name=asset_name,
            )
        except Exception as e:
            print(f"Error with getting asset: {e}")

    def getAssetsFromItem(
        self, item: Item, assets: List[str]
    ) -> Union[List["Asset"], None]:
        """
        Returns array of asset objects from item.

        Args:
            item (Item): STAC catalog item.
            assets (List[str]): Names of asset to return (strings).

        Returns:
            Union[List[Asset], None]: List of asset objects.
        """
        try:
            assetArray = []

            for i in assets:
                assetArray.append(self.getAssetFromItem(item, i, fetch_data=False))

            return assetArray
        except Exception as e:
            print(f"Error with getting assets: {e}")

    def writeCID(self, cid: str, filePath: Union[str, Path]) -> None:
        """
        Write CID contents to local file system (WIP).

        Args:
            cid (str): CID to retrieve.
            filePath (Union[str, Path]): Directory to write contents to.
        """
        try:
            # Check filepath instance and convert to Path object if necessary
            if isinstance(filePath, str):
                filePath = Path(filePath).resolve()
            else:
                filePath = filePath.resolve()
            data_payload = bytes()
            with fsspec.open(f"ipfs://{cid}", "rb") as contents:
                data_payload = contents.read()
                # Write data to local file path
            with filePath.open("wb") as copy:
                copy.write(data_payload)
        except Exception as e:
            print(f"Error with CID write: {e}")

    def uploadToIPFS(
        self,
        content: Union[str, Path, bytes],
        file_name: Optional[str] = None,
        pin_content: bool = False,
        mfs_path: Optional[str] = None,
        chunker: Optional[str] = None,
    ) -> None:
        """
        Uploads a file or bytes data to IPFS.

        Args:
            content (Union[str, Path, bytes]): The path to the file or bytes data to be uploaded.
            file_name (Optional[str]): The name of the file. Defaults to None.
            pin_content (bool): Pin locally to protect added files from garbage collection. Defaults to False.
            mfs_path (Optional[str]): Add reference to Files API (MFS) at the provided path. Defaults to None.
            chunker (Optional[str]): Chunking algorithm, size-[bytes], rabin-[min]-[avg]-[max] or buzhash. Defaults to None.

        Raises:
            ValueError: If neither `file_path` nor `bytes_data` is provided.
            ValueError: If `bytes_data` is not of type bytes.
            FileNotFoundError: If the file path provided does not exist.

        Returns:
            str: The CID (Content Identifier) of the uploaded content.
        """

        # Setting param options
        param_options = f"cid-version=1&pin={pin_content}"
        if mfs_path:
            param_options = f"{param_options}&to-files={mfs_path}"
        if chunker:
            param_options = f"{param_options}&chunker={chunker}"

        # Define empty payload dictionary
        components = {"content": b"", "name": None}

        # Check the type of content and handle accordingly
        if isinstance(content, bytes):
            components["content"] = content
            if not file_name:
                components["name"] = None
        elif isinstance(content, (str, Path)):
            file_path = Path(content).resolve()
            if file_path.exists():
                with Path.open(file_path) as f:
                    components["content"] = f.read()
                    components["name"] = file_path.name
                # Override the file name if user provides one
                if file_name:
                    components["name"] = file_name
            else:
                raise FileNotFoundError(
                    f"The file path provided does not exist. Please check {content}"
                )
        else:
            raise ValueError("`content` must be of type `Union[str, Path, bytes]`.")

        # put the components together as a file payload
        if components["name"] is not None:
            file_payload = {"file": (components["name"], components["content"])}
        else:
            file_payload = {"file": components["content"]}

        try:
            response = requests.post(
                f"http://{self.local_gateway}:{self.api_port}/api/v0/add?{param_options}",
                files=file_payload,
                timeout=10,
            )
            response.raise_for_status()  # Raise an exception for HTTP errors

            # response.raise_for_status()  # Raise an exception for HTTP errors
            if response.status_code == 200:
                data = response.json()
                print(
                    f"Successfully added, {data['Name']}, to IPFS. CID: {data['Hash']}"
                )
                print(
                    f"Click here to view: http://{data['Hash']}.ipfs.{self.local_gateway}:{self.gateway_port}"
                )
                return data["Hash"]

        except requests.exceptions.Timeout:
            print("The request timed out")
        except requests.exceptions.RequestException as e:
            print(f"An error occurred: {e}")

    def pinned_list(
        self, pin_type: str = "recursive", names: bool = False
    ) -> Union[List[str], None]:
        """
        Fetch pinned CIDs from local node.

        Args:
            pin_type (str): The type of pinned keys to list. Can be "direct", "indirect", "recursive", or "all". Defaults to "recursive".
            names (bool): Include pin names in the output. Defaults to False.

        Returns:
            Union[List[str], None]: List of pinned CIDs. If `names` is True, returns list of json objects.
        """
        # Setting param options
        param_options = f"type={pin_type}&names={names}"
        response = requests.post(
            f"http://{self.local_gateway}:{self.api_port}/api/v0/pin/ls?{param_options}",
            timeout=10,
        )

        if response.status_code == 200:
            if response.json() != {}:
                if names:
                    return list(response.json())
                else:
                    return list(response.json()["Keys"].keys())
        else:
            print("Error fetching pinned CIDs")
            return [""]

    def getCSVDataframeFromCID(self, cid: str) -> pd.DataFrame:
        """
        Parse CSV CID to pandas dataframe.

        Args:
            cid (str): CID to retrieve.

        Returns:
            pd.DataFrame: Parsed DataFrame.
        """
        try:
            data = self.getFromCID(cid)

            # Parse for contents endpoint
            soup = BeautifulSoup(data, "html.parser")
            endpoint = f"{soup.find_all('a')[0].get('href').replace('.tech', '.io')}{soup.find_all('a')[-1].get('href')}"

            response = requests.get(endpoint, timeout=10)
            csv_data = StringIO(response.text)
            df = pd.read_csv(csv_data)

            return df
        except Exception as e:
            print(f"Error with dataframe retrieval: {e}")

        # Return an empty DataFrame
        return pd.DataFrame()