AXN10 Woes

I bought a Cambridge Audio AX system for my office because it was cheap and it let me play records too. It’s really not too bad but I have my suspicions over the AXN10 Network Streamer. I’ve noticed it has a tendency to stop playing after a while. It plays fine for a good four to six months and then it’s like it runs out of buffer space or something and can only play music in five second chunks followed by a five second pauses. It’s peculiar. It’s also done it three times now so it’s now something that needs investigating.

My current setup uses my music stores on a WD MyCloud Mirror (that also needs fixing) that is streamed out via AssetUPnP running on an Ubuntu 22.04.5 LTS box to my CXNv2 and my AXN10. I originally used Twonky but Twonky is now, effectively, a dead platform. It doesn’t and never will support the Disk Number fields so multi-disk albums are a problem with Twonky.

I was hoping that I could use AssetUPnP to log the music and the duration played but I’m not convinced it’s logging is that capable. It seemed to log connections from the TV but not from the CA AXN which is a bit of a problem when I’m interested in what the AXN is doing.

The AXN itself doesn’t seem to do any logging either but, obviously, the Stream Magic app connects over APIs to the device so I was wondering whether there was a hidden interface there that I could poke. I’m really not the best when it comes to reverse engineering interfaces so I did something really dumb and just put in the IP address of the AXN into a browser followed by /debug. Surprisingly, this returned a “Remote debug not enabled” message which immediately made me think that there must be a way to enable remote debug. Time to go digging…

