Introduce score power

app.py

@@ -20,7 +20,7 @@ logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 # Load models and data
-print("Loading models and data...")
+logging.info("Loading models and data...")
 all_labels = pd.read_csv(cfg.LABELS_LOCATION)
 scalers = joblib.load(cfg.SCALER_DICT_LOCATION)
 index = faiss.read_index(cfg.INDEX_LOCATION)
@@ -28,8 +28,8 @@ ds = load_dataset(cfg.LOOKUP_DS_NAME)
 lookup = ds['train'].to_pandas().set_index('track_id')[['title', 'artist']]
 
 # Initialize analysis
-ea = EmbeddingsAnalysis(index, all_labels, lookup, scalers, cfg.RADII, close_threshold=cfg.CLOSE_THRESHOLD)
-print("Models loaded successfully!")
+ea = EmbeddingsAnalysis(index, all_labels, lookup, scalers, cfg.RADII, close_threshold=cfg.CLOSE_THRESHOLD, score_power=cfg.SCORE_POWER)
+logging.info("Models loaded successfully!")
 
 # Load how it works content
 with open(cfg.HOW_IT_WORKS_MD_LOCATION, 'r') as f:
@@ -130,8 +130,8 @@ def analyze_chord_sequence_text(chord_text: str) -> tuple[Optional[float], Optio
         embeddings = get_embeddings_from_chord_sequences([chords])
         neighbour_embeddings = None
         if len(chords) < cfg.MIN_SEQUENCE_LENGTH_FOR_NEIGHBOURS:
-            padded_chords = _pad_sequence_by_repetition(chords, cfg.MIN_SEQUENCE_LENGTH_FOR_NEIGHBOURS)
-            neighbour_embeddings = get_embeddings_from_chord_sequences([padded_chords])
+            chords = _pad_sequence_by_repetition(chords, cfg.MIN_SEQUENCE_LENGTH_FOR_NEIGHBOURS)
+            neighbour_embeddings = get_embeddings_from_chord_sequences([chords])
         score, neighbours = _perform_analysis(embeddings, [len(chords)], neighbour_embeddings)
         return score, neighbours
     except AppError as e:

cfg.py

@@ -6,6 +6,7 @@ SCALER_DICT_LOCATION = './assets/quantile_transformers.joblib'
 MIN_SEQUENCE_LENGTH_FOR_NEIGHBOURS = 18
 HOW_IT_WORKS_MD_LOCATION = './how_it_works.md'
 HOW_IT_WORKS_SVG_LOCATION = './assets/harmonic_analysis_simple.svg'
-RADII = (0.8, 0.85, 0.9, 0.925, 0.95)
-# RADII = (0.7, 0.925, 0.95, 0.99, 0.995)
-# RADII = (0.7, 0.85, 0.9, 0.99, 0.995)
+# RADII = (0.8, 0.85, 0.9, 0.925, 0.95)
+RADII = (0.85, 0.9) + tuple(range(1, 50)) + (0.925, 0.95, 0.975)
+SCORE_POWER = 0.5
+

src/analysis.py

@@ -17,12 +17,12 @@ class EmbeddingsAnalysis:
     :param scalers: Dictionary mapping length ranges to quantile transformers for score normalization.
     :param close_threshold: Similarity threshold for neighbor search.
     """
-    def __init__(self, index, all_labels, lookup, scalers, radii, close_threshold=0.95):
+    def __init__(self, index, all_labels, lookup, scalers, radii, close_threshold=0.95, score_power=1.0):
         all_labels_np = all_labels['track_id'].to_numpy()
         all_lengths_np = all_labels['length'].to_numpy()
         self._ecn = EmbeddingClosestNeighbours(index, all_labels_np, all_lengths_np, lookup, close_threshold=close_threshold)
         specific_scalers = {i: scaler for (l, r), scaler in scalers.items() for i in range(l, r)}
-        sm = SimpleMethodology(specific_scalers, specific_scalers[_FALLBACK_INDEX])
+        sm = SimpleMethodology(specific_scalers, specific_scalers[_FALLBACK_INDEX], score_power=score_power)
         self._scorer = EmbeddingsOriginalityScorer(index, all_labels_np, radii, sm)
 
 

src/convert.py

@@ -1,3 +1,4 @@
+import logging
 import numpy as np
 from gradio_client import Client
 import os
@@ -5,6 +6,8 @@ import json
 import time
 import httpx
 
+logger = logging.getLogger(__name__)
+
 from chord_extractor.extractors import Chordino
 from chord_extractor import clear_conversion_cache, LabelledChordSequence
 
@@ -28,7 +31,7 @@ def _create_client():
 _client = _create_client()
 
 def _call_embedding_service(chords_w_timestamps):
-    print(chords_w_timestamps)
+    logger.info(chords_w_timestamps)
     result = _client.predict(json.dumps(chords_w_timestamps), api_name="/predict")
     return json.loads(result)
 

src/methodology.py

@@ -18,15 +18,16 @@ class CountBasedMethodology(ABC):
 
 
 class SimpleMethodology(CountBasedMethodology):
-    def __init__(self, scalers: dict[int, _TransformerProtocol], fallback_scaler: _TransformerProtocol):
+    def __init__(self, scalers: dict[int, _TransformerProtocol], fallback_scaler: _TransformerProtocol, score_power: float = 1.0):
         self._scalers = scalers
         self._fallback_scaler = fallback_scaler
+        self._score_power = score_power
 
     def execute(self, neighbours_df: pd.DataFrame, lengths: pd.Series) -> pd.Series:
-        unscaled = sum((neighbours_df[col] - 1) * (i + 1) for i, col in enumerate(neighbours_df.columns))
+        unscaled = sum(neighbours_df[col] * (i + 1) for i, col in enumerate(neighbours_df.columns))
         concat = pd.concat([unscaled.rename('unscaled'), lengths.rename('length')], axis=1)
         scaled = concat.apply(
             lambda row: self._scalers.get(row['length'], self._fallback_scaler).transform(pd.DataFrame({_SCALER_X_LABEL: row['unscaled']}, index=[0]))[0][0],
             axis=1
         )
-        return 1 - scaled
+        return 1 - scaled ** self._score_power