Pad short chord sequences

app.py

@@ -18,8 +18,10 @@ logger = logging.getLogger(__name__)
 INDEX_LOCATION = './assets/chords_20251021.index'
 LABELS_LOCATION = './assets/all_labels.csv'
 LOOKUP_DS_NAME = 'ohollo/lmd_chords'
-CLOSE_THRESHOLD = 0.99
+CLOSE_THRESHOLD = 0.9
 SCALER_DICT_LOCATION = './assets/quantile_transformers.joblib'
+MIN_SEQUENCE_LENGTH_FOR_NEIGHBOURS = 24
+HOW_IT_WORKS_MD_LOCATION = './how_it_works.md'
 
 # Load models and data
 print("Loading models and data...")
@@ -33,18 +35,31 @@ lookup = ds['train'].to_pandas().set_index('track_id')[['title', 'artist']]
 ea = EmbeddingsAnalysis(index, all_labels, lookup, scalers, close_threshold=CLOSE_THRESHOLD)
 print("Models loaded successfully!")
 
+# Load how it works content
+with open(HOW_IT_WORKS_MD_LOCATION, 'r') as f:
+    how_it_works_content = f.read()
+
 def _parse_chord_input(chord_text):
     if not chord_text.strip():
         return []
-    
+
     # Try comma separation first, then space separation
     if ',' in chord_text:
         chords = [chord.strip() for chord in chord_text.split(',') if chord.strip()]
     else:
         chords = chord_text.split()
-    
+
     return chords
 
+
+def _pad_sequence_by_repetition(sequence, min_length):
+    if len(sequence) >= min_length:
+        return sequence
+    result = sequence.copy()
+    while len(result) < min_length:
+        result.extend(sequence)
+    return result
+
 def _neighbours_to_dict(neighbours_list):
     result = []
     for neighbours_group in neighbours_list:
@@ -59,9 +74,9 @@ def _neighbours_to_dict(neighbours_list):
         result.append(group_result)
     return result
 
-def _perform_analysis(embeddings, sequence_lengths):
+def _perform_analysis(embeddings, sequence_lengths, neighbour_embeddings=None):
     scores = ea.get_scores(embeddings, sequence_lengths)
-    neighbours = ea.get_neighbours(embeddings, limit=10)
+    neighbours = ea.get_neighbours(neighbour_embeddings if neighbour_embeddings is not None else embeddings, limit=10)
     score = scores[0]
     # Convert neighbours to list of dicts for the first sequence
     neighbours_dict = []
@@ -81,11 +96,11 @@ def analyze_chord_sequence_text(chord_text: str) -> tuple[float, list[dict]]:
     Analyze a chord sequence from text input. Analysis is in the form of
     an originality score and a list of similar songs from a non-exhaustive
     sample set of songs in the system data store.
-    
+
     Args:
-        chord_text: Chord sequence input as text (comma-separated or space-separated). IMPORTANT: Unless length is explicitly specified, for accurate results, provide the complete sequence of chords that would feature in a typical song, e.g. "C, Am, F, G, C, Am, F, G, C, Am, F, G, C, Am, F, G, ..." NOT just "C, Am, F, G". This could be 20-30 chords for a three minute song. 
+        chord_text: Chord sequence input as text (comma-separated or space-separated). IMPORTANT: Unless length is explicitly specified, for accurate results, provide the complete sequence of chords that would feature in a typical song, e.g. "C, Am, F, G, C, Am, F, G, C, Am, F, G, C, Am, F, G, ..." NOT just "C, Am, F, G". This could be 20-30 chords for a three minute song.
     Returns:
