Components Layer

Purpose: Contain heavy, domain-specific logic (analytics, tagging, ML, etc.).

Components are the workhorses that do the real computational work:

• ML inference and embeddings
• Tag aggregation and resolution
• Statistical analysis
• Complex data transformations

Rule: Heavy business logic lives here. Wiring lives in services. Control flow lives in workflows.

Directory Structure

components/
├── analytics/          # Tag statistics, co-occurrence
├── ml/                 # ML inference, embeddings, calibration
│   └── ml_backend_essentia_comp.py  # ONLY file that imports Essentia
├── tagging/            # Tag parsing, writing, aggregation
├── queue/              # Queue operations
├── library/            # Library file operations
├── metadata/           # Metadata extraction
├── platform/           # GPU, bootstrap
└── workers/            # Worker crash handling, job recovery

Allowed Imports

# ✅ Allowed
from nomarr.persistence import Database
from nomarr.helpers.dto import ProcessFileResult
from nomarr.components.ml.model_loading import load_model  # Cross-component OK
from nomarr.helpers.files_helper import discover_audio_files

Forbidden Imports

# ❌ NEVER import these in components
from nomarr.services import ...      # No services
from nomarr.workflows import ...     # No workflows
from nomarr.interfaces import ...    # No interfaces
from pydantic import BaseModel       # No Pydantic

# ❌ ESSENTIA ONLY IN backend_essentia.py
import essentia_tensorflow  # Only in ml_backend_essentia_comp.py

Essentia Isolation Rule

Only

nomarr/components/ml/ml_backend_essentia_comp.py

All other ML code calls functions in that file:

# In ml_backend_essentia_comp.py:
try:
    import essentia_tensorflow as essentia_tf
except ImportError:
    essentia_tf = None

def compute_embeddings_essentia(file_path: str, ...) -> np.ndarray:
    if essentia_tf is None:
        raise RuntimeError("Essentia not installed")
    # ... use essentia_tf

LibraryPath Authority

path_comp

LibraryPath

All raw user paths and DB paths must flow through:

• build_library_path_from_input()

  — for API/CLI/user-provided paths

• build_library_path_from_db()

  — for paths retrieved from database

Components are responsible for:

• Calling factory functions before any filesystem access
• Checking

  path.is_valid()

  before proceeding
• Never accepting raw path strings where

  LibraryPath

  exists

Raw strings are never acceptable substitutes for

LibraryPath

Function Style

Components prefer stateless, pure functions over classes:

# ✅ Good - stateless function
def compute_tag_statistics(db: Database, library_id: int) -> TagStats:
    rows = _query_tag_data(db, library_id)
    stats = _aggregate_tag_stats(rows)
    return TagStats(stats)

# ❌ Avoid - stateful class (unless truly needed)
class TagStatsComputer:
    def __init__(self, db: Database):
        self.db = db  # Unnecessary state

Naming Conventions

• Module:

  snake_case_comp.py

  (e.g.,

  analytics_comp.py

  ,

  ml_embed_comp.py

  )
• Public functions:

  compute_embeddings

  ,

  aggregate_mood_tags

• Private helpers:

  _load_model

  ,

  _format_tag_stats

Complexity & Private Helpers

Components are where large, complex functions are acceptable if well-structured.

Use

_private

# Before - too large
def compute_tag_statistics(db: Database, library_id: int) -> TagStats:
    # 40 lines of querying
    # 40 lines of aggregation
    # 40 lines of formatting
    ...

# After - split into phases
def compute_tag_statistics(db: Database, library_id: int) -> TagStats:
    rows = _query_tag_data(db, library_id)
    stats = _aggregate_tag_stats(rows)
    formatted = _format_tag_stats(stats)
    return TagStats(formatted)

def _query_tag_data(db: Database, library_id: int) -> list[dict]: ...
def _aggregate_tag_stats(rows: list[dict]) -> dict[str, int]: ...
def _format_tag_stats(stats: dict[str, int]) -> dict[str, Any]: ...

Example: Worker Crash Handling

def should_restart_worker(restart_count: int, last_restart_ms: int) -> RestartDecision:
    """
    Decide whether to restart a crashed worker or mark it as failed.
    
    Uses two-tier limits:
    - Rapid: 5 restarts in 5 minutes (catches OOM loops)
    - Lifetime: 20 total restarts (catches slow thrashing)
    """
    current_time_ms = now_ms()
    
    if restart_count >= 5 and (current_time_ms - last_restart_ms) < 300_000:
        return RestartDecision(action="mark_failed", reason="Rapid restart limit exceeded")
    
    if restart_count >= 20:
        return RestartDecision(action="mark_failed", reason="Lifetime restart limit exceeded")
    
    backoff = calculate_backoff(restart_count)
    return RestartDecision(action="restart", backoff_seconds=backoff)

Validation Checklist

Before committing component code, verify:

• Does this file import from services, workflows, or interfaces? → Violation
• Does this file import Essentia directly (and isn't

  ml_backend_essentia_comp.py

  )? → Violation
• Does this accept raw path strings where

  LibraryPath

  should be used? → Violation
• Is this doing orchestration instead of computation? → Should be a workflow
• Are heavy functions split into

  _private

  helpers? → Recommended
• Does the module name end in

  _comp.py

  ? → Convention

Layer Scripts

This skill includes validation scripts in

.github/skills/layer-components/scripts/

lint.py

Runs all linters on the components layer:

python .github/skills/layer-components/scripts/lint.py

Executes: ruff, mypy, vulture, bandit, radon, lint-imports

check_naming.py

Validates components naming conventions:

python .github/skills/layer-components/scripts/check_naming.py

Checks:

• Files must end in

  _comp.py

• Essentia imports only allowed in

  ml_backend_essentia_comp.py