light9 Changeset - cc69faa87c27

Changeset - cc69faa87c27

Parent rev.

Child rev.

[Not reviewed]

default

0 6 0

drewp@bigasterisk.com - 8 months ago 2024-05-25 22:44:11
drewp@bigasterisk.com

tear up and rewrite ascoltami to emit player state into the graph. web ui works but displays nothing but songs

6 files changed with 286 insertions and 210 deletions:

src/light9/ascoltami/main.py

src/light9/ascoltami/player.py

src/light9/ascoltami/playlist.py

src/light9/ascoltami/webapp.py

src/light9/showconfig.py

web/ascoltami/main.ts

0 comments (0 inline, 0 general)

src/light9/ascoltami/main.py

➞

Show inline comments

 #!bin/python
 import logging
 import optparse
 import sys
 from typing import cast
 import gi
 from rdflib import URIRef
 import louie
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import ConjunctiveGraph, Literal, URIRef
 from rdflib.graph import _ContextType
 from starlette.applications import Starlette
 from starlette.routing import Route
 from starlette_exporter import PrometheusMiddleware, handle_metrics
 from twisted.internet import reactor
 from twisted.internet.interfaces import IReactorCore
 from light9.run_local import log
 from light9 import networking, showconfig
 from light9.ascoltami import webapp
 from light9.ascoltami.player import Player
 from light9.ascoltami.import_gst import Gst
 from light9.ascoltami.player import Player, PlayerState
 from light9.ascoltami.playlist import NoSuchSong, Playlist
 reactor = cast(IReactorCore, reactor)
 from light9.namespaces import L9
 from light9.newtypes import decimalLiteral
 from light9.run_local import log
 class Ascoltami:
     def __init__(self, graph, show):
+    def __init__(self, graph: SyncedGraph, show: URIRef):
         self.graph = graph
-        self.player = Player(onEOS=self.onEOS, autoStopOffset=0)
+        self.player = Player(onStateChange=self.onStateChange, autoStopOffset=0)
         self.show = show
         self.playlist = Playlist.fromShow(graph, show)
         self.ctx = cast(_ContextType, (URIRef(self.show + '/ascoltami')))
         self.playerState = PlayerState()
         self.playlist = Playlist(graph, show)
     def onStateChange(self, s: PlayerState):
         log.info('louie send')
         louie.send(webapp.OnStateChange, s=s)
         g = self.stateAsGraph(s)
         self.graph.patchSubgraph(newGraph=g, context=self.ctx)
         self.playerState = s
     def stateAsGraph(self, s):
         g = ConjunctiveGraph()
         asc = L9['ascoltami']
         if s.song:
             try:
                 g.add((asc, L9['song'], self.playlist.songUri(s.song), self.ctx))
             except KeyError:
                 pass
         g.add((asc, L9['duration'], decimalLiteral(s.duration), self.ctx))
         g.add((asc, L9['playing'], Literal(s.playing), self.ctx))
         if s.wallStartTime is not None:
             g.add((asc, L9['wallStartTime'], decimalLiteral(s.wallStartTime), self.ctx))
         if s.pausedSongTime is not None:
             g.add((asc, L9['pausedSongTime'], decimalLiteral(s.pausedSongTime), self.ctx))
         g.add((asc, L9['endOfSong'], Literal(s.endOfSong), self.ctx))
         return g
     def getPlayerState(self) -> PlayerState:
         return self.playerState
     def onEOS(self, song):
         self.player.pause()
         self.player.seek(0)
         thisSongUri = webapp.songUri(self.graph, URIRef(song))
@@ @@ -47,35 +73,32 @@ class Ascoltami: @@
 def main():
     logging.getLogger('sse_starlette.sse').setLevel(logging.INFO)
     Gst.init(None)
