#!/usr/bin/env python3 cmd_desc = "Run a scripted performance using Arduino-controlled stepper motors and DMX-controlled lighting." import argparse import time import math import logging # import the calendar policy from HuntLibrary import current_time_properties # import the StepperWinch interface import steppers # import the DMXUSBPro interface import dmxusbpro # import the specific performance script from demo_script import SuitcasePerformance # initialize logging for this module log = logging.getLogger('show') #================================================================ class Show: def __init__(self, args): # configure event timing self.dt_ns = 50*1000*1000 # 20 Hz in nanoseconds per cycle # self.dt_ns *= 10 # tmp slow the rate # Open the DMX controller serial port log.info("Opening DMX serial port.") self.dmx = dmxusbpro.DMXUSBPro(port=args.dmx, verbose=args.verbose, debug=args.debug) self.dmx.open_serial_port() # Current light state used for smoothing changes. self.spot_intensity = [0.0, 0.0] # Open the Arduino serial port log.info("Opening Arduino serial port.") self.steppers = steppers.StepperWinchClient(port=args.arduino, verbose=args.verbose, debug=args.debug) log.info("Waiting for Arduino wakeup.") self.steppers.wait_for_wakeup() self.steppers.set_freq_damping(0.5, 1.0) # Create the performance controller(s) self.controller = SuitcasePerformance() # Test the current time policy = current_time_properties() log.info("Current time policy: %s", policy) # configure party mode which ignores calendar self.party_mode = args.party if self.party_mode: log.info("Party mode enabled.") return def close(self): # Issue a command to turn off the drivers, then shut down the connections. log.info("Closing serial ports.") self.steppers.motor_enable(False) self.steppers.close() self.dmx.close_serial_port() #--------------------------------------------------------------- # Action primitives for current setup. def set_spotlight(self, name, level): # This is the right light as viewed from outside, i.e. stage left. if name == 'right': # convert a unit value (0.0,1.0) to 8-bit binary (0, 255) value = min(max(int(level*255), 0), 255) self.dmx.universe[0] = value # This is the left light as viewed from outside, i.e. stage right. elif name == 'left': value = min(max(int(level*255), 0), 255) self.dmx.universe[1] = value # This sets the available lights from an array, list, tuple, or sequence elif name == 'all': values = [min(max(int(val*255), 0), 255) for val in level] self.dmx.universe[0:2] = values # Update the hardware self.dmx.send_universe() def send_angles(self, angles): steps = [int(angle * (800/(2*math.pi))) for angle in angles] self.steppers.send_move(steps) return #--------------------------------------------------------------- # Event loop for operating the performance def run(self): start_t = time.monotonic_ns() next_cycle_t = start_t + self.dt_ns while True: # wait for the next cycle timepoitn, keeping the long # term rate stable even if the short term timing jitters now_ns = time.monotonic_ns() delay = max(next_cycle_t - now_ns, 0) if (delay > 0): time.sleep(delay * 1e-9) next_cycle_t += self.dt_ns now_ns = time.monotonic_ns() now_seconds = (now_ns - start_t)*1e-9 log.debug("Time is %f" %(now_seconds)) # update the performance self.controller.poll(now_seconds) # keep the stepper input buffer empty self.steppers.poll_status() # check the time policy policy = current_time_properties() if self.party_mode or policy['is_show_time']: # output motion commands to the stepper hardware self.send_angles(self.controller.target) # apply first-order smoothing to the lighting commands self.spot_intensity[0] += 0.2 * (self.controller.lights['left'] - self.spot_intensity[0]) self.spot_intensity[1] += 0.2 * (self.controller.lights['right'] - self.spot_intensity[1]) self.set_spotlight('all', self.spot_intensity) else: # keep sending commands for a quiescent state self.send_angles([0.0, 0.0, 0.0, 0.0]) self.spot_intensity[0] *= 0.8 self.spot_intensity[1] *= 0.8 self.set_spotlight('all', self.spot_intensity) #================================================================ # The following section is run when this is loaded as a script. if __name__ == "__main__": # set up logging log_format= '%(asctime)s:%(levelname)s:%(name)s: %(message)s' log_datefmt="%Y-%m-%d %H:%M:%S" log_stream = open('show.log', 'a') logging.basicConfig(stream=log_stream, level=logging.INFO, format=log_format, datefmt=log_datefmt) log.info("Starting run-show.py") # Initialize the command parser. parser = argparse.ArgumentParser(description = cmd_desc) parser.add_argument( '--dmx', default='/dev/ttyUSB0', help='DMX serial port device (default: %(default)s).') parser.add_argument( '--arduino', default='/dev/ttyACM0', help='Arduino serial port device (default is %(default)s.)') parser.add_argument( '--party', action='store_true', help='Enable party mode which overrides calendar policy.') parser.add_argument( '--debug', action='store_true', help='Enable debugging output to log file.' ) parser.add_argument( '--verbose', action='store_true', help='Enable even more detailed logging output.' ) # Parse the command line, returning a Namespace. args = parser.parse_args() # Enable debug messages in logs on request. if args.debug: logging.getLogger().setLevel(logging.DEBUG) # Create the show controller. show = Show(args) # Begin the performance. This may be safely interrupted by the user pressing Control-C. try: show.run() except KeyboardInterrupt: log.info("User interrupted performance.") print("User interrupted performance, shutting down.") show.close()