ddelange

import re
from datetime import timedelta

# Ordered case-insensitive duration strings, loosely based on the [GEP-2257](https://gateway-api.sigs.k8s.io/geps/gep-2257/) spec
# Discrepancies between this pattern and GEP-2257 duration strings:
# - this pattern accepts units `w|d|h|m|s|ms|[uµ]s` (all units supported by the datetime.timedelta constructor), GEP-2257 accepts only `h|m|s|ms`
# - this pattern allows for optional whitespace around the units, GEP-2257 does not
# - this pattern is compiled to be case-insensitive, GEP-2257 expects lowercase units
# - this pattern expects ordered (descending) units, GEP-2257 allows arbitrary order
# - this pattern does not allow duplicate unit occurrences, GEP-2257 does

ddelange

# pip install click
# python itunes_xml_to_m3u.py --help
import logging
import plistlib
import re
import typing as tp
from pathlib import Path
from urllib.parse import unquote

import click

ddelange

#!/usr/bin/env bash

set -euxo pipefail

video_in="video_in.mkv"
video_out="video_out.mp4"

# full metadata payload: $(ffprobe -loglevel error -print_format json -show_format -show_streams "${video_in}")
is_avc=$(ffprobe -loglevel error -select_streams v:0 -show_entries stream=is_avc -of default=noprint_wrappers=1:nokey=1 "${video_in}")

ddelange

from collections import deque
from concurrent.futures import ProcessPoolExecutor as _ProcessPoolExecutor
from multiprocessing.shared_memory import SharedMemory

import numpy as np


class ProcessPoolExecutor(_ProcessPoolExecutor):
    """Subclass with a lazy consuming imap method."""

ddelange

#!/usr/bin/env bash
# original gist from pescobar/build-git.sh
# changes by LaggAt:
# * to be usable on Raspbian / tested RPi3 and 
# * for automatic depency resolving
# changes by ivan-c:
# * add `apt-get update`
# changes by ddelange:
# * add `set -euxo pipefail`
# * remove `--allow-downgrades`

ddelange

import logging
from collections import deque
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from subprocess import PIPE, CalledProcessError, CompletedProcess, Popen


def stream_command(
    args,
    *,

ddelange

from collections import deque
from concurrent.futures import ThreadPoolExecutor as _ThreadPoolExecutor


class ThreadPoolExecutor(_ThreadPoolExecutor):
    """Subclass with a lazy consuming imap method."""

    def imap(self, fn, *iterables, timeout=None, queued_tasks_per_worker=2):
        """Ordered imap that lazily consumes iterables ref https://gist.github.com/ddelange/c98b05437f80e4b16bf4fc20fde9c999."""
        futures, maxlen = deque(), self._max_workers * (queued_tasks_per_worker + 1)

ddelange

# pip install smart_open[s3]
from collections import deque
from concurrent.futures import ThreadPoolExecutor as _ThreadPoolExecutor
from functools import partial
from typing import Callable, Dict, Optional, Iterable, Iterator, Sequence

import boto3
import botocore
import smart_open

ddelange

from itertools import count
from os import environ

import dominate
import requests

GITHUB_API_URL = environ["GITHUB_API_URL"]
GITHUB_REPOSITORY = environ["GITHUB_REPOSITORY"]

links = {}

ddelange

# pip install sh pycountry
import re
from pathlib import Path

import pycountry
from sh import ffmpeg


def mux_mp4_subs(inp, outp, *subs, _cwd=None, **subs_map):
    """Mux multiple subtitle files into an mp4 file.