-    graph = showconfig.getGraph()
+    graph = SyncedGraph(networking.rdfdb.url, "ascoltami")
     asco = Ascoltami(graph, showconfig.showUri())
     h = webapp.WebHandlers(graph, asco.show, asco.player, asco.playlist, asco.getPlayerState)
     app = Starlette(
         debug=True,
         routes=[
             Route("/config", webapp.get_config),
             Route("/time", webapp.get_time, methods=["GET"]),
             Route("/time", webapp.post_time, methods=["POST"]),
             Route("/time/stream", webapp.timeStream),
             Route("/song", webapp.post_song, methods=["POST"]),
             Route("/songs", webapp.get_songs),
             Route("/seekPlayOrPause", webapp.post_seekPlayOrPause),
             Route("/output", webapp.post_output, methods=["POST"]),
             Route("/go", webapp.post_goButton, methods=["POST"]),
             Route("/config", h.get_config),
             Route("/time", h.get_time, methods=["GET"]),
             Route("/time", h.post_time, methods=["POST"]),
             Route("/time/stream", h.timeStream),
             Route("/song", h.post_song, methods=["POST"]),
             Route("/songs", h.get_songs),
             Route("/seekPlayOrPause", h.post_seekPlayOrPause),
             Route("/output", h.post_output, methods=["POST"]),
             Route("/go", h.post_goButton, methods=["POST"]),
         ],
+    )
     app.add_middleware(PrometheusMiddleware)
     app.add_route("/metrics", handle_metrics)
     app.state.graph = graph
     app.state.show = asco.show
     app.state.player = asco.player
     return app
 app = main()

src/light9/ascoltami/player.py

➞

Show inline comments

 """
 alternate to the mpd music player, for ascoltami
 alternate to the mpd music player, for ascoltami.
 This module doesn't know URIRef or graph. It does use
 GstFileUri (str) for filesystem paths.
 """
 import asyncio
 import logging
 import time
 import traceback
 from dataclasses import dataclass
-from pathlib import Path
+from typing import NewType
 from rdflib import URIRef
 from twisted.internet import task
 from light9.ascoltami.import_gst import Gst  # type: ignore
 from light9.metrics import metrics
 log = logging.getLogger()
 GstFileUri = NewType('GstFileUri', str)
 @dataclass
 class PlayerState:
     song: GstFileUri | None = None
     duration: float = 0
     playing: bool = False  # time is advancing, and song time is now()-wallStartTime
     # wall time of song-time 0 (this is adjusted when you seek)
     wallStartTime: float = 0
     pausedSongTime: float | None = None  # if we're paused, this has the song time
     endOfSong: bool = False  # True if we're in the stopped state due to EOS
 class Player:
