Asset

Add description here about this class

ipfs_stac.client

Asset

__init__(cid, local_gateway, api_port, fetch_data=False, name=None)

Constructor for asset object.

Parameters:

Name	Type	Description	Default
cid	str	The CID associated with the object.	required
local_gateway	str	Local gateway endpoint.	required
api_port	int	API port.	required
fetch_data	bool	Fetch data from CID. Defaults to False.	False
name	Optional[str]	Name of the asset. Defaults to None.	None

addToMFS(filename, mfs_path)

Add CID to MFS.

Parameters:

Name	Type	Description	Default
filename	str	Name of file.	required
mfs_path	str	Path in MFS.	required

fetch()

Fetch data for the asset.

pin()

Pin the asset to the local node.

to_np_ndarray(dtype=np.float32)

Convert asset data to numpy ndarray.

Parameters:

Name	Type	Description	Default
dtype	Union[dtype, type]	Data type for the ndarray. Defaults to np.float32.	float32

Returns:

Type	Description
ndarray	np.ndarray: Numpy array of the asset data.

Source Code

ipfs_stac.client.Asset

Source code in ipfs_stac/client.py

class Asset:
    def __init__(
        self,
        cid: str,
        local_gateway: str,
        api_port: int,
        fetch_data: bool = False,
        name: Optional[str] = None,
    ) -> None:
        """
        Constructor for asset object.

        Args:
            cid (str): The CID associated with the object.
            local_gateway (str): Local gateway endpoint.
            api_port (int): API port.
            fetch_data (bool): Fetch data from CID. Defaults to False.
            name (Optional[str]): Name of the asset. Defaults to None.
        """
        self.cid: str = cid
        self.local_gateway = local_gateway
        self.api_port = api_port
        self.data: Optional[bytes] = None
        self.is_pinned = False

        if name:
            self.name = name
        else:
            self.name = cid

        if fetch_data:
            self.fetch()

    def __str__(self) -> str:
        return self.cid

    def _is_pinned_to_local_node(self) -> bool:
        """
        Check if CID is pinned to local node.

        Returns:
            bool: True if pinned, False otherwise.
        """
        resp = requests.post(
            f"http://{self.local_gateway}:{self.api_port}/api/v0/pin/ls?arg=/ipfs/{self.cid}",
            timeout=10,
        )
        if (resp.json().get("Keys")) and self.cid in resp.json()["Keys"]:
            self.is_pinned = True
            return True
        elif resp.json()["Type"] == "error":
            return False
        else:
            print("Error checking if CID is pinned")
            print(resp.json())
            return False

    def fetch(self) -> None:
        """
        Fetch data for the asset.
        """
        try:
            self.data = fetchCID(self.cid)
        except Exception as e:
            print(f"Error with CID fetch: {e}")

    # Pin to local kubo node
    # @ensure_data_fetched
    def pin(self) -> None:
        """
        Pin the asset to the local node.
        """
        self._is_pinned_to_local_node()
        if self.is_pinned:
            print("Content is already pinned")
        else:
            response = requests.post(
                f"http://{self.local_gateway}:{self.api_port}/api/v0/pin/add?arg={self.cid}",
                timeout=10,
            )

            if response.status_code == 200:
                print("Data pinned successfully")
                self.is_pinned = True

            else:
                print("Error pinning data")

    def addToMFS(self, filename: str, mfs_path: str) -> None:
        """
        Add CID to MFS.

        Args:
            filename (str): Name of file.
            mfs_path (str): Path in MFS.
        """
        if filename is None or filename == "":
            filename = self.name

        response = requests.post(
            f"http://{self.local_gateway}:{self.api_port}/api/v0/files/cp?arg=/ipfs/{self.cid}&arg={mfs_path}/{filename}",
            timeout=10,
        )

        if response.status_code == 200:
            print("Data added to MFS successfully")
        else:
            print("Error adding data to MFS")

    # Returns asset as np array if image
    @ensure_data_fetched
    def to_np_ndarray(self, dtype: Union[np.dtype, type] = np.float32) -> np.ndarray:
        """
        Convert asset data to numpy ndarray.

        Args:
            dtype (Union[np.dtype, type]): Data type for the ndarray. Defaults to np.float32.

        Returns:
            np.ndarray: Numpy array of the asset data.
        """
        if self.data is None:
            raise ValueError("Data for asset has not been fetched yet")
        with rasterio.open(BytesIO(self.data)) as dataset:
            return dataset.read(1).astype(dtype)