Filesystem

def __truediv__(self, path_str: str) -> DevicePath:
    """Join a path component to the root directory."""
    return self.root / path_str

def close(self):
    """Close connection with device filesystem."""
    self.manager.disconnect()

def delete_all_files(self, confirm: bool = False) -> None:
    """Delete all files on the device.

    Parameters
    ----------
    confirm : bool, optional
        If True, clear all files on the device. Default is False.
        This acts as a sentinel to avoid accidental deletes in
        interactive REPL or Jupyter environments.
    """
    if confirm:
        with self.manager._lock():
            self._client_connection.ClearDeviceFiles()

def exists(self, path: str | DevicePath) -> bool:
    """Check whether a path exists on the device.

    Parameters
    ----------
    path : str | DevicePath
        The file or directory path to check.

    Returns
    -------
    bool
        True if the path exists, False otherwise.
    """
    if not isinstance(path, DevicePath):
        path = DevicePath(path)

    if str(path) == '':  # root always exists
        return True

    for entry in self.iterdir(path.parent):
        if entry == path:  # check if direct match
            return True

        if entry.is_relative_to(path):  # check for subdirectory match
            return True

    return False

def free(self) -> int:
    """Return free space on filesystem.

    Returns
    -------
    int
        Free space in bytes (1 kB = 1024 bytes).
    """
    with self.manager._lock():
        return self._client_connection.GetDeviceFree()

def iterdir(self, directory: DevicePath | str | None = None) -> Iterator[DevicePath]:
    """Iterate over entries in a device directory.

    Note that some devices like EmStat4 return all subdirectories and files
    at once.

    Parameters
    ----------
    directory : DevicePath | str | None, optional
        The directory to iterate. Defaults to the root directory.

    Yields
    ------
    paths: Iterator[DevicePath]
        Path objects for each entry in the directory.
    """
    paths = self._get_device_files(directory=directory)

    for path in paths:
        yield DevicePath(path.Dir, path.Name)

def listdir(self, directory: DevicePath | str | None = None) -> list[DevicePath]:
    """List all entries in a device directory.

    Note that some devices like EmStat4 return all subdirectories and files
    at once.

    Parameters
    ----------
    directory : DevicePath | str | None, optional
        The directory to iterate. Defaults to the root directory.

    Yields
    ------
    paths: Iterator[DevicePath]
        Path objects for each entry in the directory.
    """
    return list(self.iterdir(directory=directory))

def load_measurement(self, path: str | DevicePath) -> Measurement:
    """Load measurement from path on device.

    Parameters
    ----------
    path : str | DevicePath
        The file path to load.

    Returns
    -------
    Measurement
        Measurement data.
    """
    f = self._get_device_file(path)

    psmeasurement = self._client_connection.LoadDeviceFile(f)

    return Measurement(psmeasurement=psmeasurement)

def open(self):
    """Open connection with device filesystem."""
    self.manager.connect()

def read_text(self, path: DevicePath) -> str:
    """Read the content of a file as text.

    Parameters
    ----------
    path : DevicePath
        The file path.

    Returns
    -------
    str
        File contents as a string.
    """
    f = self._get_device_file(path)

    return f.Content

def remove(self, path: str | DevicePath) -> None:
    """Remove a file from the device.

    Parameters
    ----------
    path : str | DevicePath
        The file path to remove.
    """
    if isinstance(path, str):
        path = DevicePath(path)

    with self.manager._lock():
        self._client_connection.DeleteDeviceFile(str(path))

def size(self) -> int:
    """Return total size of filesystem.

    Returns
    -------
    int
        Total size in bytes.
    """
    with self.manager._lock():
        return self._client_connection.GetDeviceSize()

def size_of(self, path: str | DevicePath) -> int:
    """Get the size of a file.

    Parameters
    ----------
    path : str | DevicePath
        The file path.

    Returns
    -------
    int
        File size in bytes (1 kB = 1024 bytes).
    """
    f = self._get_device_file(path)

    return f.Size

def timestamp_of(self, path: str | DevicePath) -> str:
    """Get the modification timestamp of a file.

    Parameters
    ----------
    path : str | DevicePath
        The file path.

    Returns
    -------
    str
        Timestamp in ISO format.
    """
    f = self._get_device_file(path)

    return f.Timestamp.ToString('s', System.Globalization.CultureInfo.InvariantCulture)

def walk(self, directory: DevicePath | str | None = None) -> Iterator[DevicePath]:
    """Generate file names by walking the directory tree starting from a device directory.

    Parameters
    ----------
    directory : DevicePath | str | None, optional
        The directory to walk through. Defaults to the root directory.

    Yields
    ------
    paths: Iterator[DevicePath]
        Path objects for each entry in the directory.
    """
    paths = self._get_device_files(directory=directory)

    for path in paths:
        df = DevicePath(path.Dir, path.Name)

        if str(path.Type) == 'Folder':
            yield from self.walk(str(df))
        else:
            yield df

def __truediv__(self, path_str: str) -> DevicePath:
    """Join a path component to the root directory."""
    return self.root / path_str