-    def __init__(self, autoStopOffset=4, onEOS=None):
+    def __init__(self, autoStopOffset=4, onStateChange=lambda ps: None):
         """autoStopOffset is the number of seconds before the end of
         song before automatically stopping (which is really pausing).
         onEOS is an optional function to be called when we reach the
         end of a stream (for example, can be used to advance the song).
         It is called with one argument which is the URI of the song that
         just finished."""
         """
         self.autoStopOffset = autoStopOffset
         self.playbin = self.pipeline = Gst.ElementFactory.make('playbin', None)
         self.onStateChange = onStateChange
         self._playbin = self.pipeline = Gst.ElementFactory.make('playbin', None)
         self._playStartTime = 0
         self.playerState: PlayerState = PlayerState()
         self._lastWatchTime = 0
         self._autoStopTime = 0
         self._lastSetSongUri = None
         self._onEOS = onEOS
         self._lastSetSongUri: GstFileUri | None = None
         task.LoopingCall(self.watchTime).start(.050)
         # task.LoopingCall(self._watchTime).start(.050)
         asyncio.create_task(self._watchTime())
         #bus = self.pipeline.get_bus()
         # not working- see notes in pollForMessages
-        #self.watchForMessages(bus)
+        #self._watchForMessages(bus)
     def watchTime(self):
     async def _watchTime(self):
         while True:
             now = time.time()
         try:
             self.pollForMessages()
                 self._pollForMessages()
             t = self.currentTime()
             log.debug("watch %s < %s < %s", self._lastWatchTime, self._autoStopTime, t)
+                # log.debug("watch %s < %s < %s", self._lastWatchTime, self._autoStopTime, t)
             if self._lastWatchTime < self._autoStopTime < t:
                 log.info("autostop")
                 self.pause()
                 ps = PlayerState(
                     song=self._getSongFileUri(),
                     duration=round(self.duration(), 2),
                     wallStartTime=round(now - t, 2) if self.isPlaying() else None,
                     playing=self.isPlaying(),
                     pausedSongTime=None if self.isPlaying() else t,
                     endOfSong=t >= self.duration() - .1,  #todo
+                )
                 if self.playerState != ps:
                     self.onStateChange(ps)
                     self.playerState = ps
             self._lastWatchTime = t
         except Exception:
             traceback.print_exc()
             await asyncio.sleep(0.5)
-    def watchForMessages(self, bus):
+    def _watchForMessages(self, bus):
         """this would be nicer than pollForMessages but it's not working for
         me. It's like add_signal_watch isn't running."""
         bus.add_signal_watch()
         def onEos(*args):
             print("onEos", args)
             if self._onEOS is not None:
-                self._onEOS(self.getSong())
+                self._onEOS(self._getSongFileUri())
         bus.connect('message::eos', onEos)
         def onStreamStatus(bus, message):
             print("streamstatus", bus, message)
             (statusType, _elem) = message.parse_stream_status()
             if statusType == Gst.StreamStatusType.ENTER:
                 self.setupAutostop()
         bus.connect('message::stream-status', onStreamStatus)
-    def pollForMessages(self):
+    def _pollForMessages(self):
         """bus.add_signal_watch seems to be having no effect, but this works"""
         bus = self.pipeline.get_bus()
         mt = Gst.MessageType
         msg = bus.poll(
             mt.EOS | mt.STREAM_STATUS | mt.ERROR,  # | mt.ANY,
 )
@@ @@ -87,45 +116,47 @@ class Player: @@
             # output has an error, since that's otherwise kind of
             # mysterious to diagnose. I don't think this is exactly
             # working.
             if msg.type == mt.ERROR:
                 log.error(repr(msg.parse_error()))
             if msg.type == mt.EOS:
                 if self._onEOS is not None:
                     self._onEOS(self.getSong())
                 log.info("EOS msg: todo")
             if msg.type == mt.STREAM_STATUS:
                 (statusType, _elem) = msg.parse_stream_status()
                 if statusType == Gst.StreamStatusType.ENTER:
                     self.setupAutostop()
     def seek(self, t):
-        isSeekable = self.playbin.seek_simple(Gst.Format.TIME, Gst.SeekFlags.FLUSH | Gst.SeekFlags.ACCURATE | Gst.SeekFlags.SKIP, t * Gst.SECOND)
+        isSeekable = self._playbin.seek_simple(Gst.Format.TIME, Gst.SeekFlags.FLUSH | Gst.SeekFlags.ACCURATE | Gst.SeekFlags.SKIP, t * Gst.SECOND)
         if not isSeekable:
             raise ValueError('seek_simple failed')
         self._playStartTime = time.time()
     def setSong(self, songLoc, play=True):
+    def setSong(self, songLoc: GstFileUri, play=True):
         """
         uri like file:///my/proj/light9/show/dance2010/music/07.wav
         """
         log.info("set song to %r" % songLoc)
         self.pipeline.set_state(Gst.State.READY)
-        self.preload(songLoc)
+        self._preload(songLoc)
         self.pipeline.set_property("uri", songLoc)
         self._lastSetSongUri = songLoc
         # todo: don't have any error report yet if the uri can't be read
         if play:
             self.pipeline.set_state(Gst.State.PLAYING)
             self._playStartTime = time.time()
     def getSong(self):
         """Returns the URI of the current song."""
     def _getSongFileUri(self) -> GstFileUri | None:
         """Returns the file URI of the current song."""
         # even the 'uri' that I just set isn't readable yet
-        return self.playbin.get_property("uri") or self._lastSetSongUri
+        return GstFileUri(self._playbin.get_property("uri")) or self._lastSetSongUri
     def preload(self, songPath):
     def getSong(self) -> GstFileUri | None:
         return self.playerState.song
     def _preload(self, songPath):
         """
         to avoid disk seek stutters, which happened sometimes (in 2007) with the
         non-gst version of this program, we read the whole file to get
         more OS caching.
         i don't care that it's blocking.
@@ @@ -137,27 +168,27 @@ class Player: @@
         except IOError as e:
             log.error("couldn't preload %s, %r", songPath, e)
             raise
     @metrics('current_time').time()
     def currentTime(self):
-        success, cur = self.playbin.query_position(Gst.Format.TIME)
+        success, cur = self._playbin.query_position(Gst.Format.TIME)
         if not success:
             return 0
         return cur / Gst.SECOND
     def duration(self):
