How to use hotplug method in avocado

Best Python code snippet using avocado_python

Hotplug.py

Source:Hotplug.py

1#!/usr/bin/python2# -*- coding: utf-8 -*- 3#  Advanced Movie Selection for Dreambox-Enigma24#5#  Coded by cmikula (c)20136#  Support: www.i-have-a-dreambox.com7#8#  This plugin is licensed under the Creative Commons 9#  Attribution-NonCommercial-ShareAlike 3.0 Unported 10#  License. To view a copy of this license, visit11#  http://creativecommons.org/licenses/by-nc-sa/3.0/ or send a letter to Creative12#  Commons, 559 Nathan Abbott Way, Stanford, California 94305, USA.13#14#  Alternatively, this plugin may be distributed and executed on hardware which15#  is licensed by Dream Multimedia GmbH.16#17#  This plugin is NOT free software. It is open source, you are allowed to18#  modify it (if you keep the license), but it may not be commercially 19#  distributed other than under the conditions noted above.20#21import os22from Globals import printStackTrace23from MovieConfig import MovieConfig24from ServiceProvider import eServiceReferenceHotplug25from enigma import eTimer26class Hotplug():27    NTFS_3G_DRIVER_DELAY = 300028    def __init__(self):29        self.notifier = []30        self.hotplugServices = []31        self.hotplug_timer = eTimer()32        self.hotplug_timer.callback.append(self.updateHotplugDevices)33        self.addHotplugNotifier()34        self.hotplugChanged()35    def addHotplugNotifier(self):36        from Plugins.SystemPlugins.Hotplug.plugin import hotplugNotifier37        if not self.hotplugNotifier in hotplugNotifier:38            print "add hotplugNotifier" 39            hotplugNotifier.append(self.hotplugNotifier)40        41    def removeHotplugNotifier(self):42        from Plugins.SystemPlugins.Hotplug.plugin import hotplugNotifier43        if self.hotplugNotifier in hotplugNotifier:44            print "remove hotplugNotifier" 45            hotplugNotifier.remove(self.hotplugNotifier)46    47    def hotplugNotifier(self, dev, media_state):48        print "[hotplugNotifier]", dev, media_state49        if len(dev) > 2 and dev[0:2] in ("sd") and dev[-1].isdigit():50            if media_state == "add":51                self.hotplugChanged(self.NTFS_3G_DRIVER_DELAY)52            else:53                self.hotplugChanged(200)54    def hotplugChanged(self, delay=200):55        print "[start hotplugNotifier]", str(delay) + "ms"56        self.hotplug_timer.start(delay, True)57    def updateHotplugDevices(self):58        self.hotplugServices = []59        print "[update hutplug]"60        try:61            from Components.Harddisk import Harddisk62            import commands63            movieConfig = MovieConfig()64            lines = commands.getoutput('mount | grep /dev/sd').split('\n')65            print lines66            for mount in lines:67                if len(mount) < 2:68                    continue69                m = mount.split(' type')[0].split(' on ')70                m_dev, m_path = m[0], m[1]71                label = os.path.split(m_path)[-1]72                blkid = commands.getoutput('blkid ' + m_dev).split("\"")73                if len(blkid) > 2 and blkid[1]:74                    label = blkid[1]75                if os.path.normpath(m_path) == "/media/hdd" or label in ("DUMBO", "TIMOTHY"):76                    continue77                if not movieConfig.isHiddenHotplug(label):78                    if m_path[-1] != "/":79                        m_path += "/"80                    service = eServiceReferenceHotplug(m_path)81                    hdd = Harddisk(m_dev.replace("/dev/", "")[:-1])82                    if label:83                        label += " - "84                    service.setName(label + hdd.model() + " - " + hdd.capacity())85                    self.hotplugServices.append(service)86            87            for callback in self.notifier:88                try:89                    callback()90                except:91                    printStackTrace()92        except:93            printStackTrace()94    95    def getHotplugServices(self):96        return self.hotplugServices...

plugin.py

Source:plugin.py

1from Plugins.Plugin import PluginDescriptor2from Components.Harddisk import harddiskmanager3from twisted.internet.protocol import Protocol, Factory4hotplugNotifier = []5def processHotplugData(self, v):6	print "hotplug:", v7	action = v.get("ACTION")8	device = v.get("DEVPATH")9	physdevpath = v.get("PHYSDEVPATH")10	media_state = v.get("X_E2_MEDIA_STATUS")11	dev = device.split('/')[-1]12	if action == "add":13		error, blacklisted, removable, is_cdrom, partitions, medium_found = harddiskmanager.addHotplugPartition(dev, physdevpath)14	elif action == "remove":15		harddiskmanager.removeHotplugPartition(dev)16	elif media_state is not None:17		if media_state == '1':18			harddiskmanager.removeHotplugPartition(dev)19			harddiskmanager.addHotplugPartition(dev, physdevpath)20		elif media_state == '0':21			harddiskmanager.removeHotplugPartition(dev)22	for callback in hotplugNotifier:23		try:24			callback(dev, action or media_state)25		except AttributeError:26			hotplugNotifier.remove(callback)27class Hotplug(Protocol):28	def __init__(self):29		pass30	def connectionMade(self):31		print "HOTPLUG connection!"32		self.received = ""33	def dataReceived(self, data):34		print "hotplug:", data35		self.received += data36		print "complete", self.received37	def connectionLost(self, reason):38		print "HOTPLUG connection lost!"39		data = self.received.split('\0')[:-1]40		v = {}41		for x in data:42			i = x.find('=')43			var, val = x[:i], x[i+1:]44			v[var] = val45		processHotplugData(self, v)46def autostart(reason, **kwargs):47	if reason == 0:48		print "starting hotplug handler"49		from twisted.internet import reactor50		import os51		try:52			os.remove("/tmp/hotplug.socket")53		except OSError:54			pass55		factory = Factory()56		factory.protocol = Hotplug57		reactor.listenUNIX("/tmp/hotplug.socket", factory)58def Plugins(**kwargs):...

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.