Ecosia wasn’t really of much help here but Google, especially Googles AI bot was most helpful. Initially it pointed me to a very nice project on GitHub (https://github.com/noahhusby/aiostreammagic)

Now, I’m no python developer (I’m no developer at all, really) but by explaining what I was trying to achieve Google came up with the following program:

import asyncio
import csv
import os
import sys
from datetime import datetime
from aiostreammagic import StreamMagicClient

# CONFIGURATION
AXN10_IP = "IP GOES HERE"  # Replace with your exact AXN10 IP
LOG_FILE = "/usr/local/bin/axn10-tracker/streamer_playback_log.csv"

class PlaybackTracker:
    def __init__(self):
        self.current_track = None
        self.current_artist = None
        self.play_start_time = None
        self.last_state = None

    def log_to_csv(self, start_time, stop_time, artist, track, duration_seconds):
        """Appends a completed playback record to the CSV file."""
        file_exists = os.path.isfile(LOG_FILE)
        
        try:
            with open(LOG_FILE, mode='a', newline='', encoding='utf-8') as f:
                writer = csv.writer(f)
                if not file_exists:
                    writer.writerow(["Start Time", "Stop Time", "Artist", "Track", "Played Duration (Sec)", "Played Duration (Min)"])
                
                duration_minutes = round(duration_seconds / 60, 2)
                writer.writerow([
                    start_time.strftime("%Y-%m-%d %H:%M:%S"),
                    stop_time.strftime("%Y-%m-%d %H:%M:%S"),
                    artist or "Unknown Artist",
                    track or "Unknown Track",
                    round(duration_seconds, 1),
                    duration_minutes
                ])
            print(f"Logged to CSV: '{track}' by {artist} - {duration_minutes} mins.", flush=True)
        except Exception as e:
            print(f"Failed writing to CSV: {e}", file=sys.stderr, flush=True)

    async def on_state_change(self, client: StreamMagicClient, *args):
        """Official callback engine triggered immediately on AXN10 updates."""
        # Grab the core play state object safely
        ps_obj = client.play_state
        if ps_obj:
            # Extract simple state string (e.g., 'play', 'pause', 'stop', 'idle')
            play_state = getattr(ps_obj, 'state', 'stopped').lower()
            
            # Extract nested metadata objects
            metadata = getattr(ps_obj, 'metadata', None)
            track_title = getattr(metadata, 'title', None) if metadata else None
            artist_name = getattr(metadata, 'artist', None) if metadata else None
        else:
            play_state = "stopped"
            track_title = None
            artist_name = None

        # Clean diagnostic line showing exactly what song has been processed
        print(f"[TRACKER STATUS] State: {play_state.upper()} | Title: '{track_title}' by {artist_name}", flush=True)

        now = datetime.now()

        # CASE 1: Streamer just entered a play cycle
        if play_state == "play" and self.last_state != "play":
            self.play_start_time = now
            self.current_track = track_title
            self.current_artist = artist_name
            self.last_state = "play"
            print(f"Playback started: '{track_title}' by {artist_name}", flush=True)

        # CASE 2: Tracks shifted during active streaming
        elif play_state == "play" and self.current_track != track_title:
            if self.play_start_time:
                duration = (now - self.play_start_time).total_seconds()
                if duration > 2:
                    self.log_to_csv(self.play_start_time, now, self.current_artist, self.current_track, duration)
            
            self.play_start_time = now
            self.current_track = track_title
            self.current_artist = artist_name
            print(f"Track changed: '{track_title}' by {artist_name}", flush=True)

        # CASE 3: Playback stopped, paused, or returned to idle
        elif play_state in ["pause", "stop", "idle"] and self.last_state == "play":
            if self.play_start_time:
                duration = (now - self.play_start_time).total_seconds()
                if duration > 2:
                    self.log_to_csv(self.play_start_time, now, self.current_artist, self.current_track, duration)
            self.play_start_time = None
            self.current_track = None
            self.last_state = play_state
            print(f"Playback stopped ({play_state}).", flush=True)

async def main():
    tracker = PlaybackTracker()
    print(f"Initializing event pipeline for AXN10 at {AXN10_IP}...", flush=True)
    
    while True:
        try:
            client = StreamMagicClient(AXN10_IP)
            
            # 1. Bind our handler logic block
            await client.register_state_update_callbacks(tracker.on_state_change)
            
            # 2. ACTUALLY EXECUTE THE NETWORK HANDSHAKE CONNECTION
            await client.connect()
            
            print("Connected! Active socket listening for streaming metrics...", flush=True)
            
            # Keep the context block alive indefinitely
            while True:
                await asyncio.sleep(3600)
                
        except Exception as e:
            print(f"Network drop encountered: {e}. Reconnecting in 15 seconds...", file=sys.stderr, flush=True)
            await asyncio.sleep(15)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nShutdown signal captured.")

This logs every event to a CSV and lets me keep track of everything that has been played and for how long. It also came up with a companion program to summarise the data for me. I probably could have done that bit myself but I was more interested in having something work that hacking around trying to make things work.

import csv
from datetime import datetime
from collections import defaultdict

LOG_FILE = "/usr/local/bin/axn10-tracker/streamer_playback_log.csv"

def analyze_logs():
    total_seconds = 0
    track_counts = defaultdict(int)
    track_durations = defaultdict(float)
    
    try:
        with open(LOG_FILE, mode='r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            
            for row in reader:
                # Calculate overall cumulative time
                seconds = float(row["Played Duration (Sec)"])
                total_seconds += seconds
                
                # Group data by track identifier
                track_identifier = f"'{row['Track']}' by {row['Artist']}"
                track_counts[track_identifier] += 1
                track_durations[track_identifier] += seconds
                
    except FileNotFoundError:
        print(f"Error: Could not find '{LOG_FILE}'. Make sure the logger script has run and recorded at least one song.")
        return

    # Convert total seconds into hours and minutes
    total_minutes = total_seconds / 60
    total_hours = total_minutes / 60
    
    print("=" * 50)
    print("      CAMBRIDGE AUDIO AXN10 PLAYBACK REPORT      ")
    print("=" * 50)
    print(f"Total Listening Time: {total_hours:.2f} Hours ({int(total_minutes)} Minutes)")
    print("-" * 50)
    
    # Sort and find the tracks listened to for the longest total duration
    print("Top 5 Most Listened-To Tracks (By Actual Duration):")
    sorted_by_duration = sorted(track_durations.items(), key=lambda x: x[1], reverse=True)
    
    for i, (track, seconds) in enumerate(sorted_by_duration[:5], 1):
        mins = seconds / 60
        print(f" {i}. {track} — Listened for {mins:.1f} mins total")

if __name__ == "__main__":
    analyze_logs()

The next thing we worked on was making it into a service so that it would always run. The last thing I needed was to forget to log a week or month’s worth of data and complete stuff my numbers up. Adding it as a service was pretty trivial.

First create the logger service:

sudo vi /etc/systemd/system/axn10-logger.service

   [Unit]
   Description=Cambridge Audio AXN10 Playback Logger
   After=network-online.target
   Wants=network-online.target

   [Service]
   Type=simple
   User=YOUR_UBUNTU_USERNAME
   WorkingDirectory=/usr/local/bin/axn10-tracker
   ExecStart=/usr/local/bin/axn10-tracker/venv/bin/python3 /usr/local/bin/axn10-tracker/axn10_logger.py
   Restart=always
   RestartSec=10
   StandardOutput=journal
   StandardError=journal

   [Install]
   WantedBy=multi-user.target

then enable and run the service:

sudo systemctl daemon-reload
sudo systemctl enable axn10-logger.service
sudo systemctl start axn10-logger.service

It actually took a few iterations to get the logging code to properly run as a service and to log the data correctly but Google helped me achieve something I could not have done on my own.

So now, every time I play something on my AXN it gets logged to a file and over time it will show me, not only the total play time between failures, it will also show me what artists / albums / songs I’m listening to the most. That’s pretty cool.

Pip Woes

I was trying to hack some python together to read an Excel spreadsheet for work today and I just could not get openpyxl installed, pip kept throwing up errors:

% pip install openpyxl
WARNING: Retrying (Retry(total=4, connect=None, read=None, redirect=None, status=None)) after connection broken by 'SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)'))': /simple/openpyxl/
WARNING: Retrying (Retry(total=3, connect=None, read=None, redirect=None, status=None)) after connection broken by 'SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)'))': /simple/openpyxl/
WARNING: Retrying (Retry(total=2, connect=None, read=None, redirect=None, status=None)) after connection broken by 'SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)'))': /simple/openpyxl/
WARNING: Retrying (Retry(total=1, connect=None, read=None, redirect=None, status=None)) after connection broken by 'SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)'))': /simple/openpyxl/
WARNING: Retrying (Retry(total=0, connect=None, read=None, redirect=None, status=None)) after connection broken by 'SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)'))': /simple/openpyxl/
Could not fetch URL https://pypi.org/simple/openpyxl/: There was a problem confirming the ssl certificate: HTTPSConnectionPool(host='pypi.org', port=443): Max retries exceeded with url: /simple/openpyxl/ (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)'))) - skipping
ERROR: Could not find a version that satisfies the requirement openpyxl (from versions: none)
ERROR: No matching distribution found for openpyxl
Could not fetch URL https://pypi.org/simple/pip/: There was a problem confirming the ssl certificate: HTTPSConnectionPool(host='pypi.org', port=443): Max retries exceeded with url: /simple/pip/ (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)'))) - skipping

The way I managed to get around it was based on some information I found on Stack Exchange:

% pip install --trusted-host files.pythonhosted.org --trusted-host pypi.org --upgrade pip

Now pip works just fine and everything is hunky-dory.