webXMPP

Check-in [dd70b2d995]
Login

Many hyperlinks are disabled.
Use anonymous login to enable hyperlinks.

Overview
Comment:By using a system app name, I get blinking light, notification popup, sound, and longevity on the notification list "for free".
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | master | trunk
Files: files | file ages | folders
SHA3-256:dd70b2d995805ace0385c6431a6f2120a658b0899d34b2f302ed8c47b4d4dd3b
User & Date: vandys 2019-08-24 22:28:34
Context
2019-08-25
00:46
Nope, doesn't play well. Go back to doing our own notify/dbus/led stuff. Leaf check-in: 10182dc9d8 user: vandys tags: master, trunk
2019-08-24
22:28
By using a system app name, I get blinking light, notification popup, sound, and longevity on the notification list "for free". check-in: dd70b2d995 user: vandys tags: master, trunk
2019-05-21
15:33
Support gif, too check-in: 85559f8e8c user: web tags: master, trunk
Changes
Hide Diffs Unified Diffs Ignore Whitespace Patch

Changes to tools/notified.py.

46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
..
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
...
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
#
import sys, json, time, os, traceback
w = sys.stdout.write
import notify2
import pong
import pdb

# For DBus sniffing
import dbus, dbus.exceptions, dbus.mainloop.glib
import threading
from gi.repository import GLib

# Are LEDs available for notification indication?
LED = "/sys/class/leds/red"
blinking = leds = False

# Initial condition, no events ever seen
gen = 0

# Timestamp of our start, for suppressing initial message
#  notifications
Started = time.time()

................................................................................
    def __init__(self, d):
        for k,v in d.items():
            setattr(self, str(k), v)

# Configuration, a DictOb
cfg = None

# DBus monitoring.  We need to know if the device is on/active, since
#  we only want to blink the LED if it's off (and only until it
#  goes back on)
is_locked = True
def dbus_watcher(*args, **kwargs):
    global blinking, is_locked

    dd = args[1]
    if "IsActive" not in dd:
        return
    is_locked = bool(dd["IsActive"])
    if blinking and (not is_locked):
        unblink()

def setup_dbus():
    global leds

    # We only need this to turn the LED indication back off, so
    #  skip it if there's no LED's in the first place
    if not leds:
        return

    # We have to give it a thread for watching the bus
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    l = GLib.MainLoop()
    t = threading.Thread(target=l.run)
    t.start()

    # Now register a callback to see "desktop" changes
    sess = dbus.SessionBus()
    sess.add_signal_receiver(dbus_watcher, "PropertiesChanged",
        dbus_interface="org.freedesktop.DBus.Properties")

# LED device manipulation
def set_led(base, val):
    f = open(LED + "/" + base, "w")
    f.write(val)
    f.close()
def get_led(base):
    f = open(LED + "/" + base, "r")
    res = f.read()
    f.close()
    return res

# Set LED mode, if we can
def setup_leds():
    global leds

    try:
        set_led("on_off_ms", "500 2500")
        brt = get_led("max_brightness")
        set_led("brightness", brt)
        set_led("rgb_start", "0")
        leds = True
    except:
        pass

# Start/stop blinking
def blink():
    global blinking
    setup_leds()
    set_led("rgb_start", "1")
    blinking = True
def unblink():
    global blinking
    set_led("rgb_start", "0")
    blinking = False

# Pull config out of JSON save format
def load_cfg():
    global cfg

    # Get our configuration
    f = open(os.getenv("HOME") + "/.config/notify.json", "r")
    d = json.loads(f.read())
    f.close()
    cfg = DictOb(d)

# New notifications contained in this packet
def notify(pak):
    global gen, w, leds, blinking, is_locked

    inner = pak.inner
    w("Notification: gen %d -> %d\n" % (gen, inner["gen"]))
    anyrx = False
    for tup in inner["msgs"]:
        w("  %s\n" % (tup,))
        lt = len(tup)
................................................................................
        log("Notification: " + n1 + ((", " + n2) if n2 else ""))
        try:
            n.show()
        except:
            traceback.format_exc()
            log("Notification not displayed")

    # Only shadows of our own typing on another device, so stay
    #  quiet
    if not anyrx:
        return

    # Worry about the LEDs?
    if leds and (not blinking) and is_locked:
        blink()

    # Play sound
    os.system("paplay --volume=40000 %s" % (cfg.sound,))

# Endless execution, notification client
def run():
    global cfg, gen

    # Get a wrapper for our pong network connection
    conn = pong.Client(cfg.server, cfg.port, cfg.user, cfg.password)

    # Prep LEDs for use if possible
    setup_leds()

    # Watch DBus to turn LED's off when screen unlocks
    setup_dbus()

    # Prepare for notifications
    notify2.init("Messsaging")

    # Our notification server
    dest = (cfg.server, cfg.port)

    # We need a send-receive-send pattern as a minimum to become
    #  an assured/streaming UDP "connection".  This loop starts
    #  a new socket, does a param get/got, and then a timed







<
<
<
<
<
<
<
<
<







 







<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<












|







 







<
<
<
<
<
<
<
<
<
<
<
<







<
<
<
<
<
<

|







46
47
48
49
50
51
52









53
54
55
56
57
58
59
..
67
68
69
70
71
72
73




































































74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
...
122
123
124
125
126
127
128












129
130
131
132
133
134
135






136
137
138
139
140
141
142
143
144
#
import sys, json, time, os, traceback
w = sys.stdout.write
import notify2
import pong
import pdb










# Initial condition, no events ever seen
gen = 0

# Timestamp of our start, for suppressing initial message
#  notifications
Started = time.time()

................................................................................
    def __init__(self, d):
        for k,v in d.items():
            setattr(self, str(k), v)

# Configuration, a DictOb
cfg = None





































































# Pull config out of JSON save format
def load_cfg():
    global cfg

    # Get our configuration
    f = open(os.getenv("HOME") + "/.config/notify.json", "r")
    d = json.loads(f.read())
    f.close()
    cfg = DictOb(d)

# New notifications contained in this packet
def notify(pak):
    global gen, w

    inner = pak.inner
    w("Notification: gen %d -> %d\n" % (gen, inner["gen"]))
    anyrx = False
    for tup in inner["msgs"]:
        w("  %s\n" % (tup,))
        lt = len(tup)
................................................................................
        log("Notification: " + n1 + ((", " + n2) if n2 else ""))
        try:
            n.show()
        except:
            traceback.format_exc()
            log("Notification not displayed")













# Endless execution, notification client
def run():
    global cfg, gen

    # Get a wrapper for our pong network connection
    conn = pong.Client(cfg.server, cfg.port, cfg.user, cfg.password)







    # Prepare for notifications
    notify2.init("Telephony Service Indicator")

    # Our notification server
    dest = (cfg.server, cfg.port)

    # We need a send-receive-send pattern as a minimum to become
    #  an assured/streaming UDP "connection".  This loop starts
    #  a new socket, does a param get/got, and then a timed