-        success, dur = self.playbin.query_duration(Gst.Format.TIME)
+        success, dur = self._playbin.query_duration(Gst.Format.TIME)
         if not success:
             return 0
         return dur / Gst.SECOND
     def states(self):
         """json-friendly object describing the interesting states of
         the player nodes"""
-        success, state, pending = self.playbin.get_state(timeout=0)
+        success, state, pending = self._playbin.get_state(timeout=0)
         return {"current": {"name": state.value_nick}, "pending": {"name": state.value_nick}}
     def pause(self):
         self.pipeline.set_state(Gst.State.PAUSED)
     def isAutostopped(self):

src/light9/ascoltami/playlist.py

➞

Show inline comments

 import logging
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import RDFS, URIRef
 from light9.ascoltami.player import GstFileUri
 from light9.namespaces import L9
 from light9.showconfig import songOnDisk
 from light9.namespaces import L9
 log = logging.getLogger('list')
 class NoSuchSong(ValueError):
     """Raised when a song is requested that doesn't exist (e.g. one
     after the last song in the playlist)."""
 class Playlist:
     """
     maintains a view of the playlist in the rdf graph. Also produces GstFileUri strs
     for Player
     """
     _songs: list[URIRef]
     _labels: dict[URIRef, str]
     _nextSong: dict[URIRef, URIRef]
     _songUri: dict[GstFileUri, URIRef]
     _fileUri: dict[URIRef, GstFileUri]
-    def __init__(self, graph, playlistUri):
+    def __init__(self, graph: SyncedGraph, show: URIRef):
         self.graph = graph
         self.playlistUri = playlistUri
         self.songs = list(graph.items(playlistUri))
         self.show = show
         self.graph.addHandler(self._readSongs)
     def nextSong(self, currentSong):
     def _readSongs(self):
         self._songs = []
         self._labels = {}
         self._songUri = {}
         self._fileUri = {}
         playlistUri = self.graph.value(self.show, L9['playList'], default=None)
         if not playlistUri:
             # expected at startup that we'd have no data
             return
         prevSong = None
         for s in self.graph.items(playlistUri):
             f = GstFileUri('file://' + str(songOnDisk(self.graph, s)))
             self._songs.append(s)
             self._labels[s] = self.graph.value(s, RDFS.label, default="").toPython()
             self._songUri[f] = s
             self._fileUri[s] = f
             if prevSong is not None:
                 self._nextSong[prevSong] = s
         log.info(f'read {len(self._songs)} songs')
     def nextSong(self, currentSong: URIRef) -> URIRef:
         """Returns the next song in the playlist or raises NoSuchSong if
         we are at the end of the playlist."""
         try:
             currentIndex = self.songs.index(currentSong)
         except IndexError:
             raise ValueError("%r is not in the current playlist (%r)." %
                              (currentSong, self.playlistUri))
             return self._nextSong[currentSong]
         except KeyError:
             raise NoSuchSong()
         try:
             nextSong = self.songs[currentIndex + 1]
         except IndexError:
             raise NoSuchSong("%r is the last item in the playlist." %
                              currentSong)
     def fileUri(self, song: URIRef) -> GstFileUri:
         return self._fileUri[song]
         return nextSong
     def songUri(self, fileUri: GstFileUri) -> URIRef:
         return self._songUri[fileUri]
     def label(self, song: URIRef) -> str:
         return self._labels[song]
     def allSongs(self):
         """Returns a list of all song URIs in order."""
         return self.songs
     def allSongPaths(self):
         """Returns a list of the filesystem paths to all songs in order."""
         paths = []
         for song in self.songs:
             paths.append(songOnDisk(song))
         return paths
         return self._songs
     def songPath(self, uri):
         """filesystem path to a song"""
         raise NotImplementedError("see showconfig.songOnDisk")
         # maybe that function should be moved to this method
     @classmethod
     def fromShow(cls, graph, show):
         playlistUri = graph.value(show, L9['playList'])
         if not playlistUri:
             raise ValueError("%r has no l9:playList" % show)
         return cls(graph, playlistUri)
     # def allSongPaths(self) -> list[Path]:
     #     """Returns a list of the filesystem paths to all songs in order."""
     #     paths = []
     #     for row in self._songs:
     #         paths.append(row.songPath)
     #     return paths

