StateMap does not have data regarding current position of track?

[PulzWave]

PyStageLinQ is part of my setup for a good while now and I cannot record any video or livestream without it. So first of all, THANK YOU :)

I am using it for:

• Extracting current track information
• With some self made logic, detect a transition which triggers a specific OBS overlay
• Current BPM in an overlay
• Showing fader positions in real-time in an OBS overlay

Now I have a new idea, but I'd require to know the current position of a track. Eg. a track is 5:10 long, I can get that data from EngineDeck[1/2/3/4]TrackTrackLength. However right now, I dont think it's possible to extract the current "position". Eg. when I just started playing, I want this value to be 00:00. And about halfway, this should say 2:35 etc.

Am I missing something or isn't this possible right now?

The Resolume/StageLinq integration tells me it should be possible.

Any hints would be appreciated!

[Jaxc]

Hello

Sorry for my late answer. Its super fun to hear someone using the library!

From my knowledge this is not possible using the current implementation. It should be possible using the Beatinfo interface mentioned in #27, but I have not managed to implement this properly so far.

[Jaxc]

I finally got some time to look into this over the winter vacation and pushed some inital code to https://github.com/Jaxc/PyStageLinQ/tree/BeatInfo_implementation. It seems to work on my Prime Go, but I haven't tested it on any other device.

If you want to try to use it you can give it a shot:
It works will callbacks as before and outputs the class BeatInfoData as a variable to the callback. The current time of the track is given in beats, so you will have to do some math to get the position as time in the song

[PulzWave]

Thanks, this is awesome. I will definitely play with this.

If I can figure how out to know how many beats a track has, and using the total track time I should be able to calculate the current position.

[Jaxc]

Thanks, this is awesome. I will definitely play with this.

If I can figure how out to know how many beats a track has, and using the total track time I should be able to calculate the current position.

The data class in the return has the current beat, total beats for the track on deck and current BPM available, so I think everything you need should be there. Otherwise just say and I can check it :)

[PulzWave]

Thanks! I finally got some time to test your test branch.

However using that branch it fails to connect with my Denon DJ Prime 4+. When I revert to the main branch, it works immediately. Any ideas?

Minimal test script

#!/usr/bin/env python3
"""
Minimal test script for PyStageLinQ BeatInfo_implementation branch.
Use this to test if the branch works correctly.

Install branch version:
  pip install git+https://github.com/Jaxc/PyStageLinQ.git@BeatInfo_implementation

Install stable version:
  pip install PyStageLinQ
"""
import asyncio
import logging
from PyStageLinQ.PyStageLinQ import PyStageLinQ
from PyStageLinQ import EngineServices

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
log = logging.getLogger("TestBeatInfo")

# Global state
linq = None
channels = {1: {}, 2: {}, 3: {}, 4: {}}


def state_map_callback(states):
    """Callback for StateMap data."""
    for st in states:
        name = st.ParameterName
        raw = st.ParameterValue
        v = raw.get("string", raw.get("value", raw.get("state", None)))
        parts = name.split('/')
        try:
            if len(parts) >= 4 and parts[1] == "Engine" and parts[2].startswith("Deck"):
                deck_id = int(parts[2][-1])
                param = parts[3]
                if param == "CurrentBPM":
                    channels[deck_id]["bpm"] = round(v, 2)
                elif param == "Track" and len(parts) > 4:
                    if parts[4] == "ArtistName":
                        channels[deck_id]["artist"] = v
                    elif parts[4] == "SongName":
                        channels[deck_id]["title"] = v
                elif param == "PlayState":
                    channels[deck_id]["state"] = "playing" if v else "paused"
        except (ValueError, IndexError):
            pass


def beatinfo_callback(data):
    """Callback for BeatInfo data."""
    log.info(f"BeatInfo received!")
    if data.deck1:
        log.info(f"  Deck 1: Beat {data.deck1.currentBeat:.2f} / {data.deck1.trackTotalBeats:.2f} @ {data.deck1.currentBPM:.2f} BPM")
    if data.deck2:
        log.info(f"  Deck 2: Beat {data.deck2.currentBeat:.2f} / {data.deck2.trackTotalBeats:.2f} @ {data.deck2.currentBPM:.2f} BPM")


def new_device_callback(ip, discovery, services):
    """Callback when a StageLinQ device is discovered."""
    log.info(f"DISCOVERED: {discovery.device_name} at {ip}")
    log.info(f"Available services: {[s.service for s in services]}")
    
    if linq:
        for svc in services:
            try:
                if svc.service == "StateMap":
                    log.info("Subscribing to StateMap...")
                    linq.subscribe_to_statemap(svc, EngineServices.common_functions, state_map_callback)
                    log.info("StateMap subscription OK")
                elif svc.service == "BeatInfo":
                    log.info("Subscribing to BeatInfo...")
                    # This is the new method from BeatInfo_implementation branch
                    linq.subscribe_to_beatinfo(svc, beatinfo_callback)
                    log.info("BeatInfo subscription OK")
            except Exception as e:
                log.error(f"ERROR subscribing to {svc.service}: {e}")


async def print_status():
    """Print deck status every second."""
    while True:
        await asyncio.sleep(1)
        print("\n--- Deck Status ---")
        for deck_id, data in sorted(channels.items()):
            if data:
                artist = data.get("artist", "?")
                title = data.get("title", "?")
                bpm = data.get("bpm", 0)
                state = data.get("state", "?")
                print(f"  Deck {deck_id}: {artist} - {title} @ {bpm} BPM [{state}]")
            else:
                print(f"  Deck {deck_id}: (empty)")


