"""hog4.py: Class methods for Hog 4 representation.""" import logging from time import sleep from typing import Any, Dict, Union from pythonosc import udp_client, osc_message_builder, osc_bundle, osc_message from antlr4 import ParserRuleContext __all__ = [ "HogNet", ] log = logging.getLogger(__name__) class HogNet: """Class definition of a hognet participant.""" # button state constants buttonDOWN = 1 buttonUP = 0 buttonMap: Dict[str, str] = { "backspace": "/hog/hardware/backspace", "beam": "/hog/hardware/beam", "blind": "/hog/hardware/blind", "clear": "/hog/hardware/clear", "colour": "/hog/hardware/colour", "copy": "/hog/hardware/copy", "cue": "/hog/hardware/cue", "delete": "/hog/hardware/delete", "down": "/hog/hardware/down", "effect": "/hog/hardware/effect", "enter": "/hog/hardware/enter", "fan": "/hog/hardware/fan", "fixture": "/hog/hardware/fixture", "goto": "/hog/hardware/goto", "group": "/hog/hardware/group", "highlight": "/hog/hardware/highlight", "intensity": "/hog/hardware/intensity", "left": "/hog/hardware/left", "list": "/hog/hardware/list", "live": "/hog/hardware/live", "macro": "/hog/hardware/macro", "merge": "/hog/hardware/merge", "move": "/hog/hardware/move", "open": "/hog/hardware/open", "page": "/hog/hardware/page", "pig": "/hog/hardware/pig", "position": "/hog/hardware/position", "record": "/hog/hardware/record", "right": "/hog/hardware/right", "scene": "/hog/hardware/scene", "set": "/hog/hardware/set", "setup": "/hog/hardware/setup", "time": "/hog/hardware/time", "up": "/hog/hardware/up", "update": "/hog/hardware/update", ".": "/hog/hardware/period", "@": "/hog/hardware/at", "-": "/hog/hardware/minus", "+": "/hog/hardware/plus", "/": "/hog/hardware/slash", "0": "/hog/hardware/zero", "1": "/hog/hardware/one", "2": "/hog/hardware/two", "3": "/hog/hardware/three", "4": "/hog/hardware/four", "5": "/hog/hardware/five", "6": "/hog/hardware/six", "7": "/hog/hardware/seven", "8": "/hog/hardware/eight", "9": "/hog/hardware/nine" } def __init__(self, servers: Dict[int, object]) -> None: """Init method.""" self.servers = servers def number_entry(self, device: ParserRuleContext, number: str) -> None: """Press digit buttons to make a number.""" for digit in number: try: self.button_press(device, self.buttonMap[digit]) except KeyError: continue 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) sleep(delay) 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: ParserRuleContext, msg: Union[osc_message.OscMessage, osc_bundle.OscBundle]) -> None: """Send python-osc messages.""" if device is None: # first configured server osc: udp_client = list(self.servers.values())[0] else: if device.nodeType().getText().lower() != 'h': log.error("ERROR: Only Hog type devices are supported.") return try: osc = self.servers[device.number().value] except KeyError: log.error("ERROR: Net# %d not found.", device.number().value) return try: osc.send(msg) except OSError as exception: log.error(exception)