Core API

The core module provides the main station orchestrator and configuration.

QFZZStation

qfzz.core.station.QFZZStation

Main orchestrator for a QFZZ Radio Station.

Coordinates personalized DJ, dataset management, blockchain trust, edge optimization, and streaming capabilities.

Source code in qfzz/core/station.py

class QFZZStation:
    """
    Main orchestrator for a QFZZ Radio Station.

    Coordinates personalized DJ, dataset management, blockchain trust,
    edge optimization, and streaming capabilities.
    """

    def __init__(self, config: StationConfig):
        """
        Initialize QFZZ Station.

        Args:
            config: Station configuration
        """
        self.config = config
        self._running = False
        self._current_playlist: list[dict[str, Any]] = []
        self._listeners: dict[str, Any] = {}

        # Components will be initialized lazily
        self._dj = None
        self._dataset_manager = None
        self._trust_network = None
        self._edge_optimizer = None
        self._quantum_provider = None
        self._player = None

        logger.info(f"Initialized QFZZ Station: {config.station_name} ({config.station_id})")

    def start(self) -> None:
        """Start the radio station."""
        if self._running:
            logger.warning("Station is already running")
            return

        logger.info(f"Starting station: {self.config.station_name}")
        self._running = True
        self._initialize_components()
        logger.info("Station started successfully")

    def stop(self) -> None:
        """Stop the radio station."""
        if not self._running:
            logger.warning("Station is not running")
            return

        logger.info(f"Stopping station: {self.config.station_name}")
        self._running = False
        self._cleanup_components()
        logger.info("Station stopped successfully")

    def _initialize_components(self) -> None:
        """Initialize station components based on configuration."""
        # Import here to avoid circular dependencies
        from qfzz.datasets.manager import DatasetManager
        from qfzz.dj.personalized_dj import PersonalizedDJ
        from qfzz.streaming.player import MusicPlayer

        self._dj = PersonalizedDJ()
        self._dataset_manager = DatasetManager(allowed_licenses=self.config.allowed_licenses)
        self._player = MusicPlayer()

        if self.config.enable_blockchain:
            from qfzz.blockchain.trust_network import BlockchainTrustNetwork

            self._trust_network = BlockchainTrustNetwork()
            logger.info("Blockchain trust network enabled")

        if self.config.enable_edge_optimization:
            from qfzz.edge.optimizer import EdgeOptimizer

            self._edge_optimizer = EdgeOptimizer()
            logger.info("Edge optimization enabled")

        if self.config.enable_quantum:
            from qfzz.quantum.provider import QuantumProvider

            self._quantum_provider = QuantumProvider()
            if self._quantum_provider.is_available:
                logger.info("Quantum provider enabled (Microsoft QDK)")
            else:
                logger.info("Quantum provider enabled (classical fallback)")

    def _cleanup_components(self) -> None:
        """Cleanup station components."""
        self._dj = None
        self._dataset_manager = None
        self._trust_network = None
        self._edge_optimizer = None
        self._quantum_provider = None
        self._player = None

    def add_listener(self, user_id: str, preferences: Optional[dict[str, Any]] = None) -> None:
        """
        Add a listener to the station.

        Args:
            user_id: Unique user identifier
            preferences: Optional user preferences
        """
        if not self._running:
            raise RuntimeError("Station is not running")

        self._listeners[user_id] = {
            "user_id": user_id,
            "preferences": preferences or {},
            "connected_at": datetime.now().isoformat(),
            "playlist": [],
        }

        logger.info(f"Added listener: {user_id}")

    def remove_listener(self, user_id: str) -> None:
        """
        Remove a listener from the station.

        Args:
            user_id: User identifier to remove
        """
        if user_id in self._listeners:
            del self._listeners[user_id]
            logger.info(f"Removed listener: {user_id}")

    def generate_playlist(self, user_id: str) -> list[dict[str, Any]]:
        """
        Generate personalized playlist for a user.

        Args:
            user_id: User identifier

        Returns:
            List of track dictionaries
        """
        if not self._running:
            raise RuntimeError("Station is not running")

        if user_id not in self._listeners:
            raise ValueError(f"User {user_id} is not a listener")

        user_data = self._listeners[user_id]
        preferences = user_data.get("preferences", {})

        # Get recommendations from DJ
        recommendations = self._dj.recommend(user_id, preferences)

        # Filter by trust threshold if blockchain is enabled
        if self._trust_network:
            recommendations = [
                track
                for track in recommendations
                if self._trust_network.get_trust_score(
                    track.get("content_id", ""), track.get("creator_id", "")
                )
                >= self.config.trust_threshold
            ]

        # Limit playlist size
        playlist = recommendations[: self.config.max_playlist_size]

        # Store playlist for user
        self._listeners[user_id]["playlist"] = playlist

        logger.info(f"Generated playlist for {user_id}: {len(playlist)} tracks")
        return playlist

    def record_interaction(
        self, user_id: str, track_id: str, interaction_type: str, rating: Optional[float] = None
    ) -> None:
        """
        Record user interaction with a track.

        Args:
            user_id: User identifier
            track_id: Track identifier
            interaction_type: Type of interaction (play, skip, like, etc.)
            rating: Optional rating value
        """
        if not self._running:
            raise RuntimeError("Station is not running")

        if user_id not in self._listeners:
            raise ValueError(f"User {user_id} is not a listener")

        # Record with DJ for learning
        self._dj.record_feedback(user_id, track_id, interaction_type, rating)

        logger.debug(f"Recorded {interaction_type} for user {user_id} on track {track_id}")

    def get_station_stats(self) -> dict[str, Any]:
        """
        Get current station statistics.

        Returns:
            Dictionary of station statistics
        """
        return {
            "station_id": self.config.station_id,
            "station_name": self.config.station_name,
            "running": self._running,
            "listener_count": len(self._listeners),
            "blockchain_enabled": self.config.enable_blockchain,
            "edge_optimization_enabled": self.config.enable_edge_optimization,
            "quantum_enabled": self.config.enable_quantum,
            "trust_threshold": self.config.trust_threshold,
            "streaming_quality": self.config.streaming_quality,
        }

    def is_running(self) -> bool:
        """Check if station is running."""
        return self._running

