BlinkenArea - GitList
Repositories
Blog
Wiki
stage_director
Code
Commits
Branches
Tags
Search
Tree:
014a03d
Branches
Tags
master
stage_director
stage_director.py
fix warning about Gtk version
Stefan Schuermans
commited
014a03d
at 2018-07-18 20:23:37
stage_director.py
Blame
History
Raw
#! /usr/bin/env python # BlinkenArea Stage Director # Copyright 2013-2014 Stefan Schuermans <stefan@schuermans.info> # Copyleft: GNU public license - http://www.gnu.org/copyleft/gpl.html # a blinkenarea.org project - https://www.blinkenarea.org/ import gi gi.require_version('Gtk', '3.0') import os from gi.repository import GObject from gi.repository import Gtk from gi.repository import Pango import socket import struct import sys import time import playlist import time_fmt scriptdir = os.path.dirname(os.path.abspath(__file__)) class StageDirector: def __init__(self): """construct a StageDirector object""" self.builder = Gtk.Builder() self.builder.add_from_file(scriptdir + "/stage_director.glade") self.widMainWindow = self.builder.get_object("MainWindow") self.widPlaylistView = self.builder.get_object("PlaylistView") self.widPlaylistStore = self.builder.get_object("PlaylistStore") self.widPosition = self.builder.get_object("Position") self.widPositionScale = self.builder.get_object("PositionScale") self.widPositionAt = self.builder.get_object("PositionAt") self.widPositionRemaining = self.builder.get_object("PositionRemaining") self.widBtnPlay = self.builder.get_object("Play") self.widBtnPause = self.builder.get_object("Pause") self.widLogoStop = self.builder.get_object("LogoStop") self.widLogoPlay = self.builder.get_object("LogoPlay") self.widLogoUdpErr = self.builder.get_object("LogoUdpErr") self.widLogoUdpOk = self.builder.get_object("LogoUdpOk") self.widSingleStep = self.builder.get_object("SingleStep") self.widUdpOutput = self.builder.get_object("UdpOutput") self.widStatus = self.builder.get_object("Status") handlers = { "onDestroy": self.onDestroy, "onFileOpen": self.onFileOpen, "onFileExit": self.onFileExit, "onExtrasDestination": self.onExtrasDestination, "onExtrasAbout": self.onExtrasAbout, "onPlaylistDblClick": self.onPlaylistDblClick, "onNewPosition": self.onNewPosition, "onPrevious": self.onPrevious, "onPlay": self.onPlay, "onPause": self.onPause, "onStop": self.onStop, "onNext": self.onNext, "onUdpOutput": self.onUdpOutput, } self.builder.connect_signals(handlers) self.playlist = playlist.Playlist() if len(sys.argv) >= 2: # load initial playlist from command line self.playlist.read(sys.argv[1]) self.playlist.update(self.widPlaylistStore) self.sock = None self.stEntryIdx = -1 # no entry selected self.stName = "" # no current entry name self.stDuration = 0 # current entry has zero size self.stPosition = 0 # at begin of current entry self.stPlaying = False # not playing self.stDestination = "255.255.255.255" # local LAN broadcast by default self.setupSock() self.updateEntry() self.updateButtonVisibility() GObject.timeout_add(10, self.onTimer10ms) GObject.timeout_add(100, self.onTimer100ms) def showPosition(self): """update the position texts next to the position slider""" # format current time and remaining time posAt = time_fmt.sec2str(self.stPosition) posRemaining = time_fmt.sec2str(self.stDuration - self.stPosition) self.widPositionAt.set_text(posAt) self.widPositionRemaining.set_text(posRemaining) def updatePositionState(self): """update the position in the state, but not the slider""" # calculate (virtual) start time of playing # i.e. the time the playing would have had started to arrive at the # current position now if it had played continuosly self.stPlayStart = time.time() - self.stPosition # update position texts self.showPosition() def updatePosition(self): """update the position including the position slider""" # update GUI slider self.widPositionScale.set_value(self.stPosition) # update position state self.updatePositionState() def updateDuration(self): """update the duration (i.e. range for the slider) based on the current playlist entry""" # get duration of new playlist entry self.stDuration = 0 if self.stEntryIdx >= 0: entry = self.playlist.entries[self.stEntryIdx] if entry["type"] == "normal": self.stDuration = entry["duration"] # set position to begin self.stPosition = 0 # update value range self.widPosition.set_upper(self.stDuration) # update position of slider self.updatePosition() def updateEntry(self): """update current entry of playlist and duration, position, ...""" # clear selection of playlist sel = self.widPlaylistView.get_selection() if sel: sel.unselect_all() # sanity check for entry index if self.stEntryIdx < -1 or self.stEntryIdx >= len(self.playlist.entries): self.stEntryIdx = -1 # get name of current entry self.stName = "" if self.stEntryIdx >= 0: self.stName = self.playlist.entries[self.stEntryIdx]["name"] # make current entry bold, all others non-bold # scroll current entry into view def update(model, path, it, stageDirector): (idx,) = model.get(it, 0) if idx == self.stEntryIdx: weight = Pango.Weight.BOLD stageDirector.widPlaylistView.scroll_to_cell(path) else: weight = Pango.Weight.NORMAL model.set(it, 1, weight) self.widPlaylistStore.foreach(update, self) # playing and (no entry or stop entry or single-step mode) # -> stop playing and update button visibility if self.stPlaying and \ (self.stEntryIdx < 0 or \ self.playlist.entries[self.stEntryIdx]["type"] == "stop" or \ self.widSingleStep.get_active()): self.stPlaying = False self.updateButtonVisibility() # update duration, position, ... self.updateDuration() def updateButtonVisibility(self): """update the visibility of the buttons based on if playing or not""" self.widBtnPlay.set_visible(not self.stPlaying) self.widBtnPause.set_visible(self.stPlaying) self.widLogoStop.set_visible(not self.stPlaying) self.widLogoPlay.set_visible(self.stPlaying) def closeSock(self): """close UDP socket""" self.widLogoUdpErr.set_visible(True) self.widLogoUdpOk.set_visible(False) self.widStatus.remove_all(0) self.widStatus.push(0, "UDP output turned off") if self.sock is not None: self.sock.close() self.sock = None def setupSock(self): """create a new UDP socket and "connect" it to the destination address""" self.closeSock() if self.widUdpOutput.get_active(): try: self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) self.sock.connect((self.stDestination, 5740)) self.widStatus.remove_all(0) self.widStatus.push(0, "UDP output to \"" + self.stDestination + "\" port 5740") self.widLogoUdpErr.set_visible(False) self.widLogoUdpOk.set_visible(True) except: self.closeSock() self.widStatus.push(0, "UDP output ERROR") def onDestroy(self, widget): """window will be destroyed""" Gtk.main_quit() def onFileOpen(self, widget): """File Open clicked in menu""" #print("DEBUG stage_director File Open") # create and run file chooser dialog dialog = Gtk.FileChooserDialog( "BlinkenArea Stage Director - File Open...", self.widMainWindow, Gtk.FileChooserAction.OPEN, (Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL, Gtk.STOCK_OPEN, Gtk.ResponseType.OK)) dialog.set_default_response(Gtk.ResponseType.OK) filt = Gtk.FileFilter() filt.set_name("All files") filt.add_pattern("*") dialog.add_filter(filt) response = dialog.run() if response == Gtk.ResponseType.OK: # dialog closed with OK -> load new playlist filename = dialog.get_filename() self.playlist.read(filename) self.playlist.update(self.widPlaylistStore) self.stEntryIdx = -1 # no entry selected self.stPlaying = False # not playing self.updateEntry() self.updateButtonVisibility() # clean up dialog.destroy() def onFileExit(self, widget): """File Exit clicked in menu""" #print("DEBUG stage_director File Exit") Gtk.main_quit() def onExtrasDestination(self, widget): """Extras Destination Address clicked in menu""" #print("DEBUG stage_director Extras Destination") # run input dialog to ask for new destination dialog = self.builder.get_object("DialogDestination") cur = self.builder.get_object("DiaDestCur") new = self.builder.get_object("DiaDestNew") cur.set_text(self.stDestination) new.set_text(self.stDestination) response = dialog.run() if response == 1: self.stDestination = new.get_text() # hide input dialog dialog.hide() # re-create UDP socket self.setupSock() def onExtrasAbout(self, widget): """Extras About clicked in menu""" #print("DEBUG stage_director Extras About") # run about dialog dialog = self.builder.get_object("DialogAbout") dialog.run() dialog.hide() def onPlaylistDblClick(self, widget, row, col): """playlist entry has been double-clicked""" # get index of selected entry idx = -1 sel = self.widPlaylistView.get_selection() if sel is not None: (model, it) = sel.get_selected() if it is not None: (idx,) = model.get(it, 0) #print("DEBUG stage_director playlist double-click idx=%d" % (idx)) # update playlist entry self.stEntryIdx = idx # set position to zero if playing if self.stPlaying: self.stPosition = 0 # update entry self.updateEntry() def onNewPosition(self, widget, scroll, value): """slider has been moved to a new position""" #print("DEBUG stage_director new position " + str(value)); # clamp position to valid range if value < 0: value = 0 if value > self.stDuration: value = self.stDuration # update current position - and play start time if playing self.stPosition = value # update position state (do not touch the slider) self.updatePositionState() def onPrevious(self, widget): """previous button as been pressed""" #print("DEBUG stage_director previous") # go to begin of previous entry (with wrap around) self.stPosition = 0 self.stEntryIdx = self.stEntryIdx - 1 if self.stEntryIdx < 0: self.stEntryIdx = len(self.playlist.entries) - 1 self.updateEntry() def onPlay(self, widget): """play button has been pressed""" #print("DEBUG stage_director play") self.stPlaying = True self.updatePosition() self.updateButtonVisibility() def onPause(self, widget): """pause button has been pressed""" #print("DEBUG stage_director pause") self.stPlaying = False self.updateButtonVisibility() def onStop(self, widget): """stop button has been pressed""" #print("DEBUG stage_director stop") self.stPlaying = False self.stPosition = 0 # stop goes back to begin self.updatePosition() self.updateButtonVisibility() def onNext(self, widget): """next button has been pressed""" #print("DEBUG stage_director next") # go to begin of next entry (with wrap around) self.stPosition = 0 self.stEntryIdx = self.stEntryIdx + 1 if self.stEntryIdx >= len(self.playlist.entries): self.stEntryIdx = 0 self.updateEntry() def onUdpOutput(self, widget): """UDP Output check box toggled""" # re-create UDP socket self.setupSock() def onTimer10ms(self): """timer callback, every 10ms""" # update position if playing if self.stPlaying: self.stPosition = time.time() - self.stPlayStart if self.stPosition >= self.stDuration: # end of entry reached --> go to begin of next entry (with wrap around) self.stPosition = 0 self.stEntryIdx = self.stEntryIdx + 1 if self.stEntryIdx >= len(self.playlist.entries): self.stEntryIdx = 0 self.updateEntry() else: self.updatePosition() # request being called again return True def onTimer100ms(self): """timer callback, every 100ms""" # send sync packet if self.sock is not None: flags = 0 if not self.stPlaying: flags = flags | 1 name = self.stName pos_ms = round(self.stPosition * 1000) data = "PoSy" + struct.pack("!I64sI", flags, name, pos_ms) try: self.sock.send(data) except: self.closeSock() # request being called again return True # main application entry point if __name__ == "__main__": # configure settings try: settings = Gtk.Settings.get_default() settings.props.gtk_button_images = True except: pass # create window app = StageDirector() # run application Gtk.main()
