60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
"""Common utilities"""
|
|
from asyncio import gather
|
|
from datetime import datetime
|
|
from os.path import expandvars
|
|
from pathlib import Path
|
|
from re import compile as re_compile
|
|
from typing import Awaitable, cast
|
|
|
|
from click import progressbar
|
|
from xdg import xdg_config_home
|
|
|
|
|
|
def get_default_screenshot_directory() -> Path:
|
|
"""Return the default folder where to store screenshot (XDG pictures dir by default)"""
|
|
pictures_dir_re = re_compile(r'\w*XDG_PICTURES_DIR\w*\="(?P<path>.*)"\w*$')
|
|
pictures_dir = None
|
|
with open(
|
|
Path(xdg_config_home()) / "user-dirs.dirs", encoding="utf-8"
|
|
) as user_dirs:
|
|
for line in user_dirs.readlines():
|
|
match = pictures_dir_re.match(line)
|
|
if match is None:
|
|
continue
|
|
|
|
path_string = match.group("path")
|
|
path_string = expandvars(path_string)
|
|
pictures_dir = Path(path_string)
|
|
break
|
|
|
|
if pictures_dir is None:
|
|
pictures_dir = Path.home()
|
|
|
|
return pictures_dir / datetime.now().strftime("%Y-%m-%d - %H-%m-%S")
|
|
|
|
|
|
TaskListType = list[tuple[str, Awaitable[None]]]
|
|
|
|
|
|
async def report_progress(label: str, task_list: TaskListType) -> None:
|
|
"""Show a progress bar for a long running task list"""
|
|
iterator = iter(task_list)
|
|
|
|
with progressbar(
|
|
length=len(task_list),
|
|
label=label,
|
|
item_show_func=lambda it: cast(str, it),
|
|
) as progress:
|
|
|
|
async def worker() -> None:
|
|
try:
|
|
while True:
|
|
item_name, task = next(iterator)
|
|
progress.update(1, current_item=item_name) # type: ignore
|
|
await task
|
|
except StopIteration:
|
|
pass
|
|
|
|
workers = [worker() for _ in range(0, 5)]
|
|
await gather(*workers)
|