__init__(config)

Initialize QFZZ Station.

Parameters:

Name	Type	Description	Default
config	StationConfig	Station configuration	required

Source code in qfzz/core/station.py

def __init__(self, config: StationConfig):
    """
    Initialize QFZZ Station.

    Args:
        config: Station configuration
    """
    self.config = config
    self._running = False
    self._current_playlist: list[dict[str, Any]] = []
    self._listeners: dict[str, Any] = {}

    # Components will be initialized lazily
    self._dj = None
    self._dataset_manager = None
    self._trust_network = None
    self._edge_optimizer = None
    self._quantum_provider = None
    self._player = None

    logger.info(f"Initialized QFZZ Station: {config.station_name} ({config.station_id})")

add_listener(user_id, preferences=None)

Add a listener to the station.

Parameters:

Name	Type	Description	Default
user_id	str	Unique user identifier	required
preferences	Optional[dict[str, Any]]	Optional user preferences	None

Source code in qfzz/core/station.py

def add_listener(self, user_id: str, preferences: Optional[dict[str, Any]] = None) -> None:
    """
    Add a listener to the station.

    Args:
        user_id: Unique user identifier
        preferences: Optional user preferences
    """
    if not self._running:
        raise RuntimeError("Station is not running")

    self._listeners[user_id] = {
        "user_id": user_id,
        "preferences": preferences or {},
        "connected_at": datetime.now().isoformat(),
        "playlist": [],
    }

    logger.info(f"Added listener: {user_id}")

generate_playlist(user_id)

Generate personalized playlist for a user.

Parameters:

Name	Type	Description	Default
user_id	str	User identifier	required

Returns:

Type	Description
list[dict[str, Any]]	List of track dictionaries

Source code in qfzz/core/station.py

def generate_playlist(self, user_id: str) -> list[dict[str, Any]]:
    """
    Generate personalized playlist for a user.

    Args:
        user_id: User identifier

    Returns:
        List of track dictionaries
    """
    if not self._running:
        raise RuntimeError("Station is not running")

    if user_id not in self._listeners:
        raise ValueError(f"User {user_id} is not a listener")

    user_data = self._listeners[user_id]
    preferences = user_data.get("preferences", {})

    # Get recommendations from DJ
    recommendations = self._dj.recommend(user_id, preferences)

    # Filter by trust threshold if blockchain is enabled
    if self._trust_network:
        recommendations = [
            track
            for track in recommendations
            if self._trust_network.get_trust_score(
                track.get("content_id", ""), track.get("creator_id", "")
            )
            >= self.config.trust_threshold
        ]

    # Limit playlist size
    playlist = recommendations[: self.config.max_playlist_size]

    # Store playlist for user
    self._listeners[user_id]["playlist"] = playlist

    logger.info(f"Generated playlist for {user_id}: {len(playlist)} tracks")
    return playlist

get_station_stats()

Get current station statistics.

Returns:

Type	Description
dict[str, Any]	Dictionary of station statistics

Source code in qfzz/core/station.py

