dol snippets

[thorwhalen]

A collection of random dol code snippets.

[thorwhalen]

Copy data from one store to another

src = ... # make source store
targ = ... # make target store 
targ.update(src)

Note that here, targ will use the keys and values of src as they're provided.
If some specific key or value transformations need to take place, they must be specified in the construction of src and/or targ.

[thorwhalen]

Copy pickled data into gzipped csv data

We'll work with dictionaries instead of files here, so we can test more easily.

Say you have a source backend that has pickles of some lists-of-lists-of-strings, using the .pkl extension,
and you want to copy this data to a target backend, but saving them as gzipped csvs with the csv.gz extension.

import pickle

src_backend = {
    'file_1.pkl': pickle.dumps([['A', 'B', 'C'], ['one', 'two', 'three']]),
    'file_2.pkl': pickle.dumps([['apple', 'pie'], ['one', 'two'], ['hot', 'cold']]),
}
targ_backend = dict()

Here's how you can do it:

from dol import ValueCodecs, KeyCodecs, Pipe

# decoder here will unpickle data and remove remove the .pkl extension from the key
src_wrap = Pipe(KeyCodecs.suffixed('.pkl'), ValueCodecs.pickle())

# encoder here will convert the lists to csv string, the string into bytes, and the bytes will be gzipped. 
# ... also, we'll add .csv.gz on write.
targ_wrap = Pipe(
    KeyCodecs.suffixed('.csv.gz'), 
    ValueCodecs.csv() + ValueCodecs.str_to_bytes() + ValueCodecs.gzip()
)

# Let's wrap our backends:
src = src_wrap(src_backend)
targ = targ_wrap(targ_backend)

# and copy src over to targ
print(f"Before: {list(targ_backend)=}")
targ.update(src)
print(f"After: {list(targ_backend)=}")

From the point of view of src and targ, you see the same thing:

assert list(src) == list(targ) == ['file_1', 'file_2']
assert (
    src['file_1'] 
    == targ['file_1']
    == [['A', 'B', 'C'], ['one', 'two', 'three']]
)

But the backend of targ is different:

src_backend['file_1.pkl']
# b'\x80\x04\x95\x19\x00\x00\x00\x00\x00\x00\x00]\x94(]\x94(K\x01K\x02K\x03e]\x94(K\x04K\x05K\x06ee.'
targ_backend['file_1.csv.gz']
# b'\x1f\x8b\x08\x00*YWe\x02\xff3\xd41\xd21\xe6\xe52\xd11\xd51\xe3\xe5\x02\x00)4\x83\x83\x0e\x00\x00\x00'

[thorwhalen]

A path getter with Pipe

from operator import methodcaller, attrgetter, itemgetter
from dol import Pipe

def simple_path_getter(*keys, getter=itemgetter):
    """
    Make a function that gets a path of keys from an object.

    >>> path_getter = simple_path_getter('a', 'b', 'c')
    >>> path_getter({'a': {'b': {'c': 1}}})
    1
    """
    return Pipe(*map(getter, keys))

# a function that does obj['a']['b']['c']
path_getter = simple_path_getter('a', 'b', 'c')
d = {'a': {'b': {'c': 1}}}
assert path_getter(d) == 1

# function that does obj.x  (attrgetter('x') would do, but just for illustration)
x_attr_getter = simple_path_getter('x', getter=attrgetter)
from collections import namedtuple
Point = namedtuple('Point', ['x', 'y'])
p = Point(1, 2)
assert x_attr_getter(p) == 1

[thorwhalen]

Renaming files to a normalized form

tldr;

# files: mapping to your files (names and content)
# rename_mapping: your old_name: new_name dict

for old, new in rename_mapping.items():
    # gets the contents of the file (and deletes the (old) file) and writes the contents under the new file name
    files[new] = files.pop(old)  

Having AI make rename_mapping

Here, we'll demo dol, but also involve the oa.
You don't need it to rename your files, but I thought it would be cool to have AI make the rename_mapping dict for you.

The oa function we'll involve is oa.prompt_function, which takes a prompt (and possibly other arguments) and makes a python function out of it.

You could do this for your own specific case, or even prompt engineer a nice general "extract-and-generate" name normalizing prompt. Here, though, I'm applying this to book filenames.

Getting a old_name->new_name mapping:

import oa
import json 

