Diff Coverage

def _tensor_data_descriptor():
    """Return the raw Tensor.data descriptor from Tensor or its base classes."""
    for tensor_cls in getattr(Tensor, "__mro__", ()):  # tolerate tests patching Tensor with a mock object
        descriptor = vars(tensor_cls).get("data")
        if descriptor is not None:
            return descriptor
    descriptor = getattr(Tensor, "data", None)
    if descriptor is not None and hasattr(descriptor, "__get__"):
        return descriptor
    raise AttributeError("Tensor.data descriptor is not available")


class DTensorBase(Tensor):
    """torch dtensor base"""

    @property
    # pylint: disable=C2801
    def data(self):
        data_descriptor = _tensor_data_descriptor()
        local_data = data_descriptor.__get__(self._local_tensor, type(self._local_tensor))
        # pylint: disable=C0415
        from hyper_parallel.core.dtensor.dtensor import DTensor
        return DTensor.from_local(
            local_data,
            self._device_mesh,
            self._alias_placements(),
        )

        requires_grad = getattr(self, "requires_grad", None)
        if isinstance(value, DTensorBase):
            local_value = value.to_local()
            if hasattr(value, "device_mesh"):
                self._device_mesh = value.device_mesh
            if hasattr(value, "placements"):
                self._placements = value.placements
            if hasattr(value, "layout"):
                self._layout = value.layout
        else:
            local_value = value
        data_descriptor = _tensor_data_descriptor()
        with getattr(torch, "_C").DisableTorchFunctionSubclass():

        with getattr(torch, "_C").DisableTorchFunctionSubclass():
            data_descriptor.__set__(self, local_value)
            data_descriptor.__set__(self._local_tensor, local_value)
        if requires_grad is not None and hasattr(self, "requires_grad_"):
            self.requires_grad_(requires_grad)

    def cpu(self):
        """Move the local shard to CPU while preserving DTensor metadata."""
        # pylint: disable=C0415
        from hyper_parallel.core.dtensor.dtensor import DTensor
        return DTensor.from_local(
            self._local_tensor.cpu(),
            self._device_mesh,
            self._alias_placements(),
        )

        ``dtensor.data.cpu().numpy()``. NumPy has no distributed layout
        inference path; delegate to the local tensor to keep Tensor-like local
        data semantics.
        """
        return self._local_tensor.numpy(*args, **kwargs)

    @property
    def dtype(self) -> torch.dtype:
        """