sascha/devlib
1
0
Fork 0
mirror of https://github.com/ARM-software/devlib.git synced 2025-01-31 02:00:45 +00:00
Code Issues Releases Wiki Activity

collector/screencapture: Refactor to use new collector interface

Browse Source 
Update the interface to make use of the collector interface.
Notable changes are the removal of the `output_path` path provided on
initialisation which will now be provided by the dedicated `set_output`
method.
This commit is contained in:
Marc Bonnici 2019-12-05 11:09:00 +00:00
parent 9bf9f2dd1b
commit 15a77a841d
1 changed files with 23 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
devlib/collector/screencapture.py
View File

@ -19,13 +19,14 @@ import sys
import threading import threading
import time import time
from devlib.collector import BaseCollector from devlib.collector import (CollectorBase, CollectorOutput,
CollectorOutputEntry)
from devlib.exception import WorkerThreadError from devlib.exception import WorkerThreadError
class ScreenCapturePoller(threading.Thread): class ScreenCapturePoller(threading.Thread):
def __init__(self, target, period, output_path=None, timeout=30): def __init__(self, target, period, timeout=30):
super(ScreenCapturePoller, self).__init__() super(ScreenCapturePoller, self).__init__()
self.target = target self.target = target
self.logger = logging.getLogger('screencapture') self.logger = logging.getLogger('screencapture')
@ -36,11 +37,16 @@ class ScreenCapturePoller(threading.Thread):
self.last_poll = 0 self.last_poll = 0
self.daemon = True self.daemon = True
self.exc = None self.exc = None
self.output_path = None
def set_output(self, output_path):
self.output_path = output_path self.output_path = output_path
def run(self): def run(self):
self.logger.debug('Starting screen capture polling') self.logger.debug('Starting screen capture polling')
try: try:
if self.output_path is None:
raise RuntimeError("Output path was not set.")
while True: while True:
if self.stop_signal.is_set(): if self.stop_signal.is_set():
break break
@ -66,24 +72,33 @@ class ScreenCapturePoller(threading.Thread):
self.target.capture_screen(os.path.join(self.output_path, "screencap_{ts}.png")) self.target.capture_screen(os.path.join(self.output_path, "screencap_{ts}.png"))
class ScreenCaptureCollector(BaseCollector): class ScreenCaptureCollector(CollectorBase):
def __init__(self, target, output_path=None, period=None): def __init__(self, target, period=None):
super(ScreenCaptureCollector, self).__init__(target) super(ScreenCaptureCollector, self).__init__(target)
self._collecting = False self._collecting = False
self.output_path = output_path self.output_path = None
self.period = period self.period = period
self.target = target self.target = target
self._poller = ScreenCapturePoller(self.target, self.period,
self.output_path) def set_output(self, output_path):
self.output_path = output_path
def reset(self): def reset(self):
pass self._poller = ScreenCapturePoller(self.target, self.period)
def get_data(self):
if self.output_path is None:
raise RuntimeError("No data collected.")
return CollectorOutput([CollectorOutputEntry(self.output_path, 'directory')])
def start(self): def start(self):
""" """
Start collecting the screenshots Start collecting the screenshots
""" """
if self.output_path is None:
raise RuntimeError("Output path was not set.")
self._poller.set_output(self.output_path)
self._poller.start() self._poller.start()
self._collecting = True self._collecting = True