normalize_book_filenames = oa.prompt_function(
    """
In the following list, I want you to parse out the title, author, and year of the file 
name, and give me normalized filenames that have the format 
"TITLE - AUTHOR - YEAR" (and if the filename has extension, keep the same extension.
If you can't parse out the title, author, and year, just replace with "UNKNOWN TITLE",
"UNKNOWN AUTHOR", and "UNKNOWN YEAR" respectively.

Please give me your answer in the form of a json with keys being the original file names
and the normalized filenames as the values. 

{filenames}
""",
    prompt_func_kwargs=dict(
        model='gpt-4o-mini',  # some others don't work with json_object
        response_format={'type': 'json_object'}
    ),
    egress=json.loads,
)

Renaming my files.
I'll use a dict here, so you can test it yourself in the safety of your own RAM (no actual file operations here!)

file_contents_placeholder = b"These would normally be the contents of the file"
files = {
    "David Baum's book: Generative AI and LLMs for Dummies (2024) - libgen.li.pdf": file_contents_placeholder,
    "Prompt Engineering Using ChatGPT_ Crafting Effective Interactions and Building GPT Apps (MLI Generative AI Series 2019) by Mehrzad Tabatabaian PhD PEng.pdf": file_contents_placeholder,
    "Critical Thinking and Ethics in the Age of Generative AI in Education (Pedro Noguera, 2022).epub": file_contents_placeholder,
}
rename_mapping = normalize_book_filenames('\n'.join(files))

rename_mapping

{"David Baum's book: Generative AI and LLMs for Dummies (2024) - libgen.li.pdf": 'Generative AI and LLMs for Dummies - David Baum - 2024.pdf',
 'Prompt Engineering Using ChatGPT_ Crafting Effective Interactions and Building GPT Apps (MLI Generative AI Series 2019) by Mehrzad Tabatabaian PhD PEng.pdf': 'Crafting Effective Interactions and Building GPT Apps - Mehrzad Tabatabaian PhD PEng - 2019.pdf',
 'Critical Thinking and Ethics in the Age of Generative AI in Education (Pedro Noguera, 2022).epub': 'Critical Thinking and Ethics in the Age of Generative AI in Education - Pedro Noguera - 2022.epub'}

for old, new in rename_mapping.items():
    # gets the contents of the file (and deletes the (old) file) and writes the contents under the new file name
    files[new] = files.pop(old)  

files

{'Generative AI and LLMs for Dummies - David Baum - 2024.pdf': b'These would normally be the contents of the file',
 'Crafting Effective Interactions and Building GPT Apps - Mehrzad Tabatabaian PhD PEng - 2019.pdf': b'These would normally be the contents of the file',
 'Critical Thinking and Ethics in the Age of Generative AI in Education - Pedro Noguera - 2022.epub': b'These would normally be the contents of the file'}

Note that I'm doing files[new] = files.pop(old) for educational purposes.
It's a little risky, since given the order of operations, if something happens during
the writing, you might lose the original file
In real life, I'd probably do:

for old, new in rename_mapping.items():
    files[new] = files[old]
    del files[old]  # delete the old file once the new file has been written

And just as a matter of exercise (trying to understand each element is helpful for learning) note the one liner solution:

# You need to set up your test again (in case the original filenames have been renamed from the above!)
file_contents_placeholder = b"These would normally be the contents of the file"
files = {
    "David Baum's book: Generative AI and LLMs for Dummies (2024) - libgen.li.pdf": file_contents_placeholder,
    "Prompt Engineering Using ChatGPT_ Crafting Effective Interactions and Building GPT Apps (MLI Generative AI Series 2019) by Mehrzad Tabatabaian PhD PEng.pdf": file_contents_placeholder,
    "Critical Thinking and Ethics in the Age of Generative AI in Education (Pedro Noguera, 2022).epub": file_contents_placeholder,
}

# And then the one line:
list(__import__('itertools').starmap(lambda old, new: files.__setitem__(new, files.pop(old)), rename_mapping.items()))  

Ingress: Transforming the inputs

If you're really picky (or really nice to your users -- see Postel's Law), you might
want to have your AI-enabled function not have to take a string, but be able to take an iterable of filenames for example.
In that case, you could wrap your prompt function with some input-preprocessing code, or have oa.prompt_function do it for you by feeding an ingress argument.

