mendevi.database.meta.merge_extractors

mendevi.database.meta.merge_extractors(labels: set[str], alias: dict[str, str] | None = None, select: str | None = None, *, return_callable: bool = False) → tuple[set[str], list[str] | Callable]

Return the source code of the function that extracts all variables.

Parameters

labelsset[str]

The returned variable names. These are the keys to the output dictionary.

aliasdict[str, str], optional

To each new name, associate an expression. By default, the label extraction method is defined by the function get_extractor(). This list of aliases allows any unknown key to define a customised access method.

selectstr, optional

A Python Boolean expression that raises a RejectError exception if it evaluates to False.

return_callableboolean, default=False

By default, returns the source code of the function. If this option is set to True, an executable function is returned.

Returns

lbls_atomset[str]

The name of the primary value to be extracted for the SQL query.

funclist[str] or callable

The function that consumes a line from the SQL query, and returns the dictionary of extracted values.

Examples

>>> from mendevi.database.meta import merge_extractors
>>> labels = {"enc_scenario", "rate", "power", "foo"}
>>> alias = {"foo": "(abaca, energy)", "abaca": "(hostname, nbr_frames, enc_scenario)"}
>>> select = '"yeti" not in hostname'
>>> print("\n".join(merge_extractors(labels, alias, select)[1]))
def line_extractor(raw: dict[str]) -> dict[str]:
    """Get the labels: enc_scenario, foo, power, rate.

    Causality constraint:
        * abaca -> foo
        * enc_scenario -> abaca
        * encode_cmd -> enc_scenario
        * energy -> foo
        * energy -> power
        * frames -> nbr_frames
        * hostname -> abaca
        * hostname -> enc_scenario
        * hostname -> reject
        * nbr_frames -> abaca
        * powers -> energy
        * powers -> power
        * size -> rate
        * video_duration -> rate
        * video_name -> enc_scenario
    """
    # reject wrong line
    hostname = extract.extract_hostname(raw)  # The machine name
    reject = not ("yeti" not in hostname)  # not ("yeti" not in hostname)
    if reject:
        msg = "this line must be filtered"
        raise RejectError(msg)

    # extract data
    video_name = extract.extract_video_name(raw)  # Full video basename
    encode_cmd = extract.extract_encode_cmd(raw)  # The ffmpeg command used for encoding
    enc_scenario = f"cmd: {encode_cmd}, video_name: {video_name}, hostname: {hostname}"  # Un...
    frames = extract.extract_frames(raw)  # The metadata of each frame
    powers = extract.extract_powers(raw)  # The interval duration and the average power in ea...
    nbr_frames = None if frames is None else len(frames)  # The real number of frames of the ...
    energy = None if powers is None else float((powers[0] * powers[1]).sum())  # Total energy...
    abaca = (hostname, nbr_frames, enc_scenario)  # alias
    foo = (abaca, energy)  # alias
    power = None if energy is None or powers is None else energy / float(powers[0].sum())  # ...
    video_duration = extract.extract_video_duration(raw)  # Video duration in seconds
    size = extract.extract_video_size(raw)  # The total video file size in bytes
    rate = None if size is None or video_duration is None else 8.0 * size / video_duration  #...

    return {
        'enc_scenario': enc_scenario,
        'foo': foo,
        'power': power,
        'rate': rate,
    }
>>>