async def close(self):
    """Close connection with device filesystem."""
    await self.manager.disconnect()

async def delete_all_files(self, confirm: bool = False) -> None:
    """Delete all files on the device.

    Parameters
    ----------
    confirm : bool, optional
        If True, clear all files on the device. Default is False.
        This acts as a sentinel to avoid accidental deletes in
        interactive REPL or Jupyter environments.
    """
    if confirm:
        async with self.manager._lock():
            await create_future(self._client_connection.ClearDeviceFilesAsync())

async def exists(self, path: str | DevicePath) -> bool:
    """Check whether a path exists on the device.

    Parameters
    ----------
    path : str | DevicePath
        The file or directory path to check.

    Returns
    -------
    bool
        True if the path exists, False otherwise.
    """
    if not isinstance(path, DevicePath):
        path = DevicePath(path)

    if str(path) == '':  # root always exists
        return True

    async for entry in self.iterdir(path.parent):
        if entry == path:  # check if direct match
            return True

        if entry.is_relative_to(path):  # check for subdirectory match
            return True

    return False

async def free(self) -> int:
    """Return free space on filesystem.

    Returns
    -------
    int
        Free space in kB (1 kB = 1024 bytes).
    """
    async with self.manager._lock():
        return await create_future(self._client_connection.GetDeviceFreeAsync())

async def iterdir(
    self, directory: DevicePath | str | None = None
) -> AsyncIterator[DevicePath]:
    """Iterate over entries in a device directory.

    Note that some devices like EmStat4 return all subdirectories and files
    at once.

    Parameters
    ----------
    directory : DevicePath | str | None, optional
        The directory to iterate. Defaults to the root directory.

    Yields
    ------
    paths: Iterator[DevicePath]
        Path objects for each entry in the directory.
    """
    paths = await self._get_device_files(directory=directory)

    for path in paths:
        yield DevicePath(path.Dir, path.Name)

async def listdir(self, directory: DevicePath | str | None = None) -> list[DevicePath]:
    """List all entries in a device directory.

    Note that some devices like EmStat4 return all subdirectories and files
    at once.

    Parameters
    ----------
    directory : DevicePath | str | None, optional
        The directory to iterate. Defaults to the root directory.

    Yields
    ------
    paths: Iterator[DevicePath]
        Path objects for each entry in the directory.
    """
    return [_ async for _ in self.iterdir(directory=directory)]

async def load_measurement(self, path: str | DevicePath) -> Measurement:
    """Load measurement from path on device.

    Parameters
    ----------
    path : str | DevicePath
        The file path to load.

    Returns
    -------
    Measurement
        Measurement data.
    """
    f = await self._get_device_file(path)

    async with self.manager._lock():
        psmeasurement: PalmSens.Measurement = await create_future(
            self._client_connection.LoadDeviceFileAsync(f)
        )

    return Measurement(psmeasurement=psmeasurement)

async def open(self):
    """Open connection with device filesystem."""
    await self.manager.connect()

async def read_text(self, path: DevicePath) -> str:
    """Read the content of a file as text.

    Parameters
    ----------
    path : DevicePath
        The file path.

    Returns
    -------
    str
        File contents as a string.
    """
    f = await self._get_device_file(path)

    return f.Content

async def remove(self, path: str | DevicePath) -> None:
    """Remove a file from the device.

    Parameters
    ----------
    path : str | DevicePath
        The file path to remove.
    """
    if isinstance(path, str):
        path = DevicePath(path)

    async with self.manager._lock():
        await create_future(self._client_connection.DeleteDeviceFileAsync(str(path)))

async def size(self) -> int:
    """Return total size of filesystem.

    Returns
    -------
    int
        Total size in kB (1 kB = 1024 bytes).
    """
    async with self.manager._lock():
        return await create_future(self._client_connection.GetDeviceSizeAsync())

async def size_of(self, path: str | DevicePath) -> int:
    """Get the size of a file.

    Parameters
    ----------
    path : str | DevicePath
        The file path.

    Returns
    -------
    int
        File size in bytes.
    """
    f = await self._get_device_file(path)

    return f.Size

async def timestamp_of(self, path: str | DevicePath) -> str:
    """Get the modification timestamp of a file.

    Parameters
    ----------
    path : str | DevicePath
        The file path.

    Returns
    -------
    str
        Timestamp in ISO format.
    """
    f = await self._get_device_file(path)

    return f.Timestamp.ToString('s', System.Globalization.CultureInfo.InvariantCulture)

async def walk(
    self, directory: DevicePath | str | None = None
) -> AsyncIterator[DevicePath]:
    """Generate file names by walking the directory tree starting from a device directory.

    Parameters
    ----------
    directory : DevicePath | str | None, optional
        The directory to walk through. Defaults to the root directory.

    Yields
    ------
    paths: Iterator[DevicePath]
        Path objects for each entry in the directory.
    """
    paths = await self._get_device_files(directory=directory)

    for path in paths:
        df = DevicePath(path.Dir, path.Name)

        if str(path.Type) == 'Folder':
            async for item in self.walk(str(df)):
                yield item
        else:
            yield df