-        tuple[float, list[dict]]: Originality score and list of dictionaries, each representing a similar song. You may infer that some items in the list are essentially the same song - if so don't repeat them to the user. Also some songs are more famous than others in the results - if you come across a famous one, then highlight it.  
+        tuple[float, list[dict]]: Originality score and list of dictionaries, each representing a similar song. You may infer that some items in the list are essentially the same song - if so don't repeat them to the user. Also some songs are more famous than others in the results - if you come across a famous one, then highlight it.
     """
     logging.info(f"Analyzing chord sequence: {chord_text}")
     try:
@@ -93,7 +108,11 @@ def analyze_chord_sequence_text(chord_text: str) -> tuple[float, list[dict]]:
         if not chords:
             return None, None
         embeddings = get_embeddings_from_chord_sequences([chords])
-        score, neighbours = _perform_analysis(embeddings, [len(chords)])
+        neighbour_embeddings = None
+        if len(chords) < MIN_SEQUENCE_LENGTH_FOR_NEIGHBOURS:
+            padded_chords = _pad_sequence_by_repetition(chords, MIN_SEQUENCE_LENGTH_FOR_NEIGHBOURS)
+            neighbour_embeddings = get_embeddings_from_chord_sequences([padded_chords])
+        score, neighbours = _perform_analysis(embeddings, [len(chords)], neighbour_embeddings)
         return score, neighbours
     except Exception as e:
         logger.error(f"Error analyzing chord sequence: {e}")
@@ -178,7 +197,10 @@ with gr.Blocks(title="Harmonic Analysis Tool", theme=gr.themes.Soft()) as app:
                     ["Am, F, C, G"],
                     ["D, A, Bm, G"],
                     ["Em, C, G, D"],
-                    ["F, C, Dm, Bb"]
+                    ["F, C, Dm, Bb"],
+                    ["A7, D7, A7, E7, D7, A7"],
+                    ["Am, F, C, G, Am, F, C, G, C, G, Am, F, C, G, Am, F, "
+                     "Am, F, C, G, Am, F, C, G, C, G, Am, F, C, G, Am, F"]
                 ],
                 inputs=[chord_input]
             )
@@ -200,7 +222,11 @@ with gr.Blocks(title="Harmonic Analysis Tool", theme=gr.themes.Soft()) as app:
                     file_display = gr.Markdown(label="File Info")
                     audio_scores_output = gr.Markdown(label="Analysis Results")
                     audio_neighbours_output = gr.Markdown(label="Similar Songs")
-    
+
+        # Tab 3: How It Works
+        with gr.TabItem("How It Works"):
+            gr.Markdown(how_it_works_content)
+
     # Event handlers
     analyze_btn.click(
         fn=_format_chord_analysis_for_ui,

src/analysis.py

@@ -8,6 +8,15 @@ from src.scorer import EmbeddingsOriginalityScorer
 
 
 class EmbeddingsAnalysis:
+    """
+    Facade for analyzing embeddings, combining neighbor search and originality scoring.
+
+    :param index: FAISS index for similarity search.
+    :param all_labels: DataFrame containing 'track_id' and 'length' columns for indexed entries.
+    :param lookup: Pandas DataFrame containing metadata for each indexed entry.
+    :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, close_threshold=0.95):
         all_labels_np = all_labels['track_id'].to_numpy()
         all_lengths_np = all_labels['length'].to_numpy()

src/convert.py

@@ -2,18 +2,33 @@ import numpy as np
 from gradio_client import Client
 import os
 import json
+import time
+import httpx
 
 from chord_extractor.extractors import Chordino
 from chord_extractor import clear_conversion_cache, LabelledChordSequence
 
-_CONSTANT_GAP_SECS = 2 
+_CONSTANT_GAP_SECS = 2
 _SEQ_EMBED_SPACE = 'ohollo/chord-seq-embed'
 _POST_PROCESS_CHORD_LEN_RATIO = 0.7
+_MAX_RETRIES = 3
+_RETRY_DELAY_SECS = 2
 
 
-_client = Client(_SEQ_EMBED_SPACE)
+def _create_client():
+    for attempt in range(_MAX_RETRIES):
+        try:
+            return Client(_SEQ_EMBED_SPACE)
+        except httpx.ReadTimeout:
+            if attempt < _MAX_RETRIES - 1:
+                time.sleep(_RETRY_DELAY_SECS)
+            else:
+                raise
+
+_client = _create_client()
 
 def _call_embedding_service(chords_w_timestamps):
+    print(chords_w_timestamps)
     result = _client.predict(json.dumps(chords_w_timestamps), api_name="/predict")
     return json.loads(result)
 

src/methodology.py

@@ -12,7 +12,7 @@ class _TransformerProtocol:
 
 class CountBasedMethodology(ABC):
     @abstractmethod
-    def execute(self, neighbours_df: pd.DataFrame) -> pd.Series:
+    def execute(self, neighbours_df: pd.DataFrame, lengths: pd.Series) -> pd.Series:
         ...
     
     @abstractmethod