class_docstr: WebGear is a powerful ASGI Video-Broadcaster API ideal for transmitting Motion-JPEG-frames from a single source to multiple recipients via the browser.
  
  WebGear API works on Starlette's ASGI application and provides a highly extensible and flexible async wrapper around its complete framework. WebGear can
  flexibly interact with Starlette's ecosystem of shared middleware, mountable applications, Response classes, Routing tables, Static Files, Templating
  engine(with Jinja2), etc.
  
  WebGear API uses an intraframe-only compression scheme under the hood where the sequence of video-frames are first encoded as JPEG-DIB (JPEG with Device-Independent Bit compression)
  and then streamed over HTTP using Starlette's Multipart Streaming Response and a Uvicorn ASGI Server. This method imposes lower processing and memory requirements, but the quality
  is not the best, since JPEG compression is not very efficient for motion video.
  
  In layman's terms, WebGear acts as a powerful Video Broadcaster that transmits live video-frames to any web-browser in the network. Additionally, WebGear API also provides internal
  wrapper around VideoGear, which itself provides internal access to both CamGear and PiGear APIs, thereby granting it exclusive power for transferring frames incoming from any source to the network.
class_docstr_tok: ["WebGear","is","a","powerful","ASGI","Video-Broadcaster","API","ideal","for","transmitting","Motion-JPEG-frames","from","a","single","source","to","multiple","recipients","via","the","browser",".","WebGear","API","works","on","Starlette","'s","ASGI","application","and","provides","a","highly","extensible","and","flexible","async","wrapper","around","its","complete","framework",".","WebGear","can","flexibly","interact","with","Starlette","'s","ecosystem","of","shared","middleware",",","mountable","applications",",","Response","classes",",","Routing","tables",",","Static","Files",",","Templating","engine","(","with","Jinja2",")",",","etc",".","WebGear","API","uses","an","intraframe-only","compression","scheme","under","the","hood","where","the","sequence","of","video-frames","are","first","encoded","as","JPEG-DIB","(","JPEG","with","Device-Independent","Bit","compression",")","and","then","streamed","over","HTTP","using","Starlette","'s","Multipart","Streaming","Response","and","a","Uvicorn","ASGI","Server",".","This","method","imposes","lower","processing","and","memory","requirements",",","but","the","quality","is","not","the","best",",","since","JPEG","compression","is","not","very","efficient","for","motion","video",".","In","layman","'s","terms",",","WebGear","acts","as","a","powerful","Video","Broadcaster","that","transmits","live","video-frames","to","any","web-browser","in","the","network",".","Additionally",",","WebGear","API","also","provides","internal","wrapper","around","VideoGear",",","which","itself","provides","internal","access","to","both","CamGear","and","PiGear","APIs",",","thereby","granting","it","exclusive","power","for","transferring","frames","incoming","from","any","source","to","the","network","."]
docstr: This constructor method initializes the object state and attributes of the WebGear class.
  
  Parameters:
      enablePiCamera (bool): provide access to PiGear(if True) or CamGear(if False) APIs respectively.
      stabilize (bool): enable access to Stabilizer Class for stabilizing frames.
      camera_num (int): selects the camera module index which will be used as Rpi source.
      resolution (tuple): sets the resolution (i.e. `(width,height)`) of the Rpi source.
      framerate (int/float): sets the framerate of the Rpi source.
      source (based on input): defines the source for the input stream.
      stream_mode (bool): controls the exclusive YouTube Mode.
      backend (int): selects the backend for OpenCV's VideoCapture class.
      colorspace (str): selects the colorspace of the input stream.
      logging (bool): enables/disables logging.
      time_delay (int): time delay (in sec) before start reading the frames.
      options (dict): provides ability to alter Tweak Parameters of WebGear, CamGear, PiGear & Stabilizer.
