ScreenGear is designed exclusively for targeting rapid Screencasting Capabilities, which means it can grab frames from your monitor in real-time, either by defining an area on the computer screen or full-screen, at the expense of inconsiderable latency. ScreenGear also seamlessly support frame capturing from multiple monitors as well as supports multiple backends.
ScreenGear API implements a multi-threaded wrapper around dxcam, pyscreenshot, python-mss python library, and also flexibly supports its internal parameter.
classScreenGear:""" ScreenGear is designed exclusively for targeting rapid Screencasting Capabilities, which means it can grab frames from your monitor in real-time, either by defining an area on the computer screen or full-screen, at the expense of inconsiderable latency. ScreenGear also seamlessly support frame capturing from multiple monitors as well as supports multiple backends. ScreenGear API implements a multi-threaded wrapper around dxcam, pyscreenshot, python-mss python library, and also flexibly supports its internal parameter. """def__init__(self,monitor:int=None,backend:str=None,colorspace:str=None,logging:bool=False,**options:dict):""" This constructor method initializes the object state and attributes of the ScreenGear class. Parameters: monitor (int): enables `mss` backend and sets the index of the monitor screen. backend (str): select suitable backend for extracting frames. colorspace (str): selects the colorspace of the input stream. logging (bool): enables/disables logging. options (dict): provides the flexibility to easily alter backend library parameters. Such as, manually set the dimensions of capture screen area etc. """# enable logging if specifiedself.__logging=loggingifisinstance(logging,bool)elseFalse# print current versionlogcurr_vidgear_ver(logging=self.__logging)# create instances for the user-defined monitorself.__monitor_instance=Noneself.__backend=None# validate monitor instanceassert(monitorisNoneormonitorandisinstance(monitor,(int,tuple))),"[ScreenGear:ERROR] :: Invalid `monitor` value detected!"# initialize backendifbackendandmonitorisNone:self.__backend=backend.lower().strip()else:# enforce `dxcam` for Windows machines if undefined (or monitor is defined)self.__backend=("dxcam"ifplatform.system()=="Windows"anddxcamelseNone)# initiate screen dimension handlerscreen_dims={}# reformat proper mss dict and assign to screen dimension handlerscreen_dims={k.strip():vfork,vinoptions.items()ifk.strip()in["top","left","width","height"]}# check whether user-defined dimensions are providedifscreen_dimsandlen(screen_dims)==4:key_order=(("top","left","width","height")ifself.__backend!="dxcam"else("left","top","width","height"))screen_dims=OrderedDict((k,screen_dims[k])forkinkey_order)self.__loggingandlogger.debug("Setting Capture-Area dimensions: {}".format(json.dumps(screen_dims)))else:screen_dims.clear()# handle backendsifself.__backend=="dxcam":# get target fps in case of DXcamself.__target_fps=options.pop("dxcam_target_fps",0)ifself.__target_fpsandisinstance(self.__target_fps,(int,float)):# set valuesself.__target_fps=int(self.__target_fps)self.__loggingandlogger.debug("Setting Target FPS: {}".format(self.__target_fps))else:# defaults to 0fpsself.__target_fps=0# check if platform is windowsassert(platform.system()=="Windows"),"`dxcam` backend is only available for Windows Machines."# verify monitor values if tupleassert(monitorisNoneorisinstance(monitor,int)or(isinstance(monitor,tuple)andlen(monitor)==2andall(isinstance(x,int)forxinmonitor))),"For dxcam` backend, monitor` tuple value must be format `int` or `(int, int)` only."# raise error(s) for critical Class importsimport_dependency_safe("dxcam"ifdxcamisNoneelse"")ifmonitorisNone:self.__capture_object=dxcam.create(region=tuple(screen_dims.values())ifscreen_dimselseNone)else:self.__capture_object=(dxcam.create(device_idx=monitor[0],output_idx=monitor[1],region=tuple(screen_dims.values())ifscreen_dimselseNone,)ifisinstance(monitor,tuple)elsedxcam.create(device_idx=monitor,region=tuple(screen_dims.values())ifscreen_dimselseNone,))else:ifmonitorisNone:# raise error(s) for critical Class importsimport_dependency_safe("pyscreenshot"ifpysctisNoneelse"")# reset backend if not providedself.__backend="pil"ifself.__backendisNoneelseself.__backend# check if valid backendassert(self.__backendinpysct.backends()),"Unsupported backend {} provided!".format(backend)# create capture objectself.__capture_object=pysctelse:# monitor value must be integerassertmonitorandisinstance(monitor,int),"[ScreenGear:ERROR] :: Invalid `monitor` value must be integer with mss backend."# raise error(s) for critical Class importsimport_dependency_safe("from mss import mss"ifmssisNoneelse"",pkg_name="mss")# create capture objectself.__capture_object=mss()self.__backendandlogger.warning("Backends are disabled for Monitor Indexing(monitor>=0)!")self.__monitor_instance=self.__capture_object.monitors[monitor]# log backendself.__backendandself.__loggingandlogger.debug("Setting Backend: {}".format(self.__backend.upper()))# assigns special parameter to global variable and clear# separately handle colorspace value to int conversionifcolorspace:self.color_space=capPropId(colorspace.strip())self.__loggingandnot(self.color_spaceisNone)andlogger.debug("Enabling `{}` colorspace for this video stream!".format(colorspace.strip()))else:self.color_space=None# initialize mss capture instanceself.__mss_capture_instance=Nonetry:ifself.__backend=="dxcam":# extract global frame from instanceself.frame=self.__capture_object.grab()else:ifself.__monitor_instanceisNone:ifscreen_dims:self.__mss_capture_instance=tuple(screen_dims.values())# extract global frame from instanceself.frame=np.asanyarray(self.__capture_object.grab(bbox=self.__mss_capture_instance,childprocess=False,backend=self.__backend,))else:ifscreen_dims:self.__mss_capture_instance={"top":self.__monitor_instance["top"]+screen_dims["top"],"left":self.__monitor_instance["left"]+screen_dims["left"],"width":screen_dims["width"],"height":screen_dims["height"],"mon":monitor,}else:self.__mss_capture_instance=(self.__monitor_instance# otherwise create instance from monitor)# extract global frame from instanceself.frame=np.asanyarray(self.__capture_object.grab(self.__mss_capture_instance))# convert to bgr frame if applicableself.frame=(self.frame[:,:,::-1]ifself.__backend=="dxcam"ornot(pysctisNone)elseself.frame)# render colorspace if definedifnot(self.frameisNone)andnot(self.color_spaceisNone):self.frame=cv2.cvtColor(self.frame,self.color_space)exceptExceptionase:ifisinstance(e,ScreenShotError):# otherwise catch and log errorsself.__loggingandlogger.exception(self.__capture_object.get_error_details())raiseValueError("[ScreenGear:ERROR] :: ScreenShotError caught, Wrong dimensions passed to python-mss, Kindly Refer Docs!")else:raiseSystemError("[ScreenGear:ERROR] :: Unable to grab any instance on this system, Are you running headless?")# thread initializationself.__thread=None# initialize termination flagself.__terminate=Event()defstart(self)->T:""" Launches the internal *Threaded Frames Extractor* daemon **Returns:** A reference to the ScreenGear class object. """self.__thread=Thread(target=self.__update,name="ScreenGear",args=())self.__thread.daemon=Trueself.__thread.start()ifself.__backend=="dxcam":self.__capture_object.start(target_fps=self.__target_fps,video_mode=True,)self.__loggingandself.__target_fpsandlogger.debug("Targeting FPS: {}".format(self.__target_fps))returnselfdef__update(self):""" A **Threaded Frames Extractor**, that keep iterating frames from `mss` API to a internal monitored deque, until the thread is terminated, or frames runs out. """# initialize frame variableframe=None# keep looping infinitely until the thread is terminatedwhilenotself.__terminate.is_set():try:ifself.__backend=="dxcam":# extract global frame from instanceframe=self.__capture_object.get_latest_frame()else:ifself.__monitor_instance:frame=np.asanyarray(self.__capture_object.grab(self.__mss_capture_instance))else:frame=np.asanyarray(self.__capture_object.grab(bbox=self.__mss_capture_instance,childprocess=False,backend=self.__backend,))# check if valid frameassertnot(frameisNoneornp.shape(frame)==()),"[ScreenGear:ERROR] :: Failed to retrieve valid frame!"# convert to bgr frame if applicableframe=(frame[:,:,::-1]ifself.__backend=="dxcam"ornot(pysctisNone)elseframe)exceptExceptionase:ifisinstance(e,ScreenShotError):raiseRuntimeError(self.__capture_object.get_error_details())else:logger.exception(str(e))self.__terminate.set()continueifnot(self.color_spaceisNone):# apply colorspace to framescolor_frame=Nonetry:color_frame=cv2.cvtColor(frame,self.color_space)exceptExceptionase:# Catch if any error occurredcolor_frame=Noneself.color_space=Noneself.__loggingandlogger.exception(str(e))logger.warning("Assigned colorspace value is invalid. Discarding!")self.frame=color_frameifnot(color_frameisNone)elseframeelse:self.frame=frame# indicate immediate terminationself.__terminate.set()# finally release mss resourcesifself.__monitor_instance:self.__capture_object.close()ifself.__backend=="dxcam":self.__capture_object.stop()delself.__capture_objectdefread(self)->NDArray:""" Extracts frames synchronously from monitored deque, while maintaining a fixed-length frame buffer in the memory, and blocks the thread if the deque is full. **Returns:** A n-dimensional numpy array. """# return the framereturnself.framedefstop(self)->None:""" Safely terminates the thread, and release the resources. """self.__loggingandlogger.debug("Terminating ScreenGear Processes.")# indicate that the thread should be terminateself.__terminate.set()# wait until stream resources are released (producer thread might be still grabbing frame)not(self.__threadisNone)andself.__thread.join()
def__init__(self,monitor:int=None,backend:str=None,colorspace:str=None,logging:bool=False,**options:dict):""" This constructor method initializes the object state and attributes of the ScreenGear class. Parameters: monitor (int): enables `mss` backend and sets the index of the monitor screen. backend (str): select suitable backend for extracting frames. colorspace (str): selects the colorspace of the input stream. logging (bool): enables/disables logging. options (dict): provides the flexibility to easily alter backend library parameters. Such as, manually set the dimensions of capture screen area etc. """# enable logging if specifiedself.__logging=loggingifisinstance(logging,bool)elseFalse# print current versionlogcurr_vidgear_ver(logging=self.__logging)# create instances for the user-defined monitorself.__monitor_instance=Noneself.__backend=None# validate monitor instanceassert(monitorisNoneormonitorandisinstance(monitor,(int,tuple))),"[ScreenGear:ERROR] :: Invalid `monitor` value detected!"# initialize backendifbackendandmonitorisNone:self.__backend=backend.lower().strip()else:# enforce `dxcam` for Windows machines if undefined (or monitor is defined)self.__backend=("dxcam"ifplatform.system()=="Windows"anddxcamelseNone)# initiate screen dimension handlerscreen_dims={}# reformat proper mss dict and assign to screen dimension handlerscreen_dims={k.strip():vfork,vinoptions.items()ifk.strip()in["top","left","width","height"]}# check whether user-defined dimensions are providedifscreen_dimsandlen(screen_dims)==4:key_order=(("top","left","width","height")ifself.__backend!="dxcam"else("left","top","width","height"))screen_dims=OrderedDict((k,screen_dims[k])forkinkey_order)self.__loggingandlogger.debug("Setting Capture-Area dimensions: {}".format(json.dumps(screen_dims)))else:screen_dims.clear()# handle backendsifself.__backend=="dxcam":# get target fps in case of DXcamself.__target_fps=options.pop("dxcam_target_fps",0)ifself.__target_fpsandisinstance(self.__target_fps,(int,float)):# set valuesself.__target_fps=int(self.__target_fps)self.__loggingandlogger.debug("Setting Target FPS: {}".format(self.__target_fps))else:# defaults to 0fpsself.__target_fps=0# check if platform is windowsassert(platform.system()=="Windows"),"`dxcam` backend is only available for Windows Machines."# verify monitor values if tupleassert(monitorisNoneorisinstance(monitor,int)or(isinstance(monitor,tuple)andlen(monitor)==2andall(isinstance(x,int)forxinmonitor))),"For dxcam` backend, monitor` tuple value must be format `int` or `(int, int)` only."# raise error(s) for critical Class importsimport_dependency_safe("dxcam"ifdxcamisNoneelse"")ifmonitorisNone:self.__capture_object=dxcam.create(region=tuple(screen_dims.values())ifscreen_dimselseNone)else:self.__capture_object=(dxcam.create(device_idx=monitor[0],output_idx=monitor[1],region=tuple(screen_dims.values())ifscreen_dimselseNone,)ifisinstance(monitor,tuple)elsedxcam.create(device_idx=monitor,region=tuple(screen_dims.values())ifscreen_dimselseNone,))else:ifmonitorisNone:# raise error(s) for critical Class importsimport_dependency_safe("pyscreenshot"ifpysctisNoneelse"")# reset backend if not providedself.__backend="pil"ifself.__backendisNoneelseself.__backend# check if valid backendassert(self.__backendinpysct.backends()),"Unsupported backend {} provided!".format(backend)# create capture objectself.__capture_object=pysctelse:# monitor value must be integerassertmonitorandisinstance(monitor,int),"[ScreenGear:ERROR] :: Invalid `monitor` value must be integer with mss backend."# raise error(s) for critical Class importsimport_dependency_safe("from mss import mss"ifmssisNoneelse"",pkg_name="mss")# create capture objectself.__capture_object=mss()self.__backendandlogger.warning("Backends are disabled for Monitor Indexing(monitor>=0)!")self.__monitor_instance=self.__capture_object.monitors[monitor]# log backendself.__backendandself.__loggingandlogger.debug("Setting Backend: {}".format(self.__backend.upper()))# assigns special parameter to global variable and clear# separately handle colorspace value to int conversionifcolorspace:self.color_space=capPropId(colorspace.strip())self.__loggingandnot(self.color_spaceisNone)andlogger.debug("Enabling `{}` colorspace for this video stream!".format(colorspace.strip()))else:self.color_space=None# initialize mss capture instanceself.__mss_capture_instance=Nonetry:ifself.__backend=="dxcam":# extract global frame from instanceself.frame=self.__capture_object.grab()else:ifself.__monitor_instanceisNone:ifscreen_dims:self.__mss_capture_instance=tuple(screen_dims.values())# extract global frame from instanceself.frame=np.asanyarray(self.__capture_object.grab(bbox=self.__mss_capture_instance,childprocess=False,backend=self.__backend,))else:ifscreen_dims:self.__mss_capture_instance={"top":self.__monitor_instance["top"]+screen_dims["top"],"left":self.__monitor_instance["left"]+screen_dims["left"],"width":screen_dims["width"],"height":screen_dims["height"],"mon":monitor,}else:self.__mss_capture_instance=(self.__monitor_instance# otherwise create instance from monitor)# extract global frame from instanceself.frame=np.asanyarray(self.__capture_object.grab(self.__mss_capture_instance))# convert to bgr frame if applicableself.frame=(self.frame[:,:,::-1]ifself.__backend=="dxcam"ornot(pysctisNone)elseself.frame)# render colorspace if definedifnot(self.frameisNone)andnot(self.color_spaceisNone):self.frame=cv2.cvtColor(self.frame,self.color_space)exceptExceptionase:ifisinstance(e,ScreenShotError):# otherwise catch and log errorsself.__loggingandlogger.exception(self.__capture_object.get_error_details())raiseValueError("[ScreenGear:ERROR] :: ScreenShotError caught, Wrong dimensions passed to python-mss, Kindly Refer Docs!")else:raiseSystemError("[ScreenGear:ERROR] :: Unable to grab any instance on this system, Are you running headless?")# thread initializationself.__thread=None# initialize termination flagself.__terminate=Event()
Extracts frames synchronously from monitored deque, while maintaining a fixed-length frame buffer in the memory, and blocks the thread if the deque is full.
defread(self)->NDArray:""" Extracts frames synchronously from monitored deque, while maintaining a fixed-length frame buffer in the memory, and blocks the thread if the deque is full. **Returns:** A n-dimensional numpy array. """# return the framereturnself.frame
defstart(self)->T:""" Launches the internal *Threaded Frames Extractor* daemon **Returns:** A reference to the ScreenGear class object. """self.__thread=Thread(target=self.__update,name="ScreenGear",args=())self.__thread.daemon=Trueself.__thread.start()ifself.__backend=="dxcam":self.__capture_object.start(target_fps=self.__target_fps,video_mode=True,)self.__loggingandself.__target_fpsandlogger.debug("Targeting FPS: {}".format(self.__target_fps))returnself
defstop(self)->None:""" Safely terminates the thread, and release the resources. """self.__loggingandlogger.debug("Terminating ScreenGear Processes.")# indicate that the thread should be terminateself.__terminate.set()# wait until stream resources are released (producer thread might be still grabbing frame)not(self.__threadisNone)andself.__thread.join()