additional pep8 typing
This commit is contained in:
parent
c6cba182d3
commit
5bf36be3be
|
@ -2,7 +2,9 @@
|
|||
|
||||
import logging
|
||||
from time import sleep
|
||||
from pythonosc import osc_message_builder
|
||||
from typing import Any, Dict, Union
|
||||
from pythonosc import udp_client, osc_message_builder, osc_bundle, osc_message
|
||||
from antlr4 import ParserRuleContext
|
||||
|
||||
|
||||
class HogNet:
|
||||
|
@ -12,26 +14,30 @@ class HogNet:
|
|||
buttonDOWN = 1
|
||||
buttonUP = 0
|
||||
|
||||
def __init__(self, servers):
|
||||
def __init__(self, servers: Dict[int, object]) -> None:
|
||||
"""Init method."""
|
||||
self.servers = servers
|
||||
|
||||
def button_press(self, device, path, delay=0.05):
|
||||
def button_press(self, device: ParserRuleContext,
|
||||
path: str, delay: float = 0.05) -> None:
|
||||
"""Button presses are a pair of up/down OSC."""
|
||||
self.send_message(device, path, self.buttonDOWN)
|
||||
sleep(delay)
|
||||
self.send_message(device, path, self.buttonUP)
|
||||
|
||||
def send_message(self, device, path, arg):
|
||||
def send_message(self, device: ParserRuleContext,
|
||||
path: str, arg: Any) -> None:
|
||||
"""Send a simple OSC message with one argument."""
|
||||
msg = osc_message_builder.OscMessageBuilder(address=path)
|
||||
msg.add_arg(arg)
|
||||
self.send(device, msg.build())
|
||||
|
||||
def send(self, device, msg):
|
||||
def send(self, device: ParserRuleContext,
|
||||
msg: Union[osc_message.OscMessage, osc_bundle.OscBundle]) -> None:
|
||||
"""Send python-osc messages."""
|
||||
if device is None:
|
||||
osc = list(self.servers.values())[0] # first configured server
|
||||
# first configured server
|
||||
osc: udp_client = list(self.servers.values())[0]
|
||||
else:
|
||||
if device.nodeType().getText().lower() != 'h':
|
||||
logging.error("ERROR: Only Hog type devices are supported.")
|
||||
|
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
import configparser
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
from antlr4 import CommonTokenStream, InputStream, ParseTreeWalker
|
||||
from antlr4.error.ErrorListener import ErrorListener
|
||||
|
@ -16,13 +17,13 @@ class SyntaxErrorListener(ErrorListener):
|
|||
"""An error listener that raises SyntaxError exceptions."""
|
||||
|
||||
def syntaxError(self, recognizer, offendingSymbol, line, column, msg, e):
|
||||
raise SyntaxError("line "+str(line)+":"+str(column)+" "+msg)
|
||||
raise SyntaxError("line {}:{} {}".format(line, column, msg))
|
||||
|
||||
|
||||
def load_config(file='server.cfg'):
|
||||
def load_config(file: str = 'server.cfg') -> Dict[int, object]:
|
||||
"""Load an ini style configuration file."""
|
||||
# empty server dictionary
|
||||
servers = {}
|
||||
servers: Dict[int, object] = {}
|
||||
|
||||
# open config file
|
||||
config = configparser.ConfigParser(allow_no_value=True)
|
||||
|
@ -34,9 +35,9 @@ def load_config(file='server.cfg'):
|
|||
# move to config section
|
||||
server = config[name.strip().strip('\"')]
|
||||
# read settings
|
||||
addr = server.get("ip", "10.0.0.100")
|
||||
port = server.getint("port", 7001)
|
||||
net = server.getint("net", 1)
|
||||
addr = server.get("ip", fallback="127.0.0.1")
|
||||
port = server.getint("port", fallback=7001)
|
||||
net = server.getint("net", fallback=1)
|
||||
# osc clients are added to the dictionary with the net # as the key
|
||||
logging.info("Adding Hog device at net# %d", str(net))
|
||||
servers[net] = udp_client.SimpleUDPClient(addr, port)
|
||||
|
@ -47,11 +48,11 @@ def load_config(file='server.cfg'):
|
|||
|
||||
|
||||
# init reusable walker and listener
|
||||
WALKER = ParseTreeWalker()
|
||||
LISTENER = OscCommentMacroListener()
|
||||
WALKER: ParseTreeWalker = ParseTreeWalker()
|
||||
LISTENER: OscCommentMacroListener = OscCommentMacroListener()
|
||||
|
||||
|
||||
def comment(text):
|
||||
def comment(text: str) -> None:
|
||||
"""Process comment macro input."""
|
||||
# force upper case
|
||||
text = text.upper()
|
||||
|
|
Loading…
Reference in New Issue