src/light9/ascoltami/webapp.py

➞

Show inline comments

 """
 this module shouldn't be necessary for playback to work
 """
 import asyncio
 import json
 import logging
 import socket
 import subprocess
 from dataclasses import dataclass, field
 import time
 from typing import cast
 from typing import Callable
 from typing import Literal as Lit
 from rdflib import RDFS, Graph, URIRef
 import louie
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import URIRef
 from sse_starlette.sse import EventSourceResponse
 from starlette.requests import Request
 from starlette.responses import JSONResponse, PlainTextResponse
 from light9.ascoltami.player import Player
 from light9.namespaces import L9
 from light9.showconfig import getSongsFromShow, showUri, songOnDisk
 from light9.ascoltami.player import Player, PlayerState
 from light9.ascoltami.playlist import Playlist
 from light9.showconfig import showUri
 log = logging.getLogger()
 _songUris = {}  # locationUri : song
 log = logging.getLogger("web")
 class OnStateChange:
     pass
 def songLocation(graph, songUri):
     loc = URIRef("file://%s" % songOnDisk(songUri))
     _songUris[loc] = songUri
     return loc
 @dataclass
 class PlayerState2(PlayerState):
     song2: URIRef | None = None
     nextAction: Lit["finish"] | Lit['disabled'] | Lit['play'] = 'disabled'
 def songUri(graph, locationUri):
     return _songUris[locationUri]
 @dataclass
 class WebHandlers:
     graph: SyncedGraph
     show: URIRef
     player: Player
     playlist: Playlist
     getPlayerState: Callable[[], PlayerState]
 async def get_config(request: Request) -> JSONResponse:
     # _keep:list=field(default_factory=list)
     async def get_config(self, request: Request) -> JSONResponse:
     return JSONResponse(
         dict(
             host=socket.gethostname(),
             show=str(showUri()),
             times={
                 # these are just for the web display. True values are on Player.__init__
                 'intro': 4,
                 'post': 0
             }))
 def playerSongUri(graph, player):
     """or None"""
     playingLocation = player.getSong()
     if playingLocation:
         return songUri(graph, URIRef(playingLocation))
     else:
         return None
 def currentState(graph, player):
     def currentState(self, player: Player, playlist: Playlist) -> PlayerState2:
     if player.isAutostopped():
         nextAction = 'finish'
     elif player.isPlaying():
         nextAction = 'disabled'
     else:
         nextAction = 'play'
     return {
         "song": playerSongUri(graph, player),
         "started": player.playStartTime,
         "duration": player.duration(),
         "playing": player.isPlaying(),
         "t": player.currentTime(),
         "state": player.states(),
         "next": nextAction,
+    }
         ps = self.getPlayerState()
         return PlayerState2(
             song2=playlist.songUri((ps.song)) if ps.song else None,
             duration=ps.duration,
             playing=ps.playing,
             # state= player.states(),
             nextAction=nextAction,
+        )
     async def get_time(self, request: Request) -> JSONResponse:
         return JSONResponse({'t': self.player.currentTime()})
 async def get_time(request: Request) -> JSONResponse:
     player = cast(Player, request.app.state.player)
     graph = cast(Graph, request.app.state.graph)
     return JSONResponse(currentState(graph, player))
 async def post_time(request: Request) -> PlainTextResponse:
     async def post_time(self, request: Request) -> PlainTextResponse:
     """
     post a json object with {pause: true} or {resume: true} if you
     want those actions. Use {t: <seconds>} to seek, optionally
     with a pause/resume command too.
     """
     params = await request.json()
     player = cast(Player, request.app.state.player)
     if params.get('pause', False):
         player.pause()
+            self.player.pause()
     if params.get('resume', False):
         player.resume()
+            self.player.resume()
     if 't' in params:
         player.seek(params['t'])
+            self.player.seek(params['t'])
     return PlainTextResponse("ok")
 async def timeStream(request: Request):
     graph = cast(Graph, request.app.state.graph)
     player = cast(Player, request.app.state.player)
     async def timeStream(self, request: Request):
     async def event_generator():
         last_sent = None
         last_sent_time = 0.0
             def onStateChange(s: PlayerState2):
                 log.info('ws heanndlerr gets state')
             louie.connect(onStateChange, OnStateChange, weak=False)
             try:
         while True:
             now = time.time()
