๐Ÿ“ฑ Device Info Provider

November 11, 2025 ยท View on GitHub

The Device Info Provider collects comprehensive system information from client devices during registration, enabling intelligent task assignment and device selection in constellation (multi-device) scenarios.

Device information is proactively collected during client registration and pushed to the server, reducing latency and enabling immediate task routing decisions.


๐Ÿ“‹ Overview

Core Capabilities:

CapabilityDescriptionUse Case
System DetectionAuto-detect OS, version, architecturePlatform-specific task routing
Hardware ProfilingCPU count, memory capacityResource-aware task assignment
Network DiscoveryHostname, IP addressNetwork topology mapping
Feature DetectionGUI, CLI, browser, office appsCapability-based device selection
ExtensibilityCustom metadata supportEnvironment-specific configuration

Supported Platforms:

PlatformStatusFeatures Detected
Windowsโœ… Full SupportGUI, CLI, browser, file system, office, Windows apps
Linuxโœ… Full SupportGUI, CLI, browser, file system, office, Linux apps
macOSโœ… Full SupportGUI, CLI, browser, file system, office
Mobile๐Ÿ”ฎ PlannedTouch, mobile apps, sensors
IoT๐Ÿ”ฎ PlannedSensors, actuators, limited resources

๐Ÿ—๏ธ Architecture

DeviceSystemInfo Dataclass

The device info structure captures essential information to minimize registration overhead:

classDiagram
    class DeviceSystemInfo {
        +string device_id
        +string platform
        +string os_version
        +int cpu_count
        +float memory_total_gb
        +string hostname
        +string ip_address
        +List~string~ supported_features
        +string platform_type
        +string schema_version
        +Dict custom_metadata
        +to_dict() Dict
    }
    
    class DeviceInfoProvider {
        +collect_system_info() DeviceSystemInfo
        -_get_platform() string
        -_get_os_version() string
        -_get_cpu_count() int
        -_get_memory_total_gb() float
        -_get_hostname() string
        -_get_ip_address() string
        -_detect_features() List~string~
        -_get_platform_type() string
    }
    
    DeviceInfoProvider ..> DeviceSystemInfo : creates

Field Reference:

FieldTypeDescriptionExample
device_idstrUnique client identifier"device_windows_001"
platformstrOS platform (lowercase)"windows", "linux", "darwin"
os_versionstrOS version string"10.0.19045" (Windows 10)
cpu_countintNumber of CPU cores8
memory_total_gbfloatTotal RAM in GB (rounded to 2 decimals)16.0
hostnamestrNetwork hostname"DESKTOP-ABC123"
ip_addressstrLocal IP address"192.168.1.100"
supported_featuresList[str]Detected capabilities["gui", "cli", "browser", "office"]
platform_typestrDevice category"computer", "mobile", "web", "iot"
schema_versionstrSchema version for compatibility"1.0"
custom_metadataDictUser-defined metadata{"environment": "production"}

๐Ÿ” Collection Process

Automatic Collection

from ufo.client.device_info_provider import DeviceInfoProvider

# Collect system information
system_info = DeviceInfoProvider.collect_system_info(
    client_id="device_windows_001",
    custom_metadata=None  # Or load from config
)

# Result: DeviceSystemInfo object
print(system_info.platform)         # "windows"
print(system_info.cpu_count)        # 8
print(system_info.memory_total_gb)  # 16.0
print(system_info.supported_features)  # ["gui", "cli", "browser", ...]

# Convert to dict for transmission
device_dict = system_info.to_dict()

Collection Flow:

