mirror/vincentbernat.i3wm-configuration
1
0
Fork 0
mirror of https://github.com/vincentbernat/i3wm-configuration.git synced 2025-07-06 00:04:20 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

i3-companion: cosmetic fixes

Browse source 
Also update immediately if active connection change.
This commit is contained in:
Vincent Bernat 2021-07-12 07:57:50 +02:00
parent 7df5011824
commit f0a7eca578
1 changed files with 39 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

61
bin/i3-companion
View file

@ -401,16 +401,15 @@ async def output_update(i3, event):
) )
async def network_manager_notifications(i3, event, path, state, reason): async def network_manager_notifications(i3, event, path, state, reason):
"""Display notifications related to Network Manager state.""" """Display notifications related to Network Manager state."""
ofnm = "org.freedesktop.NetworkManager"
logger.debug(f"from {path} state: {state}, reason: {reason}") logger.debug(f"from {path} state: {state}, reason: {reason}")
if state not in {NM_ACTIVE_CONNECTION_STATE_ACTIVATED}: if state not in {NM_ACTIVE_CONNECTION_STATE_ACTIVATED}:
# We cannot get proper state unless the connection is # We cannot get proper state unless the connection is
# activated, unless we maintain state. # activated, unless we maintain state.
return return
peer = i3.system_bus["org.freedesktop.NetworkManager"][path] peer = i3.system_bus[ofnm][path]
try: try:
nmca = await peer.get_async_interface( nmca = await peer.get_async_interface(f"{ofnm}.Connection.Active")
"org.freedesktop.NetworkManager.Connection.Active"
)
except dbussy.DBusError: except dbussy.DBusError:
logger.info(f"interface {path} has vanished") logger.info(f"interface {path} has vanished")
return return
@ -453,13 +452,20 @@ async def network_manager_notifications(i3, event, path, state, reason):
) )
async def network_manager_status(i3, event, *args): async def network_manager_status(i3, event, *args):
"""Compute network manager status.""" """Compute network manager status."""
if isinstance(event, DBusSignal) and event.interface == "org.freedesktop.NetworkManager.AccessPoint": ofnm = "org.freedesktop.NetworkManager"
# Skip AP updates when they are not about the current AccessPoint
# or not about Strength. Most of the time, updates are about that.
if (
isinstance(event, DBusSignal)
and event.interface == f"{ofnm}.AccessPoint"
):
path, props = args path, props = args
if getattr(network_manager_status, "active_ap", None) != path: if getattr(network_manager_status, "active_ap", None) != path:
return return
if not "Strength" in props: if not "Strength" in props:
return return
# Dampen updates
running = getattr(network_manager_status, "running", None) running = getattr(network_manager_status, "running", None)
if running is not None: if running is not None:
running.cancel() running.cancel()
@ -484,15 +490,13 @@ async def network_manager_status(i3, event, *args):
status = [] status = []
# Build status from devices # Build status from devices
bus = i3.system_bus["org.freedesktop.NetworkManager"] bus = i3.system_bus[ofnm]
nm = await bus["/org/freedesktop/NetworkManager"].get_async_interface( nm = await bus["/org/freedesktop/NetworkManager"].get_async_interface(
"org.freedesktop.NetworkManager" ofnm
) )
devices = await nm.AllDevices devices = await nm.AllDevices
for device in devices: for device in devices:
nmd = await bus[device].get_async_interface( nmd = await bus[device].get_async_interface(f"{ofnm}.Device")
"org.freedesktop.NetworkManager.Device"
)
kind = await nmd.DeviceType kind = await nmd.DeviceType
state = await nmd.State state = await nmd.State
if state == NM_DEVICE_STATE_UNMANAGED: if state == NM_DEVICE_STATE_UNMANAGED:
@ -502,16 +506,14 @@ async def network_manager_status(i3, event, *args):
status.append("") status.append("")
continue continue
nmw = await bus[device].get_async_interface( nmw = await bus[device].get_async_interface(
"org.freedesktop.NetworkManager.Device.Wireless" f"{ofnm}.Device.Wireless"
) )
ap = await nmw.ActiveAccessPoint ap = await nmw.ActiveAccessPoint
if not ap: if not ap:
status.append("") status.append("")
continue continue
network_manager_status.active_ap = ap network_manager_status.active_ap = ap
nmap = await bus[ap].get_async_interface( nmap = await bus[ap].get_async_interface(f"{ofnm}.AccessPoint")
"org.freedesktop.NetworkManager.AccessPoint"
)
name = await nmap.Ssid name = await nmap.Ssid
strength = int(await nmap.Strength) strength = int(await nmap.Strength)
status.append(""[strength // 34]) status.append(""[strength // 34])
@ -526,13 +528,13 @@ async def network_manager_status(i3, event, *args):
connections = await nm.ActiveConnections connections = await nm.ActiveConnections
for conn in connections: for conn in connections:
nma = await bus[conn].get_async_interface( nma = await bus[conn].get_async_interface(
"org.freedesktop.NetworkManager.Connection.Active" f"{ofnm}.Connection.Active"
) )
vpn = await nma.Vpn vpn = await nma.Vpn
if vpn: if vpn:
state = await nma.State state = await nma.State
if state == NM_ACTIVE_CONNECTION_STATE_ACTIVATED: if state == NM_ACTIVE_CONNECTION_STATE_ACTIVATED:
status.append("︁") status.append("")
status.append(await nma.Id) status.append(await nma.Id)
# Final status line # Final status line
@ -544,25 +546,40 @@ async def network_manager_status(i3, event, *args):
network_manager_status.last = status network_manager_status.last = status
# Update cache file (for when polybar restarts) # Update cache file (for when polybar restarts)
with open(f"{os.getenv('XDG_RUNTIME_DIR')}/i3/network.txt", "w") as out: with open(
f"{os.getenv('XDG_RUNTIME_DIR')}/i3/network.txt", "w"
) as out:
out.write(status) out.write(status)
# Send it to module/network # Send it to polybar's module/network
for name in glob.glob("/tmp/polybar_mqueue.*"): for name in glob.glob("/tmp/polybar_mqueue.*"):
try: try:
with open(name, "w") as out: with open(name, "w") as out:
# Switch to non-blocking mode. If process on
# the other side is dead, we get an ENXIO.
# Buffer should ensure we don't block
# otherwise.
fd = out.fileno() fd = out.fileno()
old_flags = fcntl.fcntl(fd, fcntl.F_GETFL) old_flags = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, old_flags | os.O_NONBLOCK) fcntl.fcntl(
fd, fcntl.F_SETFL, old_flags | os.O_NONBLOCK
)
cmd = f"action:#network.send.{status}" cmd = f"action:#network.send.{status}"
out.write(cmd) out.write(cmd)
except OSError as e: except OSError as e:
if e.errno == errno.ENXIO: if e.errno == errno.ENXIO:
pass pass
network_manager_status.running = asyncio.create_task( if (
network_manager_status_now() isinstance(event, DBusSignal)
) and event.interface == f"{ofnm}.Connection.Active"
):
await network_manager_status_now(0)
else:
network_manager_status.running = asyncio.create_task(
network_manager_status_now()
)
async def main(options): async def main(options):