Mixins

mixins

Collection of classes that can be used as Mixins with other base classes.

Classes such as TorrentFile, TorrentFilev2, and all Hasher classes can use the progress bar mixin. And any class is eligible to use the callback mixin.

CbMixin

Mixin class to set a callback hook during procedure.

cb(*args, **kwargs) classmethod

Do nothing.

Source code in torrentfile\mixins.py

@classmethod
def cb(cls, *args, **kwargs):
    """Do nothing."""

set_callback(func) classmethod

Assign a callback to the Hashing class.

PARAMETER	DESCRIPTION
func	the callback function TYPE: Callable

Source code in torrentfile\mixins.py

@classmethod
def set_callback(cls, func):
    """
    Assign a callback to the Hashing class.

    Parameters
    ----------
    func : Callable
        the callback function
    """
    cls.cb = func  # pragma: nocover

ProgMixin

Progress bar mixin class.

Displays progress of hashing individual files, usefull when hashing really big files.

NoProg

Stand-in object for when progress mode is set to 0.

update(value) staticmethod

Return the value.

PARAMETER	DESCRIPTION
value	the input and output

RETURNS	DESCRIPTION
int	same as input

Source code in torrentfile\mixins.py

@staticmethod
def update(value):
    """
    Return the value.

    Parameters
    ----------
    value: int
        the input and output

    Returns
    -------
    int :
        same as input
    """
    return value

get_progress_tracker(total: int, message: str)

Return the progress bar object for external management.

PARAMETER	DESCRIPTION
total	total size to track TYPE: int
message	prompt message for what is being tracked TYPE: str

RETURNS	DESCRIPTION
ProgressBar	progress bar object instance

Source code in torrentfile\mixins.py

def get_progress_tracker(self, total: int, message: str):
    """Return the progress bar object for external management.

    Parameters
    ----------
    total: int
        total size to track
    message: str
        prompt message for what is being tracked

    Returns
    -------
    ProgressBar
        progress bar object instance
    """
    if total < 0:
        return self.NoProg()
    return ProgressBar.new(total, message)

ProgressBar(total: int, title: str, length: int, unit: str, start: int)

Holds the state and details of the terminal progress bars.

PARAMETER	DESCRIPTION
total	the total amount to be accumulated. TYPE: int
title	the subject of the progress tracker TYPE: str
length	the width of the progress bar TYPE: int
unit	the text representation incremented TYPE: str
start	column where the progress bar should be drawn TYPE: int

Construct the progress bar object and store state of it’s properties.

Source code in torrentfile\mixins.py

def __init__(self, total: int, title: str, length: int, unit: str,
             start: int):
    """
    Construct the progress bar object and store state of it's properties.
    """
    debug_is_on()
    self.total = total
    self.start = start
    self.length = length
    self.fill = chr(9608)
    self.empty = chr(9617)
    self.state = 0
    self.unit = unit
    self.show_total = total
    if not unit:
        self.unit = ""  # pragma: nocover
    elif unit == "bytes":
        if self.total > 1_000_000_000:
            self.show_total = self.total / (2**30)
            self.unit = "GiB"
        elif self.total > 1_000_000:
            self.show_total = self.total / 1048576
            self.unit = "MiB"
        elif self.total > 10000:
            self.show_total = self.total / 1024
            self.unit = "KiB"
    self.suffix = f"/{self.show_total:.02f} {self.unit}"
    title = str(title)
    if len(title) > start:
        title = title[:start - 1]  # pragma: nocover
    padding = (start - len(title)) * " "
    self.prefix = "".join([title, padding])

close_out() staticmethod

Finalize the last bits of progress bar.

Increment the terminal by one line leaving the progress bar in place, and deleting the progress bar object to clear a space for the next one.

Source code in torrentfile\mixins.py

@staticmethod
def close_out():
    """
    Finalize the last bits of progress bar.

    Increment the terminal by one line leaving the progress bar in place,
    and deleting the progress bar object to clear a space for the next one.
    """
    sys.stdout.flush()
    sys.stdout.write("\n")

get_progress() -> str

Return the size of the filled portion of the progress bar.

