Utilities

def fuse_get_context() -> tuple[int, int, int]:
    """
    Get the current FUSE request context.
    
    Returns:
        Tuple of (uid, gid, pid) for the current request:
        - uid: User ID of the calling process
        - gid: Group ID of the calling process  
        - pid: Process ID of the calling process
    """

def fuse_exit() -> None:
    """
    Signal the FUSE main loop to exit.
    
    This function can be called from within filesystem operations
    to cleanly unmount and shutdown the FUSE filesystem.
    """

from mfusepy import Operations, fuse_get_context, fuse_exit
import os

class ContextAwareFS(Operations):
    def getattr(self, path, fh=None):
        # Get information about the calling process
        uid, gid, pid = fuse_get_context()
        
        # Log access information
        print(f"Process {pid} (uid={uid}, gid={gid}) accessing {path}")
        
        # Return different attributes based on caller
        if uid == 0:  # root user
            return {'st_mode': 0o644, 'st_nlink': 1, 'st_size': 1024}
        else:
            return {'st_mode': 0o444, 'st_nlink': 1, 'st_size': 512}
    
    def write(self, path, data, offset, fh):
        # Check if current user owns the file
        uid, gid, pid = fuse_get_context()
        file_uid = self._get_file_owner(path)
        
        if uid != file_uid and uid != 0:
            raise FuseOSError(errno.EACCES)
        
        return self._write_data(path, data, offset)
    
    def special_shutdown(self, path, fh):
        """Example of programmatic shutdown."""
        uid, gid, pid = fuse_get_context()
        
        if path == '/shutdown' and uid == 0:
            # Only allow root to shutdown
            fuse_exit()
        
        return 0

def get_fuse_version(libfuse) -> tuple[int, int]:
    """
    Get the version of the loaded FUSE library.
    
    Args:
        libfuse: FUSE library handle (ctypes CDLL object)
        
    Returns:
        Tuple of (major_version, minor_version)
    """

# Global version variables (set at import time)
fuse_version_major: int  # Major version of loaded FUSE library (2 or 3)
fuse_version_minor: int  # Minor version of loaded FUSE library

from mfusepy import get_fuse_version, fuse_version_major, fuse_version_minor
import ctypes

# Check global version info
print(f"FUSE version: {fuse_version_major}.{fuse_version_minor}")

# Version-specific behavior
class VersionAwareFS(Operations):
    def __init__(self):
        if fuse_version_major >= 3:
            # Use FUSE 3.x features
            self.supports_writeback_cache = True
            self.supports_readdirplus = True
        else:
            # FUSE 2.x compatibility mode
            self.supports_writeback_cache = False
            self.supports_readdirplus = False
    
    def init_with_config(self, conn_info, config_3):
        """FUSE 3.x initialization with configuration."""
        if fuse_version_major >= 3:
            # Configure FUSE 3.x specific options
            if config_3:
                config_3.contents.use_ino = 1
                config_3.contents.readdir_ino = 1

def time_of_timespec(ts, use_ns: bool = False) -> float:
    """
    Convert a timespec structure to a float timestamp.
    
    Args:
        ts: c_timespec structure
        use_ns: If True, return nanosecond precision timestamp
        
    Returns:
        Timestamp as float (seconds since epoch)
    """

def set_st_attrs(st, attrs: dict[str, Any], use_ns: bool = False) -> None:
    """
    Set stat structure attributes from a dictionary.
    
    Args:
        st: c_stat structure to populate
        attrs: Dictionary of attributes to set
        use_ns: If True, use nanosecond precision for timestamps
    """

from mfusepy import Operations, time_of_timespec, set_st_attrs, c_timespec, c_stat
import time
import ctypes

class TimestampFS(Operations):
    def utimens(self, path, times=None):
        """Update file timestamps."""
        if times is None:
            # Use current time
            now = time.time()
            atime = mtime = now
        else:
            atime, mtime = times
        
        # Convert to timespec structures
        atime_ts = c_timespec()
        atime_ts.tv_sec = int(atime)
        atime_ts.tv_nsec = int((atime - int(atime)) * 1e9)
        
        mtime_ts = c_timespec()  
        mtime_ts.tv_sec = int(mtime)
        mtime_ts.tv_nsec = int((mtime - int(mtime)) * 1e9)
        
        # Store timestamps
        self.files[path]['atime'] = atime_ts
        self.files[path]['mtime'] = mtime_ts
        
        return 0
    
    def getattr(self, path, fh=None):
        """Get file attributes with proper timestamp handling."""
        if path not in self.files:
            raise FuseOSError(errno.ENOENT)
        
        file_info = self.files[path]
        
        # Convert timespec to float timestamps
        atime = time_of_timespec(file_info['atime'], use_ns=True)
        mtime = time_of_timespec(file_info['mtime'], use_ns=True)
        
        return {
            'st_mode': file_info['mode'],
            'st_nlink': file_info['nlink'],
            'st_size': file_info['size'],
            'st_atime': atime,
            'st_mtime': mtime,
            'st_ctime': mtime,  # Use mtime for ctime
        }