-            msg = currentState(graph, player)
+                    msg = self.currentState(self.player, self.playlist)
             if msg != last_sent or now > last_sent_time + 2:
                 event_data = json.dumps(msg)
                         event_data = json.dumps({
                             # obsolete- watch the graph for these
                             'duration': msg.duration,
                             'playing': msg.playing,
                             'song': self.playlist.songUri(msg.song) if msg.song else None,
                             'state': {},
                         })
                 yield event_data
                 last_sent = msg
                 last_sent_time = now
             await asyncio.sleep(0.1)
             finally:
                 log.info(f'bye listnner {event_generator}')
                 louie.disconnect(onStateChange, OnStateChange, weak=False)
     return EventSourceResponse(event_generator())
 async def get_songs(request: Request) -> JSONResponse:
     graph = cast(Graph, request.app.state.graph)
     songs = getSongsFromShow(graph, request.app.state.show)
     async def get_songs(self, request: Request) -> JSONResponse:
     songs_data = [
         {  #
             "uri": s,
             "path": graph.value(s, L9['songFilename']),
             "label": graph.value(s, RDFS.label)
         } for s in songs
                 "path": self.playlist.fileUri(s),
                 "label": self.playlist.label(s),
             } for s in self.playlist.allSongs()
+    ]
     return JSONResponse({"songs": songs_data})
 async def post_song(request: Request) -> PlainTextResponse:
     async def post_song(self, request: Request) -> PlainTextResponse:
     """post a uri of song to switch to (and start playing)"""
     graph = cast(Graph, request.app.state.graph)
     player = cast(Player, request.app.state.player)
     song_uri = URIRef((await request.body()).decode('utf8'))
-    player.setSong(songLocation(graph, song_uri))
+        self.player.setSong(self.playlist.fileUri(song_uri))
     return PlainTextResponse("ok")
 async def post_seekPlayOrPause(request: Request) -> PlainTextResponse:
     async def post_seekPlayOrPause(self, request: Request) -> PlainTextResponse:
     """curveCalc's ctrl-p or a vidref scrub"""
     player = cast(Player, request.app.state.player)
     data = await request.json()
     if 'scrub' in data:
         player.pause()
         player.seek(data['scrub'])
             self.player.pause()
             self.player.seek(data['scrub'])
         return PlainTextResponse("ok")
     if 'action' in data:
         if data['action'] == 'play':
             player.resume()
+                self.player.resume()
         elif data['action'] == 'pause':
             player.pause()
+                self.player.pause()
         else:
             raise NotImplementedError
         return PlainTextResponse("ok")
     if player.isPlaying():
         player.pause()
         if self.player.isPlaying():
             self.player.pause()
     else:
         player.seek(data['t'])
         player.resume()
             self.player.seek(data['t'])
             self.player.resume()
     return PlainTextResponse("ok")
 async def post_output(request: Request) -> PlainTextResponse:
     async def post_output(self, request: Request) -> PlainTextResponse:
     d = await request.json()
     subprocess.check_call(["bin/movesinks", str(d['sink'])])
     return PlainTextResponse("ok")
 async def post_goButton(request: Request) -> PlainTextResponse:
     async def post_goButton(self, request: Request) -> PlainTextResponse:
     """
     if music is playing, this silently does nothing.
     """
     player = cast(Player, request.app.state.player)
     if player.isAutostopped():
         player.resume()
     elif player.isPlaying():
         if self.player.isAutostopped():
             self.player.resume()
         elif self.player.isPlaying():
         pass
     else:
         player.resume()
+            self.player.resume()
     return PlainTextResponse("ok")

src/light9/showconfig.py

➞

Show inline comments

 import logging, warnings
 import logging
 import os
 import warnings
 from os import getenv
 from pathlib import Path
 from typing import cast
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import Graph, Literal, URIRef
 from twisted.python.filepath import FilePath
 from os import path, getenv
 from rdflib import Graph
 from rdflib import URIRef, Literal
 from .namespaces import L9
