How to use hotplug method in avocado

Best Python code snippet using avocado_python Github


Full Screen

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# 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...

Full Screen

Full Screen Github


Full Screen

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):...

Full Screen

Full Screen

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:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?