docstr_tok: ["This","constructor","method","initializes","the","object","state","and","attributes","of","the","WebGear","class",".","Parameters",":","enablePiCamera","(","bool",")",":","provide","access","to","PiGear","(","if","True",")","or","CamGear","(","if","False",")","APIs","respectively",".","stabilize","(","bool",")",":","enable","access","to","Stabilizer","Class","for","stabilizing","frames",".","camera_num","(","int",")",":","selects","the","camera","module","index","which","will","be","used","as","Rpi","source",".","resolution","(","tuple",")",":","sets","the","resolution","(","i.e",".","`","(","width",",","height",")","`",")","of","the","Rpi","source",".","framerate","(","int\/float",")",":","sets","the","framerate","of","the","Rpi","source",".","source","(","based","on","input",")",":","defines","the","source","for","the","input","stream",".","stream_mode","(","bool",")",":","controls","the","exclusive","YouTube","Mode",".","backend","(","int",")",":","selects","the","backend","for","OpenCV","'s","VideoCapture","class",".","colorspace","(","str",")",":","selects","the","colorspace","of","the","input","stream",".","logging","(","bool",")",":","enables\/disables","logging",".","time_delay","(","int",")",":","time","delay","(","in","sec",")","before","start","reading","the","frames",".","options","(","dict",")",":","provides","ability","to","alter","Tweak","Parameters","of","WebGear",",","CamGear",",","PiGear","&","Stabilizer","."]
code: def __init__(
      self,
      enablePiCamera=False,
      stabilize=False,
      source=None,
      camera_num=0,
      stream_mode=False,
      backend=0,
      colorspace=None,
      resolution=(640, 480),
      framerate=25,
      logging=False,
      time_delay=0,
      **options,
  ):
      """
      This constructor method initializes the object state and attributes of the WebGear class.
  
      Parameters:
          enablePiCamera (bool): provide access to PiGear(if True) or CamGear(if False) APIs respectively.
          stabilize (bool): enable access to Stabilizer Class for stabilizing frames.
          camera_num (int): selects the camera module index which will be used as Rpi source.
          resolution (tuple): sets the resolution (i.e. `(width,height)`) of the Rpi source.
          framerate (int/float): sets the framerate of the Rpi source.
          source (based on input): defines the source for the input stream.
          stream_mode (bool): controls the exclusive YouTube Mode.
          backend (int): selects the backend for OpenCV's VideoCapture class.
          colorspace (str): selects the colorspace of the input stream.
          logging (bool): enables/disables logging.
          time_delay (int): time delay (in sec) before start reading the frames.
          options (dict): provides ability to alter Tweak Parameters of WebGear, CamGear, PiGear & Stabilizer.
      """
      # print current version
      logcurr_vidgear_ver(logging=logging)
  
      # raise error(s) for critical Class imports
      import_dependency_safe("starlette" if starlette is None else "")
      import_dependency_safe(
          "simplejpeg" if simplejpeg is None else "",
          min_version="1.6.1",
      )
  
      # initialize global params
      self.__skip_generate_webdata = (
          False  # generate webgear data by default
      )
      # define frame-compression handler
      self.__jpeg_compression_quality = 90  # 90% quality
      self.__jpeg_compression_fastdct = (
          True  # fastest DCT on by default
      )
      self.__jpeg_compression_fastupsample = (
          False  # fastupsample off by default
      )
      self.__jpeg_compression_colorspace = (
          "BGR"  # use BGR colorspace by default
      )
      self.__logging = logging
      self.__frame_size_reduction = 25  # use 25% reduction
      # retrieve interpolation for reduction
      self.__interpolation = retrieve_best_interpolation(
          ["INTER_LINEAR_EXACT", "INTER_LINEAR", "INTER_AREA"]
      )
  
      custom_video_endpoint = ""  # custom video endpoint path
      custom_data_location = (
          ""  # path to save data-files to custom location
      )
      data_path = ""  # path to WebGear data-files
      overwrite_default = False
      self.__enable_inf = False  # continue frames even when video ends.
  
      # reformat dictionary
      options = {str(k).strip(): v for k, v in options.items()}
  
      # assign values to global variables if specified and valid
      if options:
          # check whether to disable Data-Files Auto-Generation WorkFlow
          if "skip_generate_webdata" in options:
              value = options["skip_generate_webdata"]
              # enable jpeg fastdct
              if isinstance(value, bool):
                  self.__skip_generate_webdata = value
              else:
                  logger.warning(
                      "Skipped invalid `skip_generate_webdata` value!"
                  )
              del options["skip_generate_webdata"]  # clean
  
          if "jpeg_compression_colorspace" in options:
              value = options["jpeg_compression_colorspace"]
              if isinstance(value, str) and value.strip().upper() in [
                  "RGB",
                  "BGR",
                  "RGBX",
                  "BGRX",
                  "XBGR",
                  "XRGB",
                  "GRAY",
                  "RGBA",
                  "BGRA",
                  "ABGR",
                  "ARGB",
                  "CMYK",
              ]:
                  # set encoding colorspace
                  self.__jpeg_compression_colorspace = (
                      value.strip().upper()
                  )
              else:
                  logger.warning(
                      "Skipped invalid `jpeg_compression_colorspace` value!"
                  )
              del options["jpeg_compression_colorspace"]  # clean
  
          if "jpeg_compression_quality" in options:
              value = options["jpeg_compression_quality"]
              # set valid jpeg quality
              if (
                  isinstance(value, (int, float))
                  and value >= 10
                  and value <= 100
              ):
                  self.__jpeg_compression_quality = int(value)
              else:
                  logger.warning(
                      "Skipped invalid `jpeg_compression_quality` value!"
                  )
              del options["jpeg_compression_quality"]  # clean
  
          if "jpeg_compression_fastdct" in options:
              value = options["jpeg_compression_fastdct"]
              # enable jpeg fastdct
              if isinstance(value, bool):
                  self.__jpeg_compression_fastdct = value
              else:
                  logger.warning(
                      "Skipped invalid `jpeg_compression_fastdct` value!"
                  )
              del options["jpeg_compression_fastdct"]  # clean
  
          if "jpeg_compression_fastupsample" in options:
              value = options["jpeg_compression_fastupsample"]
              # enable jpeg  fastupsample
              if isinstance(value, bool):
                  self.__jpeg_compression_fastupsample = value
              else:
                  logger.warning(
                      "Skipped invalid `jpeg_compression_fastupsample` value!"
                  )
              del options["jpeg_compression_fastupsample"]  # clean
  
          if "frame_size_reduction" in options:
              value = options["frame_size_reduction"]
              if (
                  isinstance(value, (int, float))
                  and value >= 0
                  and value <= 90
              ):
                  self.__frame_size_reduction = value
              else:
                  logger.warning(
                      "Skipped invalid `frame_size_reduction` value!"
                  )
              del options["frame_size_reduction"]  # clean
  
          if "custom_video_endpoint" in options:
              value = options["custom_video_endpoint"]
              if (
                  value
                  and isinstance(value, str)
                  and value.strip().isalnum()
              ):
                  custom_video_endpoint = value.strip()
                  logging and logger.critical(
                      "Using custom video endpoint path: `/{}`".format(
                          custom_video_endpoint
                      )
                  )
              else:
                  logger.warning(
                      "Skipped invalid `custom_video_endpoint` value!"
                  )
              del options["custom_video_endpoint"]  # clean
  
          if "custom_data_location" in options:
              value = options["custom_data_location"]
              if value and isinstance(value, str):
                  assert os.access(
                      value, os.W_OK
                  ), "[WebGear:ERROR] :: Permission Denied!, cannot write WebGear data-files to '{}' directory!".format(
                      value
                  )
                  assert os.path.isdir(
                      os.path.abspath(value)
                  ), "[WebGear:ERROR] :: `custom_data_location` value must be the path to a directory and not to a file!"
                  custom_data_location = os.path.abspath(value)
              else:
                  logger.warning(
                      "Skipped invalid `custom_data_location` value!"
                  )
              del options["custom_data_location"]  # clean
  
          if "overwrite_default_files" in options:
              value = options["overwrite_default_files"]
              if isinstance(value, bool):
                  overwrite_default = value
              else:
                  logger.warning(
                      "Skipped invalid `overwrite_default_files` value!"
                  )
              del options["overwrite_default_files"]  # clean
  
          if "enable_infinite_frames" in options:
              value = options["enable_infinite_frames"]
              if isinstance(value, bool):
                  self.__enable_inf = value
              else:
                  logger.warning(
                      "Skipped invalid `enable_infinite_frames` value!"
                  )
              del options["enable_infinite_frames"]  # clean
  
      # check if disable Data-Files Auto-Generation WorkFlow is disabled
      if not self.__skip_generate_webdata:
          # check if custom data path is specified
          if custom_data_location:
              data_path = generate_webdata(
                  custom_data_location,
                  c_name="webgear",
                  overwrite_default=overwrite_default,
                  logging=logging,
              )
          else:
              # otherwise generate suitable path
              data_path = generate_webdata(
                  os.path.join(expanduser("~"), ".vidgear"),
                  c_name="webgear",
                  overwrite_default=overwrite_default,
                  logging=logging,
              )
  
          # log it
          self.__logging and logger.debug(
              "`{}` is the default location for saving WebGear data-files.".format(
                  data_path
              )
          )
          # define Jinja2 templates handler
          self.__templates = Jinja2Templates(
              directory="{}/templates".format(data_path)
          )
          # define routing tables
          self.routes = [
              Route("/", endpoint=self.__homepage),
              Route(
                  "/{}".format(
                      custom_video_endpoint
                      if custom_video_endpoint
                      else "video"
                  ),
                  endpoint=self.__video,
              ),
              Mount(
                  "/static",
                  app=StaticFiles(
                      directory="{}/static".format(data_path)
                  ),
                  name="static",
              ),
          ]
      else:
          # log it
          self.__logging and logger.critical(
              "WebGear Data-Files Auto-Generation WorkFlow has been manually disabled."
          )
          # define routing tables
          self.routes = [
              Route(
                  "/{}".format(
                      custom_video_endpoint
                      if custom_video_endpoint
                      else "video"
                  ),
                  endpoint=self.__video,
              ),
          ]
          # log exceptions
          self.__logging and logger.warning(
              "Only `/video` route is available for this instance."
          )
  
      # define custom exception handlers
      self.__exception_handlers = {
          404: self.__not_found,
          500: self.__server_error,
      }
      # define middleware support
      self.middleware = []
      # Handle video source
      if source is None:
          self.config = {"generator": None}
          self.__stream = None
      else:
          # define stream with necessary params
          self.__stream = VideoGear(
              enablePiCamera=enablePiCamera,
              stabilize=stabilize,
              source=source,
              camera_num=camera_num,
              stream_mode=stream_mode,
              backend=backend,
              colorspace=colorspace,
              resolution=resolution,
              framerate=framerate,
              logging=logging,
              time_delay=time_delay,
              **options,
          )
          # define default frame generator in configuration
          self.config = {"generator": self.__producer}
  
      # log if specified
      if self.__logging:
          if source is None:
              logger.warning(
                  "Given source is of NoneType. Therefore, JPEG Frame-Compression is disabled!"
              )
          else:
              logger.debug(
                  "Enabling JPEG Frame-Compression with Colorspace:`{}`, Quality:`{}`%, Fastdct:`{}`, and Fastupsample:`{}`.".format(
                      self.__jpeg_compression_colorspace,
                      self.__jpeg_compression_quality,
                      "enabled"
                      if self.__jpeg_compression_fastdct
                      else "disabled",
                      "enabled"
                      if self.__jpeg_compression_fastupsample
                      else "disabled",
                  )
              )
  
      # copying original routing tables for further validation
      self.__rt_org_copy = self.routes[:]
      # initialize blank frame
      self.blank_frame = None
      # keeps check if producer loop should be running
      self.__isrunning = True
  
file_path: vidgear/vidgear/gears/asyncio/webgear.py
imports: import os
  import asyncio
  import inspect
  import numpy
  import logging
  from os.path import expanduser
  from .helper import reducer, generate_webdata, create_blank_frame
  from ..helper import logger_handler, retrieve_best_interpolation, import_dependency_safe, logcurr_vidgear_ver
  from ..videogear import VideoGear
label_desc: Use image node_id 1 to create a new WebGear object  with example: obj = WebGear(enablePiCamera, stabilize, source, camera_num, stream_mode, backend, colorspace, resolution, framerate, logging, time_delay)