Helper Methods
reducer¶
Asynchronous method that reduces frame size by given percentage.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
frame | numpy.ndarray | inputs numpy array(frame). | None |
percentage | int/float | inputs size-reduction percentage. | 0 |
interpolation | int | Change resize interpolation. | 4 |
Returns: A reduced numpy ndarray array.
Source code in vidgear/gears/asyncio/helper.py
async def reducer(
frame: NDArray = None,
percentage: Union[int, float] = 0,
interpolation: int = cv2.INTER_LANCZOS4,
) -> NDArray:
"""
## reducer
Asynchronous method that reduces frame size by given percentage.
Parameters:
frame (numpy.ndarray): inputs numpy array(frame).
percentage (int/float): inputs size-reduction percentage.
interpolation (int): Change resize interpolation.
**Returns:** A reduced numpy ndarray array.
"""
# check if frame is valid
if frame is None:
raise ValueError("[Helper:ERROR] :: Input frame cannot be NoneType!")
# check if valid reduction percentage is given
if not (percentage > 0 and percentage < 90):
raise ValueError(
"[Helper:ERROR] :: Given frame-size reduction percentage is invalid, Kindly refer docs."
)
if not (isinstance(interpolation, int)):
raise ValueError(
"[Helper:ERROR] :: Given interpolation is invalid, Kindly refer docs."
)
# grab the frame size
(height, width) = frame.shape[:2]
# calculate the ratio of the width from percentage
reduction = ((100 - percentage) / 100) * width
ratio = reduction / float(width)
# construct the dimensions
dimensions = (int(reduction), int(height * ratio))
# return the resized frame
return cv2.resize(frame, dimensions, interpolation=interpolation)
create_blank_frame¶
Create blank frames of given frame size with text
Parameters:
Name | Type | Description | Default |
---|---|---|---|
frame | numpy.ndarray | inputs numpy array(frame). | None |
text | str | Text to be written on frame. | '' |
Returns: A reduced numpy ndarray array.
Source code in vidgear/gears/asyncio/helper.py
def create_blank_frame(
frame: NDArray = None, text: str = "", logging: bool = False
) -> NDArray:
"""
## create_blank_frame
Create blank frames of given frame size with text
Parameters:
frame (numpy.ndarray): inputs numpy array(frame).
text (str): Text to be written on frame.
**Returns:** A reduced numpy ndarray array.
"""
# check if frame is valid
if frame is None or not (isinstance(frame, np.ndarray)):
raise ValueError("[Helper:ERROR] :: Input frame is invalid!")
# grab the frame size
(height, width) = frame.shape[:2]
# create blank frame
blank_frame = np.zeros(frame.shape, frame.dtype)
# setup text
if text and isinstance(text, str):
if logging:
logger.debug("Adding text: {}".format(text))
# setup font
font = cv2.FONT_HERSHEY_SCRIPT_COMPLEX
# get boundary of this text
fontScale = min(height, width) / (25 / 0.25)
textsize = cv2.getTextSize(text, font, fontScale, 5)[0]
# get coords based on boundary
textX = (width - textsize[0]) // 2
textY = (height + textsize[1]) // 2
# put text
cv2.putText(
blank_frame, text, (textX, textY), font, fontScale, (125, 125, 125), 6
)
# return frame
return blank_frame
generate_webdata¶
Auto-Generates, and Auto-validates default data for WebGear and WebGear_RTC APIs.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path | string | path for generating data | required |
c_name | string | class name that is generating files | 'webgear' |
overwrite_default | boolean | overwrite existing data or not? | False |
logging | bool | enables logging for its operations | False |
Returns: A valid data path as string.
Source code in vidgear/gears/asyncio/helper.py
def generate_webdata(
path: str,
c_name: str = "webgear",
overwrite_default: bool = False,
logging: bool = False,
) -> str:
"""
## generate_webdata
Auto-Generates, and Auto-validates default data for WebGear and WebGear_RTC APIs.
Parameters:
path (string): path for generating data
c_name (string): class name that is generating files
overwrite_default (boolean): overwrite existing data or not?
logging (bool): enables logging for its operations
**Returns:** A valid data path as string.
"""
# check if path corresponds to vidgear only
if os.path.basename(path) != ".vidgear":
path = os.path.join(path, ".vidgear")
# generate parent directory
path = os.path.join(path, c_name)
mkdir_safe(path, logging=logging)
# self-generate dirs
template_dir = os.path.join(path, "templates") # generates HTML templates dir
static_dir = os.path.join(path, "static") # generates static dir
# generate js & css static and favicon img subdirs
js_static_dir = os.path.join(static_dir, "js")
css_static_dir = os.path.join(static_dir, "css")
favicon_dir = os.path.join(static_dir, "img")
mkdir_safe(static_dir, logging=logging)
mkdir_safe(template_dir, logging=logging)
mkdir_safe(js_static_dir, logging=logging)
mkdir_safe(css_static_dir, logging=logging)
mkdir_safe(favicon_dir, logging=logging)
# check if overwriting is enabled
if overwrite_default or not validate_webdata(
template_dir, ["index.html", "404.html", "500.html"]
):
logger.critical(
"Overwriting existing {} data-files with default data-files from the server!".format(
c_name.capitalize()
)
if overwrite_default
else "Failed to detect critical {} data-files: index.html, 404.html & 500.html!".format(
c_name.capitalize()
)
)
# download default files
logging and logger.info(
"Downloading default data-files from the Gitlab Server: {}".format(
"https://gitlab.com/abhiTronix/vidgear-vitals"
)
)
download_webdata(
template_dir,
c_name=c_name,
files=["index.html", "404.html", "500.html", "base.html"],
logging=logging,
)
download_webdata(
css_static_dir, c_name=c_name, files=["custom.css"], logging=logging
)
download_webdata(
js_static_dir, c_name=c_name, files=["custom.js"], logging=logging,
)
download_webdata(
favicon_dir, c_name=c_name, files=["favicon-32x32.png"], logging=logging
)
else:
# validate important data-files
if logging:
logger.debug("Found valid WebGear data-files successfully.")
return path
download_webdata¶
Downloads given list of files for WebGear and WebGear_RTC APIs(if not available) from GitHub/Gitlab Servers, and also Validates them.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path | string | path for downloading data | required |
c_name | string | class name that is generating files | 'webgear' |
files | list | list of files to be downloaded | [] |
logging | bool | enables logging for its operations | False |
Returns: A valid path as string.
Source code in vidgear/gears/asyncio/helper.py
def download_webdata(
path: str, c_name: str = "webgear", files: list = [], logging: bool = False
) -> str:
"""
## download_webdata
Downloads given list of files for WebGear and WebGear_RTC APIs(if not available) from GitHub/Gitlab Servers,
and also Validates them.
Parameters:
path (string): path for downloading data
c_name (string): class name that is generating files
files (list): list of files to be downloaded
logging (bool): enables logging for its operations
**Returns:** A valid path as string.
"""
basename = os.path.basename(path)
if logging:
logger.debug("Downloading {} data-files at `{}`".format(basename, path))
# list all registered urls
reg_urls = [
"https://gitlab.com/abhiTronix/vidgear-vitals/-/raw/main",
"https://raw.githubusercontent.com/abhiTronix/vidgear-vitals/main",
]
# create session
with requests.Session() as http:
for url in reg_urls:
try:
for file in files:
# get filename
file_name = os.path.join(path, file)
# get URL
file_url = "{}/{}{}/{}/{}".format(
url,
c_name,
"/static" if basename != "templates" else "",
basename,
file,
)
# download and write file to the given path
logging and logger.debug(
"Downloading {} data-file: {}.".format(basename, file)
)
with open(file_name, "wb") as f:
# setup retry strategy
retries = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
# Mount it for https usage
adapter = TimeoutHTTPAdapter(timeout=2.0, max_retries=retries)
http.mount("https://", adapter)
response = http.get(file_url, stream=True)
response.raise_for_status()
total_length = (
response.headers.get("content-length")
if "content-length" in response.headers
else len(response.content)
)
assert not (
total_length is None
), "[Helper:ERROR] :: Failed to retrieve files, check your Internet connectivity!"
bar = tqdm(total=int(total_length), unit="B", unit_scale=True)
for data in response.iter_content(chunk_size=256):
f.write(data)
if len(data) > 0:
bar.update(len(data))
bar.close()
except AssertionError as e:
# raise if connection error
raise e
except Exception as e:
# log error
logger.exception(str(e))
# log event if necessary
url != reg_urls[1] and logger.error(
"Download failed for Gitlab Server! Retrying from GitHub Server: {}".format(
url, "https://github.com/abhiTronix/vidgear-vitals"
)
)
else:
# break otherwise
break
if logging:
logger.debug("Verifying downloaded data:")
if validate_webdata(path, files=files, logging=logging):
if logging:
logger.info("Successful!")
return path
else:
raise RuntimeError(
"[Helper:ERROR] :: Failed to download required {} data-files at: {}, Check your Internet connectivity!".format(
basename, path
)
)
validate_auth_keys¶
Validates, and also maintains downloaded list of files.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path | string | path of downloaded files | required |
files | list | list of files to be validated | [] |
logging | bool | enables logging for its operations | False |
Returns: A boolean value, confirming whether tests passed, or not?.
Source code in vidgear/gears/asyncio/helper.py
def validate_webdata(path: str, files: list = [], logging: bool = False) -> bool:
"""
## validate_auth_keys
Validates, and also maintains downloaded list of files.
Parameters:
path (string): path of downloaded files
files (list): list of files to be validated
logging (bool): enables logging for its operations
**Returns:** A boolean value, confirming whether tests passed, or not?.
"""
# check if valid path or directory empty
if not (os.path.exists(path)) or not (os.listdir(path)):
return False
files_buffer = []
# loop over files
for file in os.listdir(path):
if file in files:
files_buffer.append(file) # store them
# return results
if len(files_buffer) < len(files):
if logging:
logger.warning(
"`{}` file(s) missing from data-files!".format(
" ,".join(list(set(files_buffer) ^ set(files)))
)
)
return False
else:
return True