async def main():
    global linq
    log.info("Starting PyStageLinQ test...")
    log.info("Waiting for StageLinQ device...")
    
    linq = PyStageLinQ(new_device_callback, name="BeatInfo-Test")
    asyncio.create_task(linq._start_stagelinq(standalone=False))
    asyncio.create_task(print_status())
    
    # Run forever
    await asyncio.Event().wait()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        log.info("Shutting down...")

Main branch output

2026-01-23 15:53:33,942 INFO: Starting PyStageLinQ test...
2026-01-23 15:53:33,942 INFO: Waiting for StageLinQ device...
2026-01-23 15:53:33,949 INFO: Looking for discovery frames on 3 IP local IP addresses:
2026-01-23 15:53:33,949 INFO: 10.0.52.27
2026-01-23 15:53:33,949 INFO: 192.168.132.1
2026-01-23 15:53:33,949 INFO: 192.168.75.1
2026-01-23 15:53:34,376 INFO: DISCOVERED: prime4plus at 10.0.52.37
2026-01-23 15:53:34,376 INFO: Available services: ['StateMap', 'Broadcast', 'TimeSynchronization', 'BeatInfo', 'FileTransfer']
2026-01-23 15:53:34,376 INFO: Subscribing to StateMap...
2026-01-23 15:53:34,376 INFO: StateMap subscription OK
2026-01-23 15:53:34,376 INFO: Subscribing to BeatInfo...
2026-01-23 15:53:34,376 ERROR: ERROR subscribing to BeatInfo: 'PyStageLinQ' object has no attribute 'subscribe_to_beatinfo'
2026-01-23 15:53:34,576 INFO: DISCOVERED: prime4plus at 10.0.52.37
2026-01-23 15:53:34,576 INFO: Available services: ['FileTransfer']
2026-01-23 15:53:34,793 INFO: DISCOVERED: prime4plus at 10.0.52.37
2026-01-23 15:53:34,793 INFO: Available services: ['FileTransfer']

--- Deck Status ---
  Deck 1: Anunnakis - Orbit (Original Mix) @ 127.0 BPM [paused]
  Deck 2: Landau, Chellow - We Dip (Extended Mix) @ 127.0 BPM [playing]
  Deck 3:  -  @ 120 BPM [paused]
  Deck 4:  -  @ 120 BPM [paused]

--- Deck Status ---
  Deck 1: Anunnakis - Orbit (Original Mix) @ 127.0 BPM [paused]
  Deck 2: Landau, Chellow - We Dip (Extended Mix) @ 127.0 BPM [playing]
  Deck 3:  -  @ 120 BPM [paused]
  Deck 4:  -  @ 120 BPM [paused]

--- Deck Status ---
  Deck 1: Anunnakis - Orbit (Original Mix) @ 127.0 BPM [paused]
  Deck 2: Landau, Chellow - We Dip (Extended Mix) @ 127.0 BPM [playing]
  Deck 3:  -  @ 120 BPM [paused]
  Deck 4:  -  @ 120 BPM [paused]

Test branche output

$ python test_beatinfo_branch.py
2026-01-23 15:54:22,345 INFO: Starting PyStageLinQ test...
2026-01-23 15:54:22,345 INFO: Waiting for StageLinQ device...

--- Deck Status ---
  Deck 1: (empty)
  Deck 2: (empty)
  Deck 3: (empty)
  Deck 4: (empty)

--- Deck Status ---
  Deck 1: (empty)
  Deck 2: (empty)
  Deck 3: (empty)
  Deck 4: (empty)

--- Deck Status ---
  Deck 1: (empty)
  Deck 2: (empty)
  Deck 3: (empty)
  Deck 4: (empty)

[Jaxc]

Hello @PulzWave !

Thanks for testing it out. I tried the your script and it works on my computer, so I'm thinking maybe there is some issue that you are using the wrong version of the code somehow?

How did you get the custom branch into your code more specifically?

[PulzWave]

Heya, thanks for reaching out once more :)

What I did was pip uninstall PyStageLinq, followed by "pip install git+https://github.com/Jaxc/PyStageLinQ.git@BeatInfo_implementation". I could see in the logging that it's compiling that version.

if you have different instructions for me let me know!

[Jaxc]

hmm I have never tried to do this myself so I have no personal experience. :)

Could you verify that your local files are the same as the one on
GitHub? Specifically row 313 of PyStateLinQ.py:
https://github.com/Jaxc/PyStageLinQ/blob/BeatInfo_implementation/PyStageLinQ/PyStageLinQ.py#L313

I tried the way you installed it and it seems to download the correct version. Could it be that you installed it with a venv activated and then try to run it outside the venv, or the reverse?

[PulzWave]

Yep the package was successfully installed and it's using the correct .venv.

I dived a bit deeper and I saw that the root cause was that a few other things changed in the test branch. I might have more network interfaces which is a difference between your setup and mine, or that we run a different OS (Mac vs. Windows?)

I have created a PR: #61

So right now I see beat info in the test branch + my fix!

[PulzWave]

Small demo:

Aaaaand one hour later, it's integrated into my OBS Overlay 😍

[Jaxc]

Thanks so much for your effort @PulzWave !

I don't have time to properly check the PR right now (I want to understand what I did wrong) But I'll try to do it next week.

I realize now that I managed to read your main branch log as the log with the error, so very sorry for my odd comments above

[Jaxc]

Oh and your stuff looks super cool and polished!

[PulzWave]

Hey no rush at all, I'm going into production with your test branch + my fix upcoming Wednesday. Feel free to join the livestream on YouTube and see YOUR magic in action!

And thank you so much, I am having so much fun with it!