sequenceDiagram
    participant Client
    participant DIP as Device Info Provider
    participant OS as Operating System
    
    Client->>DIP: collect_system_info(client_id, custom_metadata)
    
    par Collect Basic Info
        DIP->>OS: platform.system()
        OS-->>DIP: "Windows"
        
        DIP->>OS: platform.version()
        OS-->>DIP: "10.0.19045"
    and Collect Hardware Info
        DIP->>OS: os.cpu_count()
        OS-->>DIP: 8
        
        DIP->>OS: psutil.virtual_memory()
        OS-->>DIP: 16GB
    and Collect Network Info
        DIP->>OS: socket.gethostname()
        OS-->>DIP: "DESKTOP-ABC123"
        
        DIP->>OS: socket.getsockname()
        OS-->>DIP: "192.168.1.100"
    end
    
    DIP->>DIP: _detect_features()
    DIP->>DIP: _get_platform_type()
    
    DIP-->>Client: DeviceSystemInfo

๐ŸŽฏ Feature Detection

Platform-Specific Features

Features are automatically detected based on the platform to enable capability-based device selection.

Windows Features:

features = [
    "gui",           # Graphical user interface
    "cli",           # Command line interface
    "browser",       # Web browser support
    "file_system",   # File system operations
    "office",        # Office applications (Word, Excel, etc.)
    "windows_apps"   # Windows-specific applications
]

Linux Features:

features = [
    "gui",           # Graphical user interface (X11/Wayland)
    "cli",           # Bash/shell
    "browser",       # Firefox, Chrome, etc.
    "file_system",   # Linux file system
    "office",        # LibreOffice, etc.
    "linux_apps"     # Linux-specific applications
]

macOS Features:

features = [
    "gui",           # macOS GUI
    "cli",           # Terminal
    "browser",       # Safari, Chrome, etc.
    "file_system",   # macOS file system
    "office"         # Office for Mac
]

Feature Detection Logic:

PlatformDetected FeaturesRationale
windows, linux, darwinGUI, CLI, browser, file_system, officeDesktop/laptop computers have full capabilities
android, ios (future)Touch, mobile apps, cameraMobile-specific features
CustomUser-definedExtensible via custom_metadata

๐Ÿ’ก Usage Examples

Basic Collection

from ufo.client.device_info_provider import DeviceInfoProvider

# Collect with auto-detection
info = DeviceInfoProvider.collect_system_info(
    client_id="device_001",
    custom_metadata=None
)

print(f"Platform: {info.platform}")
print(f"CPU Cores: {info.cpu_count}")
print(f"Memory: {info.memory_total_gb} GB")
print(f"Features: {', '.join(info.supported_features)}")

With Custom Metadata

# Add environment-specific metadata
custom_meta = {
    "environment": "production",
    "datacenter": "us-east-1",
    "role": "automation_worker",
    "team": "qa"
}

info = DeviceInfoProvider.collect_system_info(
    client_id="device_prod_001",
    custom_metadata=custom_meta
)

# Custom metadata is preserved
print(info.custom_metadata["environment"])  # "production"

JSON Serialization

# Convert to dictionary for transmission
device_dict = info.to_dict()

# Serialize to JSON
import json
json_str = json.dumps(device_dict, indent=2)

# Example output:
# {
#   "device_id": "device_001",
#   "platform": "windows",
#   "os_version": "10.0.19045",
#   "cpu_count": 8,
#   "memory_total_gb": 16.0,
#   "hostname": "DESKTOP-ABC123",
#   "ip_address": "192.168.1.100",
#   "supported_features": ["gui", "cli", "browser", "file_system", "office", "windows_apps"],
#   "platform_type": "computer",
#   "schema_version": "1.0",
#   "custom_metadata": {}
# }

โš ๏ธ Error Handling

Graceful Degradation

If any detection method fails, the provider returns minimal info instead of crashing.

Error Handling Strategy:

try:
    # Attempt full collection
    return DeviceSystemInfo(...)
except Exception as e:
    logger.error(f"Error collecting system info: {e}", exc_info=True)
    # Return minimal info on error
    return DeviceSystemInfo(
        device_id=client_id,
        platform="unknown",
        os_version="unknown",
        cpu_count=0,
        memory_total_gb=0.0,
        hostname="unknown",
        ip_address="unknown",
        supported_features=[],
        platform_type="unknown",
        custom_metadata=custom_metadata or {}
    )

