1
0
Fork 0

pep8 cleanups

This commit is contained in:
Kevin Matz 2020-11-23 12:31:44 -05:00
parent 17081bc219
commit ef37f016d4
7 changed files with 113 additions and 80 deletions

View File

@ -1,5 +1,4 @@
"""OscCommentMacroListener.py: Hog 4 comment macro antlr4 listener for OSC. """OscCommentMacroListener.py: Hog 4 comment macro antlr4 listener for OSC."""
"""
import logging import logging
@ -12,7 +11,8 @@ from .hog4 import HogNet
# https://raw.githubusercontent.com/jszheng/py3antlr4book/master/bin/pygrun # https://raw.githubusercontent.com/jszheng/py3antlr4book/master/bin/pygrun
# this is a python version of TestRig # this is a python version of TestRig
def beautify_lisp_string(in_string): def beautify_lisp_string(in_string):
__author__ = 'jszheng' """Prety output of a lisp string."""
# __author__ = 'jszheng'
indent_size = 2 indent_size = 2
add_indent = ' ' * indent_size add_indent = ' ' * indent_size
out_string = in_string[0] # no indent for 1st ( out_string = in_string[0] # no indent for 1st (
@ -31,27 +31,35 @@ def beautify_lisp_string(in_string):
class OscCommentMacroListener(CommentMacroListener): class OscCommentMacroListener(CommentMacroListener):
"""antlr4 listener for comment macros."""
def __init__(self): def __init__(self):
"""Initialize the listener."""
self.osc = None self.osc = None
def exitWait(self, ctx: CommentMacroParser.WaitContext): def exitWait(self, ctx: CommentMacroParser.WaitContext):
logging.info("Waiting " + str(ctx.number().value) + " seconds.") """On exiting the WAIT macro."""
logging.info("Waiting %d seconds.", ctx.number().value)
sleep(ctx.number().value) sleep(ctx.number().value)
def enterStatement(self, ctx: CommentMacroParser.StatementContext): def enterStatement(self, ctx: CommentMacroParser.StatementContext):
"""On entering a statement."""
# print the lisp tree of this macro # print the lisp tree of this macro
lisp_tree_str = ctx.toStringTree(recog=ctx.parser) lisp_tree_str = ctx.toStringTree(recog=ctx.parser)
logging.debug(beautify_lisp_string(lisp_tree_str)) logging.debug(beautify_lisp_string(lisp_tree_str))
def enterTarget(self, ctx: CommentMacroParser.TargetContext): def enterTarget(self, ctx: CommentMacroParser.TargetContext):
"""On entering a target."""
ctx.targets = [] ctx.targets = []
def exitTarget(self, ctx: CommentMacroParser.TargetContext): def exitTarget(self, ctx: CommentMacroParser.TargetContext):
"""On exiting a target."""
ctx.targets = set(ctx.targets) ctx.targets = set(ctx.targets)
if isinstance(ctx.parentCtx, CommentMacroParser.TargetContext): if isinstance(ctx.parentCtx, CommentMacroParser.TargetContext):
ctx.parentCtx.targets.extend(ctx.targets) # add to parent targets ctx.parentCtx.targets.extend(ctx.targets) # add to parent targets
def exitNumber(self, ctx: CommentMacroParser.NumberContext): def exitNumber(self, ctx: CommentMacroParser.NumberContext):
"""On exiting a number."""
try: try:
ctx.value = int(ctx.getText()) ctx.value = int(ctx.getText())
except ValueError: except ValueError:
@ -60,6 +68,7 @@ class OscCommentMacroListener(CommentMacroListener):
ctx.parentCtx.targets.append(ctx.value) ctx.parentCtx.targets.append(ctx.value)
def exitSpan(self, ctx: CommentMacroParser.SpanContext): def exitSpan(self, ctx: CommentMacroParser.SpanContext):
"""On exiting a span."""
lower = min(ctx.n1.value, ctx.n2.value) lower = min(ctx.n1.value, ctx.n2.value)
upper = max(ctx.n1.value, ctx.n2.value) upper = max(ctx.n1.value, ctx.n2.value)
while lower <= upper: while lower <= upper:
@ -67,47 +76,49 @@ class OscCommentMacroListener(CommentMacroListener):
lower += 1 lower += 1
def exitMasterGo(self, ctx: CommentMacroParser.MasterGoContext): def exitMasterGo(self, ctx: CommentMacroParser.MasterGoContext):
"""On exiting a Go macro."""
if ctx.target() is None: if ctx.target() is None:
logging.info("Main GO") logging.info("Main GO")
self.osc.button_press(ctx.device(), "/hog/hardware/maingo") self.osc.button_press(ctx.device(), "/hog/hardware/maingo")
return return
for i in ctx.target().targets: for i in ctx.target().targets:
if isinstance(i, int) is not True: if isinstance(i, int) is not True:
logging.warn("GO MASTER macro targets must be intigers. " logging.error("GO MASTER macro targets must be intigers.")
+ str(i) + " is not an intigers.")
continue continue
if (i < 0): if i < 0:
logging.warn("Master "+str(i)+" is not greater than 0.") logging.error("Master %d is not greater than 0.", i)
continue continue
master = str(i) master = str(i)
logging.info("GO on master " + master) logging.info("GO on master %s", master)
self.osc.button_press(ctx.device(), "/hog/hardware/go/" + master) self.osc.button_press(ctx.device(), "/hog/hardware/go/" + master)
def exitMasterHalt(self, ctx: CommentMacroParser.MasterHaltContext): def exitMasterHalt(self, ctx: CommentMacroParser.MasterHaltContext):
"""On exiting a Master Halt macro."""
if ctx.target() is None: if ctx.target() is None:
logging.info("Main HALT") logging.info("Main HALT")
self.osc.button_press(ctx.device(), "/hog/hardware/mainhalt") self.osc.button_press(ctx.device(), "/hog/hardware/mainhalt")
return return
for i in ctx.target().targets: for i in ctx.target().targets:
if isinstance(i, int) is not True: if isinstance(i, int) is not True:
logging.warn("GO MASTER macro targets must be intigers. " logging.error("GO MASTER macro targets must be intigers.")
+ str(i) + " is not an intigers.")
continue continue
if (i < 0): if i < 0:
logging.warn("Master "+str(i)+" is not greater than 0.") logging.error("Master %d is not greater than 0.", i)
continue continue
master = str(i) master = str(i)
logging.info("HALT on master " + master) logging.info("HALT on master %s", master)
self.osc.button_press(ctx.device(), "/hog/hardware/pause/"+master) self.osc.button_press(ctx.device(), "/hog/hardware/pause/"+master)
def exitMasterAssert(self, ctx: CommentMacroParser.MasterAssertContext): def exitMasterAssert(self, ctx: CommentMacroParser.MasterAssertContext):
if ctx.target()is not None: """On exiting a Master Assert macro."""
if ctx.target() is not None:
logging.error("ERROR: limited to asserting current master only.") logging.error("ERROR: limited to asserting current master only.")
return return
logging.info("ASSERT on current master.") logging.info("ASSERT on current master.")
self.osc.button_press(ctx.device(), "/hog/hardware/assert") self.osc.button_press(ctx.device(), "/hog/hardware/assert")
def exitMasterRelease(self, ctx: CommentMacroParser.MasterReleaseContext): def exitMasterRelease(self, ctx: CommentMacroParser.MasterReleaseContext):
"""On exiting a Master Release macro."""
if ctx.target() is not None: if ctx.target() is not None:
logging.error("ERROR: limited to releasing current master only.") logging.error("ERROR: limited to releasing current master only.")
return return
@ -115,7 +126,8 @@ class OscCommentMacroListener(CommentMacroListener):
self.osc.button_press(ctx.device(), "/hog/hardware/release") self.osc.button_press(ctx.device(), "/hog/hardware/release")
def exitMasterFade(self, ctx: CommentMacroParser.MasterFadeContext): def exitMasterFade(self, ctx: CommentMacroParser.MasterFadeContext):
if ctx.target()is None: """On exiting a Master Fade macro."""
if ctx.target() is None:
logging.error("ERROR: limited to fading specified masters only.") logging.error("ERROR: limited to fading specified masters only.")
return return
level = ctx.number().value level = ctx.number().value
@ -124,14 +136,13 @@ class OscCommentMacroListener(CommentMacroListener):
return return
for i in ctx.target().targets: for i in ctx.target().targets:
if isinstance(i, int) is not True: if isinstance(i, int) is not True:
logging.warn("FADE MASTER macro targets must be intigers. " logging.error("FADE MASTER macro targets must be intigers.")
+ str(i) + " is not an intigers.")
continue continue
if (i < 0): if i < 0:
logging.warn("Master "+str(i)+" is not greater than 0.") logging.error("Master %d is not greater than 0.", i)
continue continue
master = str(i) master = str(i)
logging.info("Fade Master "+master+" to "+str(level)+"%") logging.info("Fade Master %s to %d", master, level)
level *= 255 / 100 # percent in Macro, 0>255 in OSC level *= 255 / 100 # percent in Macro, 0>255 in OSC
self.osc.send_message(ctx.device(), self.osc.send_message(ctx.device(),
"/hog/hardware/fader/" + master, "/hog/hardware/fader/" + master,
@ -139,23 +150,26 @@ class OscCommentMacroListener(CommentMacroListener):
def exitFadeGrandMaster(self, def exitFadeGrandMaster(self,
ctx: CommentMacroParser.FadeGrandMasterContext): ctx: CommentMacroParser.FadeGrandMasterContext):
"""On exiting a GM Fade Macro."""
level = ctx.number().value level = ctx.number().value
if (level < 0 or level > 100): if (level < 0 or level > 100):
logging.error("Level must be between 0 and 100.") logging.error("Level must be between 0 and 100.")
return return
logging.info("Fading Grand Master to " + str(level) + "%") logging.info("Fading Grand Master to %d", level)
level *= 255 / 100 # percent in Macro, 0>255 in OSC level *= 255 / 100 # percent in Macro, 0>255 in OSC
self.osc.send_message(ctx.device(), "/hog/hardware/fader/0", level) self.osc.send_message(ctx.device(), "/hog/hardware/fader/0", level)
def exitMasterChoose(self, ctx: CommentMacroParser.MasterChooseContext): def exitMasterChoose(self, ctx: CommentMacroParser.MasterChooseContext):
if (ctx.number().value < 0): """On exiting a Master Choose macro."""
if ctx.number().value < 0:
logging.error("Master must be greater than 0.") logging.error("Master must be greater than 0.")
return return
master = str(ctx.number().value) master = str(ctx.number().value)
logging.info("Choose Master " + master) logging.info("Choose Master %s", master)
self.osc.button_press(ctx.device(), "/hog/hardware/choose/" + master) self.osc.button_press(ctx.device(), "/hog/hardware/choose/" + master)
def exitReleaseAll(self, ctx: CommentMacroParser.ReleaseAllContext): def exitReleaseAll(self, ctx: CommentMacroParser.ReleaseAllContext):
"""On exiting a Release All Macro."""
logging.info("Release All") logging.info("Release All")
self.osc.send_message(ctx.device(), "/hog/hardware/pig", self.osc.send_message(ctx.device(), "/hog/hardware/pig",
self.osc.buttonDOWN) self.osc.buttonDOWN)
@ -164,60 +178,72 @@ class OscCommentMacroListener(CommentMacroListener):
self.osc.buttonUP) self.osc.buttonUP)
def exitListGo(self, ctx: CommentMacroParser.ListGoContext): def exitListGo(self, ctx: CommentMacroParser.ListGoContext):
"""On exiting a List GO macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Go on List " + str(i)) logging.info("Go on List %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/go/0", i) self.osc.send_message(ctx.device(), "/hog/playback/go/0", i)
def exitListGoto(self, ctx: CommentMacroParser.ListGotoContext): def exitListGoto(self, ctx: CommentMacroParser.ListGotoContext):
"""On exiting a GOTO macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
list = str(i) + "." + str(ctx.number().value) cuelist = str(i) + "." + str(ctx.number().value)
logging.info("Go on List " + str(list)) logging.info("Go on List %s", cuelist)
self.osc.send_message(ctx.device(), "/hog/playback/go/0", list) self.osc.send_message(ctx.device(), "/hog/playback/go/0", list)
def exitListHalt(self, ctx: CommentMacroParser.ListHaltContext): def exitListHalt(self, ctx: CommentMacroParser.ListHaltContext):
"""On exiting a Halt macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Halting List " + str(i)) logging.info("Halting List %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/halt/0", i) self.osc.send_message(ctx.device(), "/hog/playback/halt/0", i)
def exitListRelese(self, ctx: CommentMacroParser.ListReleseContext): def exitListRelese(self, ctx: CommentMacroParser.ListReleseContext):
"""On exiting a Release macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Releasing List " + str(i)) logging.info("Releasing List %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/release/0", i) self.osc.send_message(ctx.device(), "/hog/playback/release/0", i)
def exitSceneGo(self, ctx: CommentMacroParser.SceneGoContext): def exitSceneGo(self, ctx: CommentMacroParser.SceneGoContext):
"""On exiting a Scene GO macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Go on Scene " + str(i)) logging.info("Go on Scene %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/go/1", i) self.osc.send_message(ctx.device(), "/hog/playback/go/1", i)
def exitSceneHalt(self, ctx: CommentMacroParser.SceneHaltContext): def exitSceneHalt(self, ctx: CommentMacroParser.SceneHaltContext):
"""On exiting a Scene Halt macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Halt Scene " + str(i)) logging.info("Halt Scene %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/halt/1", i) self.osc.send_message(ctx.device(), "/hog/playback/halt/1", i)
def exitSceneRelease(self, ctx: CommentMacroParser.SceneReleaseContext): def exitSceneRelease(self, ctx: CommentMacroParser.SceneReleaseContext):
"""On exiting a Scene Release macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Release Scene " + str(i)) logging.info("Release Scene %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/release/1", i) self.osc.send_message(ctx.device(), "/hog/playback/release/1", i)
def exitMacroGo(self, ctx: CommentMacroParser.MacroGoContext): def exitMacroGo(self, ctx: CommentMacroParser.MacroGoContext):
"""On exiting a Macro GO macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Go on Macro " + str(i)) logging.info("Go on Macro %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/go/2", i) self.osc.send_message(ctx.device(), "/hog/playback/go/2", i)
def exitMacroHalt(self, ctx: CommentMacroParser.MacroHaltContext): def exitMacroHalt(self, ctx: CommentMacroParser.MacroHaltContext):
"""On exiting a Macro Halt macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Pause Macro " + str(i)) logging.info("Pause Macro %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/halt/2", i) self.osc.send_message(ctx.device(), "/hog/playback/halt/2", i)
def exitMacroStop(self, ctx: CommentMacroParser.MacroStopContext): def exitMacroStop(self, ctx: CommentMacroParser.MacroStopContext):
"""On exiting a Macro Stop macro."""
for i in ctx.target().targets: for i in ctx.target().targets:
logging.info("Stop Macro " + str(i)) logging.info("Stop Macro %d", i)
self.osc.send_message(ctx.device(), "/hog/playback/release/2", i) self.osc.send_message(ctx.device(), "/hog/playback/release/2", i)
def exitPageNext(self, ctx: CommentMacroParser.PageNextContext): def exitPageNext(self, ctx: CommentMacroParser.PageNextContext):
"""On exiting a Next Page macro."""
logging.info("Next Page") logging.info("Next Page")
self.osc.button_press(ctx.device(), "/hog/hardware/nextpage") self.osc.button_press(ctx.device(), "/hog/hardware/nextpage")
def exitPagePrev(self, ctx: CommentMacroParser.PagePrevContext): def exitPagePrev(self, ctx: CommentMacroParser.PagePrevContext):
"""On exiting a Previous Page macro."""
logging.info("Prev Page") logging.info("Prev Page")
self.osc.button_press(ctx.device(), "/hog/hardware/backpage") self.osc.button_press(ctx.device(), "/hog/hardware/backpage")

View File

@ -1 +1,2 @@
"""Main BaconScript module init."""
__all__ = ["script", "hog4", "OscListener"] __all__ = ["script", "hog4", "OscListener"]

View File

@ -1 +1,2 @@
"""Init for comment macro antlr4."""
__all__ = ["CommentMacroLexer", "CommentMacroListener", "CommentMacroParser"] __all__ = ["CommentMacroLexer", "CommentMacroListener", "CommentMacroParser"]

View File

@ -1,43 +1,48 @@
"""hog4.py: Class methods for Hog 4 representation."""
import logging import logging
from pythonosc import osc_message_builder
from time import sleep from time import sleep
from pythonosc import osc_message_builder
class HogNet: class HogNet:
"""Class definition of a hognet participant."""
# button state constants # button state constants
buttonDOWN = 1 buttonDOWN = 1
buttonUP = 0 buttonUP = 0
def __init__(self, servers): def __init__(self, servers):
"""Init method."""
self.servers = servers self.servers = servers
def button_press(self, device, path, delay=0.05): def button_press(self, device, path, delay=0.05):
"""Button presses are a pair of up/down OSC."""
self.send_message(device, path, self.buttonDOWN) self.send_message(device, path, self.buttonDOWN)
sleep(delay) sleep(delay)
self.send_message(device, path, self.buttonUP) self.send_message(device, path, self.buttonUP)
# utility function to send simple messages with one argument
def send_message(self, device, path, arg): def send_message(self, device, path, arg):
"""Send a simple OSC message with one argument."""
msg = osc_message_builder.OscMessageBuilder(address=path) msg = osc_message_builder.OscMessageBuilder(address=path)
msg.add_arg(arg) msg.add_arg(arg)
self.send(device, msg.build()) self.send(device, msg.build())
# send python-osc messages
def send(self, device, msg): def send(self, device, msg):
"""Send python-osc messages."""
if device is None: if device is None:
osc = list(self.servers.values())[0] # first configured server osc = list(self.servers.values())[0] # first configured server
else: else:
if (device.nodeType().getText().lower() != 'h'): if device.nodeType().getText().lower() != 'h':
logging.error("ERROR: Only Hog type devices are supported.") logging.error("ERROR: Only Hog type devices are supported.")
return return
else: try:
try: osc = self.servers[device.number().value]
osc = self.servers[device.number().value] except KeyError:
except KeyError: logging.error("ERROR: Net# %d not found.",
logging.error("ERROR: Net# " + str(device.number().value) + device.number().value)
" not found.") return
return
try: try:
osc.send(msg) osc.send(msg)
except Exception as e: except OSError as exception:
logging.error(e) logging.error(exception)

View File

@ -1,5 +1,4 @@
"""comment.py: Hog 4 comment macro interpreter and OSC bridge. """comment.py: Hog 4 comment macro interpreter and OSC bridge."""
"""
import configparser import configparser
import logging import logging
@ -14,13 +13,15 @@ from .commentmacro.CommentMacroParser import CommentMacroParser
from .OscListener import OscCommentMacroListener from .OscListener import OscCommentMacroListener
# define an error listener that raises SyntaxError exceptions
class SyntaxErrorListener(ErrorListener): class SyntaxErrorListener(ErrorListener):
"""An error listener that raises SyntaxError exceptions."""
def syntaxError(self, recognizer, offendingSymbol, line, column, msg, e): def syntaxError(self, recognizer, offendingSymbol, line, column, msg, e):
raise SyntaxError("line "+str(line)+":"+str(column)+" "+msg) raise SyntaxError("line "+str(line)+":"+str(column)+" "+msg)
def loadConfig(file='server.cfg'): def load_config(file='server.cfg'):
"""Load an ini style configuration file."""
# empty server dictionary # empty server dictionary
servers = {} servers = {}
@ -34,25 +35,25 @@ def loadConfig(file='server.cfg'):
# move to config section # move to config section
server = config[name.strip().strip('\"')] server = config[name.strip().strip('\"')]
# read settings # read settings
ip = server.get("ip", "10.0.0.100") addr = server.get("ip", "10.0.0.100")
port = server.getint("port", 7001) port = server.getint("port", 7001)
net = server.getint("net", 1) net = server.getint("net", 1)
# osc clients are added to the dictionary with the net # as the key # osc clients are added to the dictionary with the net # as the key
logging.info("Adding Hog device at net# " + str(net)) logging.info("Adding Hog device at net# %d", str(net))
servers[net] = udp_client.SimpleUDPClient(ip, port) servers[net] = udp_client.SimpleUDPClient(addr, port)
except KeyError as e: except KeyError as exception:
print('Error configuring net#', net, e) print('Error configuring net#', net, exception)
continue continue
return servers return servers
# init reusable walker and listener # init reusable walker and listener
walker = ParseTreeWalker() WALKER = ParseTreeWalker()
listener = OscCommentMacroListener() LISTENER = OscCommentMacroListener()
# main handler for processing input
def comment(text): def comment(text):
"""Process comment macro input."""
# force upper case # force upper case
text = text.upper() text = text.upper()
# generate text stream # generate text stream
@ -69,7 +70,7 @@ def comment(text):
# get tree from parser # get tree from parser
tree = parser.prog() tree = parser.prog()
# walk the tree # walk the tree
walker.walk(listener, tree) WALKER.walk(LISTENER, tree)
except SyntaxError as e: except SyntaxError as exception:
logging.debug(e) # antlr internal listener prints the error logging.debug(exception) # antlr internal listener prints the error
# # log it to the debug logging anyway # # log it to the debug logging anyway

25
bs.py
View File

@ -1,21 +1,20 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""Interactive BaconScript shell. """Interactive BaconScript shell."""
"""
import logging import logging
import sys import sys
from bacon.script import comment, listener from bacon.script import comment, LISTENER
from bacon.script import loadConfig as loadBsConfig from bacon.script import load_config as loadBsConfig
from bacon.hog4 import HogNet from bacon.hog4 import HogNet
# setup logging # setup logging
logger = logging.getLogger() LOGGER = logging.getLogger()
logger.setLevel(logging.DEBUG) LOGGER.setLevel(logging.DEBUG)
listener.osc = HogNet(loadBsConfig('server.cfg')) LISTENER.osc = HogNet(loadBsConfig('server.cfg'))
# handle user input if run directly # handle user input if run directly
if __name__ == '__main__': if __name__ == '__main__':
@ -24,18 +23,16 @@ if __name__ == '__main__':
logging.debug("found macro at argv[1]") logging.debug("found macro at argv[1]")
comment(sys.argv[1]) comment(sys.argv[1])
else: else:
# for input history and line editing
import readline
# be an interactive shell # be an interactive shell
while True: while True:
# get user input # get user input
try: try:
text = input("comment# ") TEXT = input("comment# ")
except (KeyboardInterrupt, EOFError): except (KeyboardInterrupt, EOFError):
text = 'exit' TEXT = 'exit'
print(text) print(TEXT)
# catch exit keyword # catch exit keyword
if text.lower() == 'exit': if TEXT.lower() == 'exit':
break break
# exec user input # exec user input
comment(text) comment(TEXT)

View File

@ -1,6 +1,8 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""Setup.py: Module install script."""
from setuptools import setup, find_packages from setuptools import setup, find_packages
setup(name='baconscript', setup(name='baconscript',
@ -12,7 +14,7 @@ setup(name='baconscript',
license='MIT', license='MIT',
packages=find_packages(), packages=find_packages(),
install_requires=[ install_requires=[
'antlr4-python3-runtime', 'antlr4-python3-runtime',
'python-osc', 'python-osc',
], ],
zip_safe=False) zip_safe=False)