RETURNS	DESCRIPTION
str	the progress bar characters TYPE: str

Source code in torrentfile\mixins.py

def get_progress(self) -> str:
    """
    Return the size of the filled portion of the progress bar.

    Returns
    -------
    str :
        the progress bar characters
    """
    if self.state >= self.total:
        fill = self.length
    else:
        fill = math.ceil((self.state / self.total) * self.length)
    empty = self.length - fill
    contents = (self.fill * fill) + (self.empty * empty)
    pbar = ["|", green(contents), "| "]
    if self.unit == "GiB":
        state = self.state / (2**30)
    elif self.unit == "MiB":
        state = self.state / 1048576
    elif self.unit == "KiB":
        state = self.state / 1024
    else:
        state = self.state
    pbar.append(f"{state:.02f}")
    return "".join(pbar)

new(total: int, path: str, length: int = 50, unit: str = 'bytes') classmethod

Generate a new progress bar for the given file path.

PARAMETER	DESCRIPTION
total	the total amount of units accumulating towards. TYPE: int
path	path to file being hashed. TYPE: str
length	the number of characters of the actual progress bar. TYPE: int DEFAULT: 50
unit	the text representation of the value being measured. TYPE: str DEFAULT: 'bytes'

Source code in torrentfile\mixins.py

@classmethod
def new(cls, total: int, path: str, length: int = 50, unit: str = "bytes"):
    """
    Generate a new progress bar for the given file path.

    Parameters
    ----------
    total : int
        the total amount of units accumulating towards.
    path : str
        path to file being hashed.
    length : int
        the number of characters of the actual progress bar.
    unit : str
        the text representation of the value being measured.
    """
    title = path
    width = shutil.get_terminal_size().columns
    if len(str(title)) >= width // 2:
        parts = list(Path(title).parts)
        while (len("//".join(parts)) > (width // 2)) and (len(parts) > 0):
            del parts[0]
        if parts:
            title = os.path.join(*parts)
        else:
            title = os.path.basename(path)  # pragma: nocover
    length = min(length, width // 2)
    start = width - int(length * 1.5)
    return cls(total, title, length, unit, start)

update(val: int)

Update progress bar.

Using the value provided, increment the progress bar by that value.

PARAMETER	DESCRIPTION
val	the number of bytes count the progress bar should increase. TYPE: int

Source code in torrentfile\mixins.py

def update(self, val: int):
    """
    Update progress bar.

    Using the value provided, increment the progress bar by that value.

    Parameters
    ----------
    val : int
        the number of bytes count the progress bar should increase.
    """
    self.state += val
    pbar = self.get_progress()
    output = f"{self.prefix}{pbar}{self.suffix}\r"
    sys.stdout.write(output)
    sys.stdout.flush()

waiting(msg: str, flag: list, timeout: int = 20)

Show loading message while thread completes processing.

PARAMETER	DESCRIPTION
msg	Message string printed before the progress bar TYPE: str
flag	Once flag is filled exit loop TYPE: list
timeout	max amount of time to run the function. TYPE: int DEFAULT: 20

Source code in torrentfile\mixins.py

def waiting(msg: str, flag: list, timeout: int = 20):
    """
    Show loading message while thread completes processing.

    Parameters
    ----------
    msg : str
        Message string printed before the progress bar
    flag : list
        Once flag is filled exit loop
    timeout : int
        max amount of time to run the function.
    """
    then = time.time()
    codes, fill = list(range(9617, 9620)), chr(9619)
    size = idx = 0
    total = shutil.get_terminal_size().columns - len(msg) - 20

    def output(text: str):
        """
        Print parameter message to the console.

        Parameters
        ----------
        text : str
            output message
        """
        sys.stdout.write(text)
        sys.stdout.flush()

    output("\n")
    time.sleep(0.16)
    while len(flag) == 0:
        time.sleep(0.16)
        filled = (fill * size) + chr(codes[idx]) + (" " * (total - size))
        output(f"{msg}: {filled}\r")
        idx = idx + 1 if idx + 1 < len(codes) else 0
        size = size + 1 if size < total else 0
        if time.time() - then > timeout:
            break
    output("\n")