# Platform-specific types (automatically selected based on system)
c_dev_t: type      # Device ID type
c_fsblkcnt_t: type # File system block count type  
c_fsfilcnt_t: type # File system file count type
c_gid_t: type      # Group ID type
c_uid_t: type      # User ID type
c_pid_t: type      # Process ID type
c_mode_t: type     # File mode type
c_off_t: type      # File offset type
c_ino_t: type      # Inode number type
c_nlink_t: type    # Link count type
c_blkcnt_t: type   # Block count type
c_blksize_t: type  # Block size type

# Function type aliases for extended attributes
setxattr_t: type   # Type for setxattr function pointer
getxattr_t: type   # Type for getxattr function pointer

# FUSE 3.x specific types
fuse_fill_dir_flags: type    # Directory filling flags (FUSE 3.x)
fuse_fill_dir_t: type        # Directory filling function type (FUSE 3.x)
fuse_readdir_flags: type     # Directory reading flags (FUSE 3.x)
fuse_buf_flags: type         # Buffer flags for FUSE operations
fuse_pollhandle_p: type      # Poll handle pointer type

# Platform-specific file locking
c_flock_t: type              # File locking structure (varies by platform)

# Error constants
ENOTSUP: int       # "Not supported" error code (platform-specific: 45 on BSD, 95 on Linux, 129/134 on Windows)

log: logging.Logger              # Logger for general FUSE operations ("fuse")
callback_logger: logging.Logger  # Logger for LoggingMixIn callbacks ("fuse.log-mixin")

_system: str        # Platform system name (Windows, Darwin, Linux, etc.)
_machine: str       # Architecture/machine type (x86_64, arm64, etc.)
_libfuse: ctypes.CDLL  # The loaded FUSE library handle
_libfuse_path: str     # Path to the loaded FUSE library

from mfusepy import log, callback_logger, Operations
import logging

class LoggedFS(Operations):
    def __init__(self):
        # Configure logging levels
        log.setLevel(logging.INFO)
        callback_logger.setLevel(logging.DEBUG)
        
        # Add custom handler
        handler = logging.StreamHandler()
        formatter = logging.Formatter('%(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        log.addHandler(handler)
    
    def getattr(self, path, fh=None):
        log.info(f"Getting attributes for {path}")
        
        try:
            attrs = self._get_file_attrs(path)
            log.debug(f"Retrieved attrs: {attrs}")
            return attrs
        except Exception as e:
            log.error(f"Error getting attributes for {path}: {e}")
            raise

from mfusepy import c_dev_t, c_mode_t, c_off_t, ENOTSUP
import os
import sys

class CrossPlatformFS(Operations):
    def __init__(self):
        # Adapt behavior based on platform
        self.is_windows = sys.platform.startswith('win')
        self.is_macos = sys.platform == 'darwin'
        self.is_linux = sys.platform.startswith('linux')
    
    def statfs(self, path):
        """Get filesystem statistics with platform adaptations."""
        try:
            if self.is_windows:
                # Windows-specific implementation
                return self._statfs_windows(path)
            elif self.is_macos:
                # macOS-specific implementation  
                return self._statfs_macos(path)
            else:
                # Linux/Unix implementation
                return self._statfs_unix(path)
        except NotImplementedError:
            raise FuseOSError(ENOTSUP)

from mfusepy import Operations, fuse_version_major
import threading
import time

class OptimizedFS(Operations):
    def __init__(self):
        self.cache = {}
        self.cache_lock = threading.RLock()
        self.cache_timeout = 5  # seconds
        
        # Use FUSE 3.x optimizations when available
        self.use_fuse3_features = fuse_version_major >= 3
    
    def _cached_getattr(self, path):
        """Cached file attribute lookup."""
        with self.cache_lock:
            cached_entry = self.cache.get(path)
            if cached_entry:
                attrs, timestamp = cached_entry
                if time.time() - timestamp < self.cache_timeout:
                    return attrs
            
            # Cache miss - fetch fresh attributes
            attrs = self._fetch_file_attrs(path)
            self.cache[path] = (attrs, time.time())
            return attrs
    
    def getattr(self, path, fh=None):
        return self._cached_getattr(path)
    
    def read_buf(self, path, bufpp, size, offset, fh):
        """Use FUSE 3.x buffer optimization when available."""
        if self.use_fuse3_features:
            # Use zero-copy buffer operations
            return self._read_buf_optimized(path, bufpp, size, offset, fh)
        else:
            # Fall back to regular read
            data = self.read(path, size, offset, fh)
            return len(data)