Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use getwindowlist method in pyatom
Best Python code snippet using pyatom_python
controllers.py
Source:controllers.py 
1# coding=utf-82"""3 Copyright (c) 2015, NetIDE Consortium (Create-Net (CN), Telefonica Investigacion Y Desarrollo SA (TID), Fujitsu4 Technology Solutions GmbH (FTS), Thales Communications & Security SAS (THALES), Fundacion Imdea Networks (IMDEA),5 Universitaet Paderborn (UPB), Intel Research & Innovation Ireland Ltd (IRIIL), Fraunhofer-Institut für6 Produktionstechnologie (IPT), Telcaria Ideas SL (TELCA) )7 All rights reserved. This program and the accompanying materials8 are made available under the terms of the Eclipse Public License v1.09 which accompanies this distribution, and is available at10 http://www.eclipse.org/legal/epl-v10.html11 Authors:12 Gregor Best, gbe@mail.upb.de13"""14import json15import logging16import os17import subprocess18import sys19import time20import uuid21import yaml22from functools import reduce23from subprocess import call24from loader import util25class Base(object):26 applications = []27 def __init__(self, dataroot):28 self.dataroot = dataroot29 def makelogdir(self, myid):30 d = os.path.join(self.dataroot, "logs", myid)31 os.makedirs(d, exist_ok=True)32 return d33 @classmethod34 def attachTmux(cls):35 sessionExists = call(["tmux", "has-session", "-t", "NetIDE"])36 if [ sessionExists != 0 ]:37 call("tmux attach -t NetIDE", shell=True)38 else:39 print("Tmux Session NetIDE was not found!")40 @classmethod41 def getControllerName(cls):42 "Returns the name of the controller"43 return None44 @classmethod45 def version(cls):46 "Returns either the version of the controller as a string or None if the controller is not installed"47 return None48 def start(self):49 "Starts the controller and all its applications. Returns a list of dictionaries of the following form:"50 "{ \"id\": process-uuid, \"pid\": process-id }"51 raise NotImplementedError()52class RyuShim(Base):53 def __init__(self, ofport="6633"):54 self.port = ofport55 def start(self):56 base = ["ryu-manager"]57 y = yaml.load(open("commands.yml"))58 ryuCommands = y['ryu_shim']59 v = os.environ.copy()60 v.update(ryuCommands["variables"])61 sessionExists = call(["tmux", "has-session", "-t", "NetIDE"])62 if [ sessionExists != 0 ]:63 call(["tmux", "new-session", "-d", "-s", "NetIDE"], env=v)64 time.sleep(1)65 cmd = util.tmux_line(ryuCommands)66 list = util.getWindowList()67 if "RyuShim" not in list:68 #6633 default for port listening69 newCmd = "bash -c \'cd ~/netide/Engine/ryu-shim/ && " + cmd + " --ofp-tcp-listen-port=" + self.port + " ryu-shim.py\'"70 call(['tmux', 'new-window', '-n', "RyuShim", '-t', 'NetIDE', newCmd])71 else:72 print("Ryu Shim already running")73class Ryu(Base):74 appNames = []75 ryubackendport = 773376 @classmethod77 def getControllerName(cls):78 return "ryu"79 @classmethod80 def version(cls):81 try:82 v = subprocess.check_output(["ryu", "--version"], stderr=subprocess.STDOUT).decode("utf-8")83 return v.strip().split(" ", 1)[1]84 except subprocess.CalledProcessError:85 return None86 except FileNotFoundError:87 return None88 def start(self):89 base = ["ryu-manager"]90 y = yaml.load(open("commands.yml"))91 ryuCommands = y['ryu']92 v = os.environ.copy()93 v.update(ryuCommands["variables"])94 sessionExists = call(["tmux", "has-session", "-t", "NetIDE"])95 if [ sessionExists != 0 ]:96 call(["tmux", "new-session", "-d", "-s", "NetIDE"], env=v)97 time.sleep(1)98 cmd = util.tmux_line(ryuCommands)99 for a in self.applications:100 self.appNames.append(a.appName)101 appPath = os.path.join(a.path, a.entrypoint)102 #check with list window if exists103 windowNames = util.getWindowList()104 if a.appName not in windowNames:105 print(a.path)106 ryubackendpath = "bash -c \' cd ~/netide/Engine/ryu-backend/ && " + cmd + " --ofp-tcp-listen-port=" + str(Ryu.ryubackendport) + " ryu-backend.py " + os.path.join(a.path, a.entrypoint) + "\' "107 call(['tmux', 'new-window', '-n', a.appName, '-t', 'NetIDE', ryubackendpath])108 Ryu.ryubackendport = Ryu.ryubackendport + 1109 #call(['tmux', 'send-keys', '-t', 'NetIDE' ,cmd, ' --ofp-tcp-listen-port=' +str(a.appPort) + " " + os.path.join(a.path, a.entrypoint), 'C-m'])110 else:111 print("app " + a.appName + " already running")112 if "sleepafter" in ryuCommands:113 time.sleep(ryuCommands["sleepafter"])114class FloodLight(Base):115 @classmethod116 def getControllerName(cls):117 return "floodlight"118 @classmethod119 def version(cls):120 pass121 def start(self):122 pass123#===============================================================================124# @classmethod125# def version(cls):126# v = subprocess.check_output(["cd ~/floodlight; git describe; exit 0"], shell=True, stderr=subprocess.STDOUT)127# return v.decode("utf-8").split("-")[0].strip()128#129# def start(self):130# # XXX: application modules are not copied into floodlight right now, they need to be copied manually131# try:132# with util.FLock(open(os.path.join(self.dataroot, "controllers.json"), "r"), shared=True) as fh:133# data = json.load(fh)134# except Exception as err:135# logging.debug("{}: {}".format(type(err), err))136# data = {}137# running_apps = data.get("controllers", {}).get("Ryu", {}).get("apps", [])138#139# for a in self.applications:140# if not a.enabled:141# logging.info("Skipping disabled application {}".format(a))142# continue143# if a in running_apps:144# logging.info("App {} is already running".format(a))145# continue146#147# prefix = os.path.expanduser("~/floodlight/src/main/resources")148#149# with open(os.path.join(prefix, "META-INF/services/net.floodlightcontroller.core.module.IFloodlightModule"), "r+") as fh:150# fh.write("{}\n".format(a.metadata.get("param", "")))151#152# with open(os.path.join(prefix, "floodlightdefault.properties"), "r+") as fh:153# lines = fh.readlines()154# idx = 0155# for l in lines:156# if not l.strip().endswith('\\'):157# break158# idx += 1159# lines.insert(idx, "{}\n".format(a.metadata.get("param", "")))160# fh.seek(0)161# fh.truncate()162# fh.writelines(lines)163#164# myid = str(uuid.uuid4())165# logdir = self.makelogdir(myid)166#167# serr = open(os.path.join(logdir, "stderr"), "w")168# sout = open(os.path.join(logdir, "stdout"), "w")169#170# # Rebuild floodlight with ant171# cmdline = ["cd ~/floodlight; ant"]172# subprocess.Popen(cmdline, stderr=serr, stdout=sout, shell=True).wait() # TODO: check exit value?173#174# # Start floodlight175# cmdline = ["cd ~/floodlight; java -jar floodlight/target/floodlight.jar"]176# return [{ "id": myid, "pid": subprocess.Popen(cmdline, stderr=serr, stdout=sout, shell=True).id }]177#===============================================================================178class Core(Base):179 packagePath = ""180 def __init__(self, path):181 self.packagePath = path182 def start(self):183 y = yaml.load(open("commands.yml"))184 coreCommands = y['core']185 sessionExists = call(["tmux", "has-session", "-t", "NetIDE"])186 if [ sessionExists != 0 ]:187 call(["tmux", "new-session", "-d", "-s", "NetIDE"])188 time.sleep(1)189 list = util.getWindowList()190 if "Core" not in list:191 call(['tmux', 'new-window', '-n', "Core", '-t', 'NetIDE', "bash -c \'cd ~/netide/core-karaf/bin/ && ./karaf\'"])192 compositionPath = os.path.join(self.packagePath, "composition/composition.xml")193 time.sleep(coreCommands['sleepafter'])194 call(['tmux', 'send-keys', '-t', 'NetIDE' , "netide:loadcomposition "+compositionPath, 'C-m'])195 #call(['tmux', 'send-keys', '-t', 'NetIDE' , "log:tail", 'C-m'])196 else:197 print("Core already running")198class ODL(Base):199 def start(self):200 y = yaml.load(open("commands.yml"))201 odlCommands = y['odl']202 sessionExists = call(["tmux", "has-session", "-t", "NetIDE"])203 if [ sessionExists != 0 ]:204 call(["tmux", "new-session", "-d", "-s", "NetIDE"])205 list = util.getWindowList()206 if "ODL" not in list:207 call(['tmux', 'new-window', '-n', "ODL", '-t', 'NetIDE', '~/netide/distribution-karaf-0.4.0-Beryllium/bin/karaf'])208 time.sleep(odlCommands['sleepafter'])209 else:210 print("ODL already running")211class Mininet(Base):212 def start(self):213 y = yaml.load(open("commands.yml"))214 mininetCommands = y['mininet']215 sessionExists = call(["tmux", "has-session", "-t", "NetIDE"])216 if [ sessionExists != 0 ]:217 call(["tmux", "new-session", "-d", "-s", "NetIDE"])218 time.sleep(1)219 list = util.getWindowList()220 if Mininet not in list:221 call(['tmux', 'new-window', '-n', "Mininet", '-t', 'NetIDE', "sudo python ~/Engine/loader/Demo/gen/mininet/Demo_run.py"])222 #call(['tmux', 'send-keys', '-t', 'NetIDE' , "sudo python ~/Engine/loader/Demo/gen/mininet/Demo_run.py", 'C-m'])223 time.sleep(mininetCommands['sleepafter'])224 else:225 print("Mininet already running")226class POX(Base):227 pass228class Pyretic(Base):...
synctex_evince_forward.py
Source:synctex_evince_forward.py
1# The code in this files is borrowed from Gedit Synctex plugin.2#3# Copyright (C) 2010 Jose Aliste4#5# This program is free software; you can redistribute it and/or modify it under6# the terms of the GNU General Public Licence as published by the Free Software7# Foundation; either version 2 of the Licence, or (at your option) any later8# version.9#10# This program is distributed in the hope that it will be useful, but WITHOUT11# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS12# FOR A PARTICULAR PURPOSE. See the GNU General Public Licence for more13# details.14#15# You should have received a copy of the GNU General Public Licence along with16# this program; if not, write to the Free Software Foundation, Inc., 51 Franklin17# Street, Fifth Floor, Boston, MA 02110-1301, USA18# Modified to Nvim-R by Jakson Aquino19import dbus20import dbus.mainloop.glib, sys, os21from gi.repository import GObject22RUNNING, CLOSED = range(2)23EV_DAEMON_PATH = "/org/gnome/evince/Daemon"24EV_DAEMON_NAME = "org.gnome.evince.Daemon"25EV_DAEMON_IFACE = "org.gnome.evince.Daemon"26EVINCE_PATH = "/org/gnome/evince/Evince"27EVINCE_IFACE = "org.gnome.evince.Application"28EV_WINDOW_IFACE = "org.gnome.evince.Window"29class EvinceWindowProxy:30 """A DBUS proxy for an Evince Window."""31 daemon = None32 bus = None33 def __init__(self, uri, spawn = False):34 self.uri = uri35 self.spawn = spawn36 self.status = CLOSED37 self.source_handler = None38 self.dbus_name = ''39 self._handler = None40 try:41 if EvinceWindowProxy.bus is None:42 EvinceWindowProxy.bus = dbus.SessionBus()43 if EvinceWindowProxy.daemon is None:44 EvinceWindowProxy.daemon = EvinceWindowProxy.bus.get_object(EV_DAEMON_NAME,45 EV_DAEMON_PATH,46 follow_name_owner_changes=True)47 EvinceWindowProxy.bus.add_signal_receiver(self._on_doc_loaded, signal_name="DocumentLoaded",48 dbus_interface = EV_WINDOW_IFACE,49 sender_keyword='sender')50 self._get_dbus_name(False)51 except dbus.DBusException:52 sys.stderr.write("Could not connect to the Evince Daemon")53 sys.stderr.flush()54 def _on_doc_loaded(self, uri, **keyargs):55 if uri == self.uri and self._handler is None:56 self.handle_find_document_reply(keyargs['sender'])57 def _get_dbus_name(self, spawn):58 EvinceWindowProxy.daemon.FindDocument(self.uri,spawn,59 reply_handler=self.handle_find_document_reply,60 error_handler=self.handle_find_document_error,61 dbus_interface = EV_DAEMON_IFACE)62 def handle_find_document_error(self, error):63 sys.stderr.write("FindDocument DBus call has failed")64 sys.stderr.flush()65 def handle_find_document_reply(self, evince_name):66 if self._handler is not None:67 handler = self._handler68 else:69 handler = self.handle_get_window_list_reply70 if evince_name != '':71 self.dbus_name = evince_name72 self.status = RUNNING73 self.evince = EvinceWindowProxy.bus.get_object(self.dbus_name, EVINCE_PATH)74 self.evince.GetWindowList(dbus_interface = EVINCE_IFACE,75 reply_handler = handler,76 error_handler = self.handle_get_window_list_error)77 def handle_get_window_list_error (self, e):78 sys.stderr.write("GetWindowList DBus call has failed")79 sys.stderr.flush()80 def handle_get_window_list_reply (self, window_list):81 if len(window_list) > 0:82 window_obj = EvinceWindowProxy.bus.get_object(self.dbus_name, window_list[0])83 self.window = dbus.Interface(window_obj,EV_WINDOW_IFACE)84 else:85 #That should never happen.86 sys.stderr.write("GetWindowList returned empty list")87 sys.stderr.flush()88 def SyncView(self, input_file, data):89 if self.status == CLOSED:90 if self.spawn:91 self._tmp_syncview = [input_file, data, 0];92 self._handler = self._syncview_handler93 self._get_dbus_name(True)94 sys.stdout.write("call Evince_Again()")95 sys.stdout.flush()96 else:97 self.window.SyncView(input_file, data, 0, dbus_interface = "org.gnome.evince.Window")98 sys.stdout.write("let g:rplugin_evince_loop = 0")99 sys.stdout.flush()100 def _syncview_handler(self, window_list):101 self.handle_get_window_list_reply(window_list)102 if self.status == CLOSED:103 return False104 self.window.SyncView(self._tmp_syncview[0],self._tmp_syncview[1], self._tmp_syncview[2], dbus_interface="org.gnome.evince.Window")105 del self._tmp_syncview106 self._handler = None107 return True108path_output = os.getcwd() + '/' + sys.argv[1]109path_output = path_output.replace(" ", "%20")110line_number = int(sys.argv[2])111path_input = os.getcwd() + '/' + sys.argv[3]112dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)113a = EvinceWindowProxy('file://' + path_output, True)114def sync_view(ev_window, path_input, line_number):115 ev_window.SyncView(path_input, (line_number, 1))116 loop.quit()117GObject.timeout_add(400, sync_view, a, path_input, line_number)118loop = GObject.MainLoop()...
synctex_evince_backward.py
Source:synctex_evince_backward.py
1# The code in this files is borrowed from Gedit Synctex plugin.2#3# Copyright (C) 2010 Jose Aliste4#5# This program is free software; you can redistribute it and/or modify it under6# the terms of the GNU General Public Licence as published by the Free Software7# Foundation; either version 2 of the Licence, or (at your option) any later8# version.9#10# This program is distributed in the hope that it will be useful, but WITHOUT11# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS12# FOR A PARTICULAR PURPOSE. See the GNU General Public Licence for more13# details.14#15# You should have received a copy of the GNU General Public Licence along with16# this program; if not, write to the Free Software Foundation, Inc., 51 Franklin17# Street, Fifth Floor, Boston, MA 02110-1301, USA18# Modified to Nvim-R by Jakson Aquino19import dbus, subprocess, time20import dbus.mainloop.glib, sys, os, signal21from gi.repository import GObject22RUNNING, CLOSED = range(2)23EV_DAEMON_PATH = "/org/gnome/evince/Daemon"24EV_DAEMON_NAME = "org.gnome.evince.Daemon"25EV_DAEMON_IFACE = "org.gnome.evince.Daemon"26EVINCE_PATH = "/org/gnome/evince/Evince"27EVINCE_IFACE = "org.gnome.evince.Application"28EV_WINDOW_IFACE = "org.gnome.evince.Window"29class EvinceWindowProxy:30 """A DBUS proxy for an Evince Window."""31 daemon = None32 bus = None33 def __init__(self, uri, spawn = False):34 self.uri = uri35 self.spawn = spawn36 self.status = CLOSED37 self.dbus_name = ''38 self._handler = None39 try:40 if EvinceWindowProxy.bus is None:41 EvinceWindowProxy.bus = dbus.SessionBus()42 if EvinceWindowProxy.daemon is None:43 EvinceWindowProxy.daemon = EvinceWindowProxy.bus.get_object(EV_DAEMON_NAME,44 EV_DAEMON_PATH,45 follow_name_owner_changes=True)46 EvinceWindowProxy.bus.add_signal_receiver(self._on_doc_loaded, signal_name="DocumentLoaded",47 dbus_interface = EV_WINDOW_IFACE,48 sender_keyword='sender')49 self._get_dbus_name(False)50 except dbus.DBusException:51 sys.stderr.write("Could not connect to the Evince Daemon")52 sys.stderr.flush()53 loop.quit()54 def _on_doc_loaded(self, uri, **keyargs):55 if uri == self.uri and self._handler is None:56 self.handle_find_document_reply(keyargs['sender'])57 def _get_dbus_name(self, spawn):58 EvinceWindowProxy.daemon.FindDocument(self.uri,spawn,59 reply_handler=self.handle_find_document_reply,60 error_handler=self.handle_find_document_error,61 dbus_interface = EV_DAEMON_IFACE)62 def handle_find_document_error(self, error):63 sys.stderr.write("FindDocument DBus call has failed")64 sys.stderr.flush()65 def handle_find_document_reply(self, evince_name):66 if self._handler is not None:67 handler = self._handler68 else:69 handler = self.handle_get_window_list_reply70 if evince_name != '':71 self.dbus_name = evince_name72 self.status = RUNNING73 self.evince = EvinceWindowProxy.bus.get_object(self.dbus_name, EVINCE_PATH)74 self.evince.GetWindowList(dbus_interface = EVINCE_IFACE,75 reply_handler = handler,76 error_handler = self.handle_get_window_list_error)77 def handle_get_window_list_error (self, e):78 sys.stderr.write("GetWindowList DBus call has failed")79 sys.stderr.flush()80 def handle_get_window_list_reply (self, window_list):81 if len(window_list) > 0:82 window_obj = EvinceWindowProxy.bus.get_object(self.dbus_name, window_list[0])83 self.window = dbus.Interface(window_obj,EV_WINDOW_IFACE)84 self.window.connect_to_signal("Closed", self.on_window_close)85 self.window.connect_to_signal("SyncSource", self.on_sync_source)86 else:87 #That should never happen.88 sys.stderr.write("GetWindowList returned empty list")89 sys.stderr.flush()90 def on_window_close(self):91 self.window = None92 self.status = CLOSED93 def on_sync_source(self, input_file, source_link, timestamp):94 input_file = input_file.replace("file://", "")95 input_file = input_file.replace("%20", " ")96 sys.stdout.write("call SyncTeX_backward('" + input_file + "', " + str(source_link[0]) + ")\n")97 sys.stdout.flush()98path_output = os.getcwd() + '/' + sys.argv[1]99path_output = path_output.replace(" ", "%20")100dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)101a = EvinceWindowProxy('file://' + path_output, True)102loop = GObject.MainLoop()...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pyatom automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