Individual Method Failures:

MethodFailure BehaviorFallback Value
_get_platform()Catch exception"unknown"
_get_os_version()Catch exception"unknown"
_get_cpu_count()Catch exception0
_get_memory_total_gb()psutil not installed or exception0.0
_get_hostname()Catch exception"unknown"
_get_ip_address()Primary method failsTry hostname resolution, then "unknown"

๐Ÿ”ง Memory Detection Details

psutil Dependency

!!!warning "Optional Dependency" Memory detection requires psutil. If not installed, memory will be reported as 0.0.

Installation:

pip install psutil

Detection Code:

@staticmethod
def _get_memory_total_gb() -> float:
    """Get total memory in GB"""
    try:
        import psutil
        total_memory = psutil.virtual_memory().total
        return round(total_memory / (1024**3), 2)  # Convert to GB, round to 2 decimals
    except ImportError:
        logger.warning("psutil not installed, memory info unavailable")
        return 0.0
    except Exception:
        return 0.0

๐ŸŒ IP Address Detection

Multi-Method Approach

!!!tip "Robust IP Detection" IP detection uses a two-stage approach for reliability.

Primary Method (Socket Connection):

# Connect to external address (doesn't actually send data)
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))  # Google DNS
ip = s.getsockname()[0]
s.close()

Fallback Method (Hostname Resolution):

# If primary fails, resolve via hostname
ip = socket.gethostbyname(socket.gethostname())

Final Fallback:

# If all methods fail
return "unknown"

๐Ÿš€ Integration Points

WebSocket Client Registration

The WebSocket client uses the Device Info Provider during registration:

# In websocket client's register_client()
from ufo.client.device_info_provider import DeviceInfoProvider

system_info = DeviceInfoProvider.collect_system_info(
    self.ufo_client.client_id,
    custom_metadata=None
)

metadata = {
    "system_info": system_info.to_dict(),
    "registration_time": datetime.now(timezone.utc).isoformat()
}

await self.registration_protocol.register_as_device(
    device_id=self.ufo_client.client_id,
    metadata=metadata,
    platform=self.ufo_client.platform
)

See WebSocket Client for complete registration flow details.

Agent Server

The server receives device info during registration and stores it in the agent profile:

# Server-side AgentProfile integration
device_info = registration_data["metadata"]["system_info"]
agent_profile.add_device(device_id, device_info)

See Server Quick Start for server-side processing details.


โœ… Best Practices

1. Add Custom Metadata for Environment Tracking

custom_meta = {
    "environment": os.getenv("ENVIRONMENT", "development"),
    "version": "1.0.0",
    "deployment_region": "us-west-2",
    "cost_center": "engineering"
}

system_info = DeviceInfoProvider.collect_system_info(
    client_id="device_001",
    custom_metadata=custom_meta
)

2. Install psutil for Accurate Memory Detection

pip install psutil

3. Use Descriptive Client IDs

# Include environment and location in client_id
client_id = f"device_{platform}_{env}_{location}_{instance_id}"
# Example: "device_windows_prod_us-west_001"

4. Log Collection Results

system_info = DeviceInfoProvider.collect_system_info(...)

logger.info(
    f"Collected device info: "
    f"platform={system_info.platform}, "
    f"cpu={system_info.cpu_count}, "
    f"memory={system_info.memory_total_gb}GB, "
    f"features={system_info.supported_features}"
)

5. Validate Before Sending

system_info = DeviceInfoProvider.collect_system_info(...)

# Validate essential fields
assert system_info.device_id, "Device ID required"
assert system_info.platform != "unknown", "Platform detection failed"
assert system_info.cpu_count > 0, "CPU detection failed"

๐Ÿš€ Next Steps