def get_station_stats(self) -> dict[str, Any]:
    """
    Get current station statistics.

    Returns:
        Dictionary of station statistics
    """
    return {
        "station_id": self.config.station_id,
        "station_name": self.config.station_name,
        "running": self._running,
        "listener_count": len(self._listeners),
        "blockchain_enabled": self.config.enable_blockchain,
        "edge_optimization_enabled": self.config.enable_edge_optimization,
        "quantum_enabled": self.config.enable_quantum,
        "trust_threshold": self.config.trust_threshold,
        "streaming_quality": self.config.streaming_quality,
    }

is_running()

Check if station is running.

Source code in qfzz/core/station.py

def is_running(self) -> bool:
    """Check if station is running."""
    return self._running

record_interaction(user_id, track_id, interaction_type, rating=None)

Record user interaction with a track.

Parameters:

Name	Type	Description	Default
user_id	str	User identifier	required
track_id	str	Track identifier	required
interaction_type	str	Type of interaction (play, skip, like, etc.)	required
rating	Optional[float]	Optional rating value	None

Source code in qfzz/core/station.py

def record_interaction(
    self, user_id: str, track_id: str, interaction_type: str, rating: Optional[float] = None
) -> None:
    """
    Record user interaction with a track.

    Args:
        user_id: User identifier
        track_id: Track identifier
        interaction_type: Type of interaction (play, skip, like, etc.)
        rating: Optional rating value
    """
    if not self._running:
        raise RuntimeError("Station is not running")

    if user_id not in self._listeners:
        raise ValueError(f"User {user_id} is not a listener")

    # Record with DJ for learning
    self._dj.record_feedback(user_id, track_id, interaction_type, rating)

    logger.debug(f"Recorded {interaction_type} for user {user_id} on track {track_id}")

remove_listener(user_id)

Remove a listener from the station.

Parameters:

Name	Type	Description	Default
user_id	str	User identifier to remove	required

Source code in qfzz/core/station.py

def remove_listener(self, user_id: str) -> None:
    """
    Remove a listener from the station.

    Args:
        user_id: User identifier to remove
    """
    if user_id in self._listeners:
        del self._listeners[user_id]
        logger.info(f"Removed listener: {user_id}")

start()

Start the radio station.

Source code in qfzz/core/station.py

def start(self) -> None:
    """Start the radio station."""
    if self._running:
        logger.warning("Station is already running")
        return

    logger.info(f"Starting station: {self.config.station_name}")
    self._running = True
    self._initialize_components()
    logger.info("Station started successfully")

stop()

Stop the radio station.

Source code in qfzz/core/station.py

def stop(self) -> None:
    """Stop the radio station."""
    if not self._running:
        logger.warning("Station is not running")
        return

    logger.info(f"Stopping station: {self.config.station_name}")
    self._running = False
    self._cleanup_components()
    logger.info("Station stopped successfully")

StationConfig

qfzz.core.config.StationConfig dataclass

Configuration for a QFZZ Radio Station.

Attributes:

Name	Type	Description
station_id	str	Unique identifier for the station
station_name	str	Human-readable station name
max_playlist_size	int	Maximum number of tracks in playlist
trust_threshold	float	Minimum trust score for content (0.0-1.0)
enable_blockchain	bool	Enable blockchain trust network
enable_edge_optimization	bool	Enable edge device optimization
streaming_quality	str	Default streaming quality
allowed_licenses	list[str]	List of acceptable content licenses
cache_size_mb	int	Cache size in megabytes
metadata	dict[str, Any]	Additional configuration metadata

Source code in qfzz/core/config.py

