import sys
import os
from fastcore.foundation import L
from pathlib import Path
from fastcore.xtras import Path
This project continues on from part 1: collecting images
Introduction
In this blog post, I am collecting some reusable functions that I am using for my multi-label classifier project. So far, they are mainly related to the image search queries. I’m exporting this stuff to a mini library called ucm.py
using nbdev.
Useful imports
Avoid warnings from tensorflow
I also added this to my ai.env
environment file.
'TF_CPP_MIN_LOG_LEVEL'] = '1' os.environ[
Collections
from itertools import chain, combinations
def powerset(iterable):
"powerset([1,2,3]) --> () (1,) (2,) (3,) (1,2) (1,3) (2,3) (1,2,3)"
= list(iterable)
s return chain.from_iterable(combinations(s, r) for r in range(len(s)+1))
def seq_diff(s1, s2):
return L(filter(lambda x: x not in s2, s1))
def test_powerset_and_seq_diff():
= L("bird", "cat", "dog")
labels for comb in powerset(labels):
print(comb, seq_diff(labels, comb))
test_powerset_and_seq_diff()
() ['bird', 'cat', 'dog']
('bird',) ['cat', 'dog']
('cat',) ['bird', 'dog']
('dog',) ['bird', 'cat']
('bird', 'cat') ['dog']
('bird', 'dog') ['cat']
('cat', 'dog') ['bird']
('bird', 'cat', 'dog') []
A helper for search query text
import inflect
= inflect.engine() p
def join_a_foo_and_a_bar(comb):
return " and ".join(p.a(x) for x in comb)
"banana", "apple"]) join_a_foo_and_a_bar([
'a banana and an apple'
Logging
import logging
def setup_logging(args):
""" Set up logging. """
# get basename of program in upper case
= os.path.basename(sys.argv[0]).upper()
prog_name_uc
= args.log or os.environ.get(f'{prog_name_uc}_LOG')
log_file = "%(message)s"
fmt if args.log_level == logging.DEBUG:
= "%(asctime)s %(levelname)s %(name)s %(message)s"
fmt
# if a log_file was specified, use it
= log_file or os.environ.get('CHATGPT_LOG_FILE')
log_file =args.log_level, format=fmt, filename=log_file)
logging.basicConfig(level
def add_logging_options(parser):
""" Add logging options to an argument parser. """
= parser.add_argument_group('Logging options')
logging_group =logging.WARNING)
logging_group.set_defaults(log_level'-d', '--debug', dest='log_level', action='store_const', const=logging.DEBUG, help="show debug messages")
logging_group.add_argument('-v', '--verbose', dest='log_level', action='store_const', const=logging.INFO, help="show verbose messages")
logging_group.add_argument('-q', '--quiet', dest='log_level', action='store_const', const=logging.ERROR, help="show only errors")
logging_group.add_argument('-Q', '--silent', dest='log_level', action='store_const', const=logging.CRITICAL, help="show nothing")
logging_group.add_argument('--log', default=None, help="log file") logging_group.add_argument(
Redirecting file handles, such as stderr to /dev/null
import sys
import os
from contextlib import contextmanager
from functools import partial
@contextmanager
def redirect(fileno, target):
""" Redirect a file descriptor temporarily """
= os.open(target, os.O_WRONLY)
target_fd = os.dup(fileno)
saved_fd
os.dup2(target_fd, fileno)try:
yield
finally:
os.dup2(saved_fd, fileno)
os.close(saved_fd)
os.close(target_fd)
= partial(redirect, sys.stderr.fileno(), "/dev/null") redirect_stderr_to_dev_null
# apparently this doesn't work in Jupyter :/
print("Hello", file=sys.stderr)
with redirect_stderr_to_dev_null():
print("this should not appear!", file=sys.stderr)
print("world", file=sys.stderr)
Hello
this should not appear!
world
Run an async coroutine, and exit immediately when the user presses ctrl-c
import os
import asyncio
def run_async(coro):
= asyncio.get_event_loop()
loop try:
loop.run_until_complete(coro)except KeyboardInterrupt as e:
loop.close()130)
os._exit(finally:
loop.close()
File-based Mutex
If lock_file is None / “” / False, it will be a no-op.
import os
import fcntl
class FileMutex:
def __init__(self, lock_file, mode=0o600):
self.lock_file = lock_file
self.fd = None
self.mode = mode
def acquire(self):
if not self.lock_file:
return
self.fd = os.open(self.lock_file, os.O_CREAT, self.mode)
self.fd, fcntl.LOCK_EX)
fcntl.flock(
def release(self):
if not self.fd:
return
self.fd, fcntl.LOCK_UN)
fcntl.flock(self.fd)
os.close(self.fd = None
def __enter__(self):
self.acquire()
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
# not exactly a thorough test...
with FileMutex("/tmp/lock"):
print("hi")
hi
Exporting code with nbdev
from nbdev.export import nb_export
import ipynbname
from pathlib import Path
import sh
def git_root():
= sh.git('rev-parse', '--show-toplevel').rstrip()
root return root
def export(nb_file=None, lib_dir=None):
if nb_file is None: nb_file = ipynbname.name() + '.ipynb'
if lib_dir is None: lib_dir = Path(git_root())/"lib"
nb_export(nb_file, lib_dir)
export()