Now, note that the ingress argument expects to get as an input, the dict of arguments (the one that oa.prompt_function will make and feed to it, and should return a dict with those same keys, but with the values transformed.

import oa
import json 
from typing import Iterable

def ensure_newline_separated_string(s):
    if isinstance(s, Iterable) and not isinstance(s, str):
        return '\n'.join(s)
    return s


normalize_book_filenames = oa.prompt_function(
    """
In the following list, I want you to parse out the title, author, and year of the file 
name, and give me normalized filenames that have the format 
"TITLE - AUTHOR - YEAR" (and if the filename has extension, keep the same extension.
If you can't parse out the title, author, and year, just replace with "UNKNOWN TITLE",
"UNKNOWN AUTHOR", and "UNKNOWN YEAR" respectively.

Please give me your answer in the form of a json with keys being the original file names
and the normalized filenames as the values. 

{filenames}
""",
    prompt_func_kwargs=dict(
        model='gpt-4o-mini',  # some others don't work with json_object
        response_format={'type': 'json_object'}
    ),
    ingress=lambda inputs: {'filenames': ensure_newline_separated_string(inputs['filenames'])},
    egress=json.loads,
)

And now you can simply do:

normalize_book_filenames(files)

[thorwhalen]

A Pipe and operations exercise

Write a Pipe that will be equivalent to the lambda function:

readme_of_package_lambda = lambda pkg: os.path.join(os.path.dirname(os.path.dirname(pkg.__file__)), 'README.md')

but without using any lambdas.

Use the following imports:

import os
from functools import partial
from operator import methodcaller, attrgetter
from dol import Pipe

readme_of_package = Pipe(
   ...
)

[thorwhalen]

Answer

import os
from functools import partial
from operator import methodcaller, attrgetter
from dol import Pipe

readme_of_package = Pipe(
    attrgetter('__file__'),
    os.path.dirname,
    os.path.dirname,
    partial(partial, os.path.join),  # partial of partial! Reflect on the uses cases for this!
    methodcaller('__call__', 'README.md'),
)

[thorwhalen]

partial(partial, partial)

See medium post.

The imports you'll need:

import os
from functools import partial

Consider the following, which enables you to make a factory that makes functions that will make absolute paths with a given root folder:

mk_abs_path_maker_lambda = lambda root: lambda *paths: os.path.join(root, *paths)

abs_path = mk_abs_path_maker_lambda('/my/root')
assert abs_path('a', 'b') == '/my/root/a/b'

This can be done this way instead (with the advantage of elegance and pickle-serializability):

mk_abs_path_maker = partial(partial, os.path.join)

abs_path = mk_abs_path_maker('/my/root')
assert abs_path('a', 'b') == '/my/root/a/b'

Our mk_abs_path_maker(root) fixes the first argument of os.path.join to be root.
How can we generalize this "make a function that is defined by fixing the first argument of some other function" pattern?

first_arg_fixer = partial(partial, partial)

# Test:
mk_abs_path_maker = first_arg_fixer_lambda(os.path.join)
abs_path = mk_abs_path_maker('/my/root')
assert abs_path('a', 'b') == '/my/root/a/b'

With lambdas this would be:

first_arg_fixer_lambda = lambda func: lambda first_arg: lambda *args, **kwargs: func(first_arg, *args, **kwargs)

# Test
mk_abs_path_maker_lambda = first_arg_fixer_lambda(os.path.join)
abs_path = mk_abs_path_maker_lambda('/my/root')
assert abs_path('a', 'b') == '/my/root/a/b'

The lambda is harder to read, and not pickle-serializable.

[thorwhalen]

Acquire (download, copy...) multiple content items

See also medium post: A tiny flexible data acquisition python function and gist.

This module provides tools for downloading and managing multiple content items from various sources, such as files or URLs, into a chosen storage system. At its core, the acquire_content function allows users to specify a uri_to_content callable that fetches content—this could range from file readers like open(filepath).read() to web fetchers like requests.get(url).content. The retrieved data can be saved to either a directory, a mutable mapping, or a custom storage function.

The acquire_content function is configurable with a save_condition parameter to validate content before saving, making it adaptable to a variety of data sources. Error handling is also considered; for example, requests can be used to handle HTTP errors gracefully when working with URLs. A flexible decorator approach is demonstrated to simplify function creation, allowing a developer to create specialized content acquisition functions easily, especially useful for repetitive tasks like downloading data science resources or cheat sheets.

import os
import requests
from typing import Dict, Union, MutableMapping, KT, VT, TypeVar, Callable, Any
from functools import partial

DFLT_STORE_DIR = os.environ.get('DFLT_DOL_DOWNLOAD_DIR', '~/Downloads')

URI = VT
Dirpath = str
ContentType = TypeVar('ContentType')
StoreFunc = Callable[[KT, ContentType], None]


def is_not_none(x):
    return x is not None


def acquire_content(
        uri_to_content: Callable[[URI], ContentType],
        uris: Dict[KT, URI] = None, 
        store: Union[Dirpath, MutableMapping, StoreFunc] = DFLT_STORE_DIR,
        *,
        save_condition: Callable[[Any], bool] = is_not_none
    ):
    """
    Downloads and stores content from a given set of URIs.

    uri_to_content is a callable function that takes a URI and returns content. This is usually set to:
    - a function that reads file content, like `open(filepath).read()`
    - a function that fetches URL content, like `requests.get(url).content`
    
    However, here, we demonstrate with a simple string operation (e.g., uppercasing strings) as a substitute 
    to show functionality.

    Note that the uri_to_content function will usually be something giving you the 
    contents of a file or URL. 

    >>> from pathlib import Path
    >>> files_uri_to_content = lambda filepath: Path(filepath).read_text()
    >>> urls_uri_to_content = lambda url: requests.get(url).content  # doctest: +SKIP

    Here, we use a simple `str.upper` to not have to deal with actual IO during tests:
    Also, we'll use a dict as a store, for test simplicity purposes.
    Usually, though, you'll want to use a directory or a MutableMapping as store, 
    or a function that stores content in a specific way.

    >>> store = {}
    >>> uris = {'example1': 'hello', 'example2': 'world'}
    >>> acquire_content(str.upper, uris, store)  # uri_to_content here is str.upper, to simulate content acquisition.
    >>> store
    {'example1': 'HELLO', 'example2': 'WORLD'}

    Note that often you want to just fix the uri_to_content function and sometimes store.
    The acquire_content acts as a function factory for your convenience. If you don't 
    specify uris (but at least specify `uri_to_content`), you get a function that takes 
    uris as the first argument, and stores the content therefrom.

    >>> content_acquirer = acquire_content(str.upper, store=store)  # doctest: +ELLIPSIS
    >>> content_acquirer({'example3': 'foo', 'example4': 'bar'})
    >>> store
    {'example1': 'HELLO', 'example2': 'WORLD', 'example3': 'FOO', 'example4': 'BAR'}
    

    # Examples that would be typical for uri_to_content:
    # acquire_content(lambda filepath: open(filepath, 'rb').read(), uris, store)  # Reads file content +SKIP
    # acquire_content(lambda url: requests.get(url).content, uris, store)  # Fetches URL content +SKIP
    """
    # if uris is None, we're parametrizing the download_content function
    store = ensure_store_func(store)

    if uris is None:
        assert callable(uri_to_content), "uri_to_content must be a callable if uris is None"
        return partial(acquire_content, uri_to_content, store=store)
        
    # Loop through uris and store the processed content
    for key, uri in uris.items():
        content = uri_to_content(uri)
        if save_condition(content):
            store(key, content)


def ensure_store_func(store: Union[Dirpath, MutableMapping, Callable]) -> StoreFunc:
    """
    Ensures a store function is returned based on the type of 'store' argument provided.
    
    - If store is a callable, it returns store directly.
    - If store is a directory path, it creates a Files object (using dol) to manage file storage in that directory.
    - If store is a MutableMapping, it returns the __setitem__ method of the store.
    - If none of these types match, a ValueError is raised.

    Examples:

    >>> store = {}
    >>> func = ensure_store_func(store)
    >>> func('key', 'value')  # should store the value in the dictionary
    >>> assert store == {'key': 'value'}

    >>> store = '~/Downloads'
    >>> try:
    ...     func = ensure_store_func(store)
    ... except ValueError:
    ...     print("Directory does not exist, as expected.")  # Simulates an invalid directory check

    >>> ensure_store_func(lambda k, v: print(f"Storing {k}: {v}"))  # doctest: +ELLIPSIS
    <function <lambda> at ...>

    # Note: For Files store handling, you'll need a valid directory:
    # >>> ensure_store_func("/valid/directory/path")  # Requires dol.Files +SKIP
    
    """
    if callable(store):
        return store
    elif isinstance(store, str):
        dirpath = os.path.expanduser(store)
        if os.path.isdir(dirpath):
            from dol import Files
            return Files(dirpath).__setitem__
        else:
            raise ValueError(f"The directory path {dirpath} does not exist.")
    elif isinstance(store, MutableMapping):
        # If store is a MutableMapping, we'll use its __setitem__ method
        store_obj = store
        return store_obj.__setitem__
    else:
        raise ValueError("uri_to_content must be a callable, or MutableMapping, or a dir path")


# A few useful uri_to_content functions, elegantly defined as (picklable) function compositions

from dol import Pipe 
from pathlib import Path
import requests
from operator import methodcaller, attrgetter

path_to_bytes = Pipe(Path, methodcaller('read_bytes'))
path_to_string = Pipe(Path, methodcaller('read_text'))
url_to_bytes = Pipe(requests.get, attrgetter('content'))

Examples

acquire_url_bytes = acquire_content(url_to_bytes)

memory_hacks = {
    "Memory Improvement in Context.pdf": "https://link.springer.com/content/pdf/10.1007/978-1-4612-2760-1_12.pdf",
    "Predicting and Improving Memory Retention.pdf": "https://home.cs.colorado.edu/~mozer/Research/Selected%20Publications/reprints/MozerLindsey2017.pdf",
    "A New Look at Memory Retention and Forgetting.pdf": "https://memorylab.nd.edu/assets/512320/2022_radvansky_doolen_pettijohn_ritchey_jep_lmc_.pdf",
    "Exercises to Work Memory.pdf": "https://neuronup.us/wp-content/uploads/2021/05/5-free-printable-memory-exercises-downloadable.pdf"
}

acquire_url_bytes(memory_hacks)  # will download all five files to the default download directory

That's nice, and it works, but if I put some urls that didn't work in there, you would have
gotten a raised errors as soon as the first one failed.
If you really want to use this for downloading content from the web, perhaps a but more
robustness is in order.

Also, in the code below, see how we use acquire_content as a decorator.
What's below is (approximately) equivalent to doing more_robust_url_acquisition = acquire_content(more_robust_url_acquisition) like above.). Basically, if you decorate a definition of a uri_to_content, you'll get a content acquisition function that uses it.

@acquire_content
def more_robust_url_acquisition(url: URI, verbose: int = 2) -> bytes:
    verbose = int(verbose)
    try:
        response = requests.get(url)
        response.raise_for_status()  # Check for HTTP errors
        if verbose >= 2:
            print(f"Successfully downloaded and stored contents from {url}")
        return response.content
    except requests.exceptions.RequestException as e:
        if verbose >= 1:
            print(f"Failed to download from {url}: {e}")



more_cheat_sheets = {
    "Scikit-Learn Cheat Sheet: Python Machine Learning.pdf": "https://www.datacamp.com/cheat-sheet/scikit-learn-cheat-sheet-python-machine-learning",
    "Machine Learning Cheat Sheet.pdf": "https://www.datacamp.com/cheat-sheet/machine-learning-cheat-sheet",
    "Scikit-Learn Cheat Sheet for Machine Learning.pdf": "https://www.kdnuggets.com/publications/sheets/Scikit-Learn_Cheatsheet_for_Machine_Learning.pdf",
    "The Complete Collection of Data Science Cheat Sheets.pdf": "https://www.kdnuggets.com/publications/sheets/The_Complete_Collection_of_Data_Science_Cheatsheets_KDnuggets.pdf",
    "Machine Learning Cheat Sheet.pdf": "https://raw.githubusercontent.com/soulmachine/machine-learning-cheat-sheet/master/machine-learning-cheat-sheet.pdf",
    "Data Science Cheat Sheets.pdf": "https://github.com/fralfaro/DS-Cheat-Sheets",
    "Data Science Cheatsheet.pdf": "https://github.com/aaronwangy/Data-Science-Cheatsheet",
    "Cheat Sheet for Python Machine Learning and Data Science.pdf": "https://github.com/ghimiresunil/Cheat-Sheet-for-Python-Machine-Learning-and-Data-Science",
    "Machine Learning and Data Science Cheat Sheet.pdf": "https://www.datasciencecentral.com/data-science-cheat-sheet/",
    "Machine Learning Interview Cheat Sheets.pdf": "https://tinniaru3005.github.io/ExploreCS/notes/machine%20learning.pdf",
    "ML Cheatsheet Documentation.pdf": "https://buildmedia.readthedocs.org/media/pdf/ml-cheatsheet/latest/ml-cheatsheet.pdf",
    "Scikit-Learn CheatSheet: Python Machine Learning Tutorial.pdf": "https://elitedatascience.com/wp-content/uploads/2018/05/Python-Machine-Learning-Cheatsheet.pdf"
}

download_to_folder = '/Users/thorwhalen/Dropbox/_odata/ai_contexts/misc/ml_cheat_sheets'
more_robust_url_acquisition(more_cheat_sheets, store=download_to_folder)  # will download all five files to the default download directory