blc2 - Python library and frontend for running theatrical lighting and SFX. Written for the English 2041F fall 2019 theatre production of "The Cenci".

git clone https://benconnors.ca/git-repos/blc2

Log | Files | Refs

ola.py (1637B) - raw

      1 import array
      2 
      3 import threading
      4 import time
      5 
      6 from ola.OlaClient import OlaClient, OLADNotRunningException
      7 
      8 class OLAOutput:
      9     name = "OLA"
     10 
     11     def set_values(self, values):
     12         send = set()
     13         for c, v in values.items():
     14             univ, addr = c.address
     15             if univ not in self.universes:
     16                 self.universes[univ] = array.array('B', (0 for i in range(512)))
     17             uni = self.universes[univ]
     18             if uni[addr] != v:
     19                 uni[addr] = v
     20                 send.add(univ)
     21         with self._clock:
     22             if self.client is not None:
     23                 for univ in send:
     24                     self.client.SendDmx(univ, self.universes[univ])
     25 
     26     def _setup_client(self):
     27         while True:
     28             try:
     29                 self.client = OlaClient()
     30             except OLADNotRunningException:
     31                 time.sleep(1)
     32             else:
     33                 break
     34 
     35     @property
     36     def ok(self):
     37         with self._clock:
     38             return self.client is not None
     39 
     40     @property
     41     def status(self):
     42         with self._clock:
     43             if self.client is None:
     44                 return "Unable to connect to olad"
     45             else:
     46                 return "Everything's good"
     47 
     48     def __init__(self):
     49         self.client_thread = None
     50         try:
     51             self.client = OlaClient()
     52         except OLADNotRunningException:
     53             self.client = None
     54             self.client_thread = threading.Thread(target=self._setup_client)
     55             self.client_thread.start()
     56 
     57         self.universes = {1: array.array('B', (0 for i in range(512)))}
     58 
     59         self._clock = threading.RLock()