forked from henrik242/OnAir
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
1 changed file
with
109 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,109 @@ | ||
#!/usr/bin/env python3 | ||
|
||
import os | ||
import platform | ||
import re | ||
import threading | ||
import time | ||
import webbrowser | ||
import rumps | ||
|
||
macos_version = int(platform.mac_ver()[0][:2]) | ||
|
||
class OnAir(object): | ||
def __init__(self): | ||
self.app = rumps.App("OnAir", "⚪") | ||
self.menuToggle = rumps.MenuItem("Turn on", callback=self.on_air) | ||
self.app.menu.add(self.menuToggle) | ||
self.app.menu.add(rumps.MenuItem("About OnAir…", callback=self.open_onair_url)) | ||
self.menubar_blinker_active = False | ||
self.camera_state_updater_active = True | ||
|
||
def run(self): | ||
threading.Thread(target=self.camera_state_updater, daemon=True).start() | ||
self.app.run() | ||
|
||
def log(self, msg): | ||
print("%s" % msg) | ||
|
||
def camera_state_updater(self): # Corrected the indentation here | ||
self.log("camera_state_updater()") | ||
predicate = 'subsystem == "com.apple.UVCExtension" and composedMessage contains "Post PowerLog"' | ||
extraopts = "" | ||
searchexpr = "guid:(.+)]" | ||
onitem = "Start" | ||
offitem = "Stop" | ||
if macos_version == 12: | ||
predicate = 'eventMessage contains "Post event kCameraStream"' | ||
extraopts = "--style ndjson" | ||
searchexpr = 'VDCAssistant_Device_GUID\\\\" = \\\\"(.+)\\\\";' | ||
onitem = "= On;" | ||
offitem = "= Off;" | ||
if macos_version >= 13: | ||
predicate = 'eventMessage contains "Cameras changed to"' | ||
extraopts = "--style ndjson" | ||
searchexpr = '"Cameras changed to (\[.*\])",' | ||
onitem = "to [ControlCenter" | ||
offitem = "to []" | ||
log_stream = os.popen("""/usr/bin/log stream %s --predicate '%s'""" % (extraopts, predicate), "r") | ||
cameras = dict() | ||
while self.camera_state_updater_active: | ||
item = log_stream.readline() | ||
self.log("reading '%s'" % item.strip()) | ||
if item == "": | ||
self.log("log stream died") | ||
break | ||
match = re.search(searchexpr, item) | ||
if match is not None: | ||
device = match.group(1) if macos_version < 13 else "dummy" | ||
if onitem in item: | ||
cameras[device] = True | ||
elif offitem in item: | ||
cameras[device] = False | ||
else: | ||
self.log("Unknown activity: %s" % item) | ||
self.log(cameras) | ||
if any(cameras.values()): | ||
self.log("Camera %s is on" % device) | ||
self.on_air() | ||
else: | ||
self.log("Camera %s is off" % device) | ||
self.off_air() | ||
self.log("camera_state_updater() done") | ||
|
||
@staticmethod | ||
def open_onair_url(callback_sender=None): | ||
webbrowser.open_new_tab("https://github.com/henrik242/OnAir") | ||
|
||
def on_air(self, callback_sender=None): | ||
self.log("on_air()") | ||
os.system("""osascript -e 'tell application "Keyboard Maestro Engine" to do script "ON AIR MACRO UUID HERE"'""") | ||
# os.system("""osascript -e 'tell application "Keyboard Maestro Engine" to do script "ON AIR MACRO NAME HERE"'""") | ||
self.menubar_blinker_active = True | ||
threading.Thread(target=self.menubar_blinker, daemon=True).start() | ||
self.menuToggle.title = "Turn off" | ||
self.menuToggle.set_callback(callback=self.off_air) | ||
self.log("on_air() done") | ||
|
||
def off_air(self, callback_sender=None): | ||
self.log("off_air()") | ||
os.system("""osascript -e 'tell application "Keyboard Maestro Engine" to do script "OFF AIR MACRO UUID HERE"'""") | ||
# os.system("""osascript -e 'tell application "Keyboard Maestro Engine" to do script "OFF AIR MACRO NAME HERE"'""") | ||
self.menubar_blinker_active = False | ||
self.menuToggle.title = "Turn on" | ||
self.menuToggle.set_callback(callback=self.on_air) | ||
self.log("off_air() done") | ||
|
||
def menubar_blinker(self): | ||
self.log("menubar_blinker()") | ||
green = True | ||
while self.menubar_blinker_active: | ||
self.app.title = "🟢" if green else "⚪️" | ||
time.sleep(1) | ||
green = not green | ||
self.app.title = "⚪" | ||
self.log("menubar_blinker() done") | ||
|
||
if __name__ == "__main__": | ||
app = OnAir() | ||
app.run() |