-from typing import List, cast
 log = logging.getLogger('showconfig')
 _config = None  # graph
 def getGraph() -> Graph:
     warnings.warn(
         "code that's using showconfig.getGraph should be "
         "converted to use the sync graph",
         stacklevel=2)
     warnings.warn("code that's using showconfig.getGraph should be "
                   "converted to use the sync graph", stacklevel=2)
     global _config
     if _config is None:
         graph = Graph()
         # note that logging is probably not configured the first time
         # we're in here
         warnings.warn("reading n3 files around %r" % root())
         for f in FilePath(root()).globChildren("*.n3") + FilePath(
                 root()).globChildren("build/*.n3"):
         for f in FilePath(root()).globChildren("*.n3") + FilePath(root()).globChildren("build/*.n3"):
             graph.parse(location=f.path, format='n3')
         _config = graph
     return _config
 def root() -> bytes:
     r = getenv("LIGHT9_SHOW")
     if r is None:
         raise OSError(
             "LIGHT9_SHOW env variable has not been set to the show root")
         raise OSError("LIGHT9_SHOW env variable has not been set to the show root")
     return r.encode('ascii')
 _showUri = None
 def showUri() -> URIRef:
     """Return the show URI associated with $LIGHT9_SHOW."""
     global _showUri
     if _showUri is None:
         _showUri = URIRef(open(path.join(root(), b'URI')).read().strip())
+        _showUri = URIRef(open(os.path.join(root(), b'URI')).read().strip())
     return _showUri
-def songOnDisk(song: URIRef) -> bytes:
+def songOnDisk(graph: SyncedGraph, song: URIRef) -> Path:
     """given a song URI, where's the on-disk file that mpd would read?"""
     graph = getGraph()
     root = graph.value(showUri(), L9['musicRoot'])
     if not root:
         raise ValueError("%s has no :musicRoot" % showUri())
     name = graph.value(song, L9['songFilename'])
     if not name:
         raise ValueError("Song %r has no :songFilename" % song)
     return path.abspath(
         path.join(
             cast(Literal, root).toPython(),
             cast(Literal, name).toPython()))
     return (Path(cast(Literal, root).toPython()) / cast(Literal, name).toPython()).absolute()
 def songFilenameFromURI(uri: URIRef) -> bytes:
     """
     'http://light9.bigasterisk.com/show/dance2007/song8' -> 'song8'
     everything that uses this should be deprecated for real URIs
     everywhere"""
     assert isinstance(uri, URIRef)
     return str(uri).split('/')[-1].encode('ascii')
-def getSongsFromShow(graph: Graph, show: URIRef) -> List[URIRef]:
+def getSongsFromShow(graph: Graph, show: URIRef) -> list[URIRef]:
     playList = graph.value(show, L9['playList'])
     if not playList:
         raise ValueError("%r has no l9:playList" % show)
     # The patch in https://github.com/RDFLib/rdflib/issues/305 fixed a
     # serious bug here.
     songs = list(graph.items(playList))
     return songs
+    return cast(list[URIRef], songs)
 def curvesDir():
     return path.join(root(), b"curves")
+    return os.path.join(root(), b"curves")
 def subFile(subname):
     return path.join(root(), b"subs", subname)
+    return os.path.join(root(), b"subs", subname)
 def subsDir():
     return path.join(root(), b'subs')
+    return os.path.join(root(), b'subs')

web/ascoltami/main.ts

➞

Show inline comments

 function byId(id: string): HTMLElement {
   return document.getElementById(id)!;
+}
 // obsolete: watch the graph instead
 export interface TimingUpdate {
   // GET /ascoltami/time response
   duration: number;
   next: string; // e.g. 'play'
   playing: boolean;
   song: string;
@@ @@ -15,13 +16,13 @@ export interface TimingUpdate { @@
 (window as any).finishOldStyleSetup = async (times: { intro: number; post: number }, timingUpdate: (data: TimingUpdate) => void) => {
   let currentHighlightedSong = "";
   // let lastPlaying = false;
-  const events = new EventSource("../service/ascoltami/time/stream");
   const events = new EventSource("/service/ascoltami/time/stream");
   events.addEventListener("message", (m)=>{
     const update = JSON.parse(m.data) as TimingUpdate
     updateCurrent(update)
     markUpdateTiming();
   })

0 comments (0 inline, 0 general)