@dataclass
class StationConfig:
    """
    Configuration for a QFZZ Radio Station.

    Attributes:
        station_id: Unique identifier for the station
        station_name: Human-readable station name
        max_playlist_size: Maximum number of tracks in playlist
        trust_threshold: Minimum trust score for content (0.0-1.0)
        enable_blockchain: Enable blockchain trust network
        enable_edge_optimization: Enable edge device optimization
        streaming_quality: Default streaming quality
        allowed_licenses: List of acceptable content licenses
        cache_size_mb: Cache size in megabytes
        metadata: Additional configuration metadata
    """

    station_id: str
    station_name: str
    max_playlist_size: int = 100
    trust_threshold: float = 0.6
    enable_blockchain: bool = True
    enable_edge_optimization: bool = True
    streaming_quality: str = "high"
    allowed_licenses: list[str] = field(default_factory=lambda: ["CC-BY", "CC-BY-SA", "CC0"])
    enable_quantum: bool = False
    cache_size_mb: int = 500
    metadata: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self):
        """Validate configuration after initialization."""
        self.validate()

    def validate(self) -> None:
        """
        Validate configuration parameters.

        Raises:
            ValueError: If any configuration parameter is invalid
        """
        if not self.station_id or not isinstance(self.station_id, str):
            raise ValueError("station_id must be a non-empty string")

        if not self.station_name or not isinstance(self.station_name, str):
            raise ValueError("station_name must be a non-empty string")

        if self.max_playlist_size < 1:
            raise ValueError("max_playlist_size must be positive")

        if not 0.0 <= self.trust_threshold <= 1.0:
            raise ValueError("trust_threshold must be between 0.0 and 1.0")

        if self.streaming_quality not in ["low", "medium", "high", "lossless"]:
            raise ValueError("streaming_quality must be one of: low, medium, high, lossless")

        if not isinstance(self.enable_quantum, bool):
            raise ValueError("enable_quantum must be a boolean")

        if self.cache_size_mb < 0:
            raise ValueError("cache_size_mb must be non-negative")

    def to_dict(self) -> dict[str, Any]:
        """Convert configuration to dictionary."""
        return {
            "station_id": self.station_id,
            "station_name": self.station_name,
            "max_playlist_size": self.max_playlist_size,
            "trust_threshold": self.trust_threshold,
            "enable_blockchain": self.enable_blockchain,
            "enable_edge_optimization": self.enable_edge_optimization,
            "enable_quantum": self.enable_quantum,
            "streaming_quality": self.streaming_quality,
            "allowed_licenses": self.allowed_licenses,
            "cache_size_mb": self.cache_size_mb,
            "metadata": self.metadata,
        }

    def to_json(self) -> str:
        """Convert configuration to JSON string."""
        return json.dumps(self.to_dict(), indent=2)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "StationConfig":
        """Create configuration from dictionary."""
        return cls(**data)

    @classmethod
    def from_json(cls, json_str: str) -> "StationConfig":
        """Create configuration from JSON string."""
        data = json.loads(json_str)
        return cls.from_dict(data)

__post_init__()

Validate configuration after initialization.

Source code in qfzz/core/config.py

def __post_init__(self):
    """Validate configuration after initialization."""
    self.validate()

from_dict(data) classmethod

Create configuration from dictionary.

Source code in qfzz/core/config.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "StationConfig":
    """Create configuration from dictionary."""
    return cls(**data)

from_json(json_str) classmethod

Create configuration from JSON string.

Source code in qfzz/core/config.py

@classmethod
def from_json(cls, json_str: str) -> "StationConfig":
    """Create configuration from JSON string."""
    data = json.loads(json_str)
    return cls.from_dict(data)

to_dict()

Convert configuration to dictionary.

Source code in qfzz/core/config.py

def to_dict(self) -> dict[str, Any]:
    """Convert configuration to dictionary."""
    return {
        "station_id": self.station_id,
        "station_name": self.station_name,
        "max_playlist_size": self.max_playlist_size,
        "trust_threshold": self.trust_threshold,
        "enable_blockchain": self.enable_blockchain,
        "enable_edge_optimization": self.enable_edge_optimization,
        "enable_quantum": self.enable_quantum,
        "streaming_quality": self.streaming_quality,
        "allowed_licenses": self.allowed_licenses,
        "cache_size_mb": self.cache_size_mb,
        "metadata": self.metadata,
    }

to_json()

Convert configuration to JSON string.

Source code in qfzz/core/config.py

def to_json(self) -> str:
    """Convert configuration to JSON string."""
    return json.dumps(self.to_dict(), indent=2)

validate()

Validate configuration parameters.

Raises:

Type	Description
ValueError	If any configuration parameter is invalid

Source code in qfzz/core/config.py

def validate(self) -> None:
    """
    Validate configuration parameters.

    Raises:
        ValueError: If any configuration parameter is invalid
    """
    if not self.station_id or not isinstance(self.station_id, str):
        raise ValueError("station_id must be a non-empty string")

    if not self.station_name or not isinstance(self.station_name, str):
        raise ValueError("station_name must be a non-empty string")

    if self.max_playlist_size < 1:
        raise ValueError("max_playlist_size must be positive")

    if not 0.0 <= self.trust_threshold <= 1.0:
        raise ValueError("trust_threshold must be between 0.0 and 1.0")

    if self.streaming_quality not in ["low", "medium", "high", "lossless"]:
        raise ValueError("streaming_quality must be one of: low, medium, high, lossless")

    if not isinstance(self.enable_quantum, bool):
        raise ValueError("enable_quantum must be a boolean")

    if self.cache_size_mb < 0:
        raise ValueError("cache_size_mb must be non-negative")

Usage Example

from qfzz import QFZZStation, StationConfig

# Create configuration
config = StationConfig(
    station_name="My Station",
    edge_mode=True,
    blockchain_enabled=True
)

# Initialize and start station
station = QFZZStation(config)
station.initialize()
station.start()

# Get status
status = station.get_status()
print(status)

# Stop station
station.stop()