Skip to content

utils.py

QuietStdOut

Context for quieting standard output, and setting datajoint loglevel to warning

Used in pytest functions to render clear output showing only pass/fail

Example

with QuietStdOut(): table.delete(safemode=False)

Source code in element_interface/utils.py
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
class QuietStdOut:
    """Context for quieting standard output, and setting datajoint loglevel to warning

    Used in pytest functions to render clear output showing only pass/fail

    Example:
        with QuietStdOut():
            table.delete(safemode=False)
    """

    def __enter__(self):
        self.prev_log_level = logger.level
        logger.setLevel(30)  # set DataJoint logger to warning
        self._original_stdout = sys.stdout
        sys.stdout = open(os.devnull, "w")

    def __exit__(self, *args):
        logger.setLevel(self.prev_log_level)
        sys.stdout.close()
        sys.stdout = self._original_stdout

dict_to_uuid(key)

Given a dictionary key, returns a hash string as UUID

Parameters:

Name Type Description Default
key dict

Any python dictionary

required
Source code in element_interface/utils.py
 98
 99
100
101
102
103
104
105
106
107
def dict_to_uuid(key: dict):
    """Given a dictionary `key`, returns a hash string as UUID

    Args:
        key (dict): Any python dictionary"""
    hashed = hashlib.md5()
    for k, v in sorted(key.items()):
        hashed.update(str(k).encode())
        hashed.update(str(v).encode())
    return uuid.UUID(hex=hashed.hexdigest())

find_full_path(root_directories, relative_path)

Given a list of roots and a relative path, search and return the full-path

Root directories are searched in the provided order

Parameters:

Name Type Description Default
root_directories list

potential root directories

required
relative_path str

the relative path to find the valid root directory

required

Returns:

Type Description
pathlib.PosixPath

full-path (pathlib.Path object)

Raises:

Type Description
FileNotFoundError

No valid full path

Source code in element_interface/utils.py
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def find_full_path(root_directories: list, relative_path: str) -> pathlib.PosixPath:
    """Given a list of roots and a relative path, search and return the full-path

    Root directories are searched in the provided order

    Args:
        root_directories (list): potential root directories
        relative_path (str): the relative path to find the valid root directory

    Returns:
        full-path (pathlib.Path object)

    Raises:
        FileNotFoundError: No valid full path
    """
    relative_path = _to_Path(relative_path)

    if relative_path.exists():
        return relative_path

    # Turn to list if only a single root directory is provided
    if isinstance(root_directories, (str, pathlib.Path)):
        root_directories = [_to_Path(root_directories)]

    for root_dir in root_directories:
        if (_to_Path(root_dir) / relative_path).exists():
            return _to_Path(root_dir) / relative_path

    raise FileNotFoundError(
        "No valid full-path found (from {})"
        " for {}".format(root_directories, relative_path)
    )

find_root_directory(root_directories, full_path)

Given multiple potential root directories and a full-path, return parent root.

Search and return one directory that is the parent of the given path.

Parameters:

Name Type Description Default
root_directories list

potential root directories

required
full_path str

the full path to search the root directory

required

Returns:

Type Description
pathlib.PosixPath

root_directory (pathlib.Path object)

Raises:

Type Description
FileNotFoundError

Full path does not exist

FileNotFoundError

No valid root directory

Source code in element_interface/utils.py
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def find_root_directory(root_directories: list, full_path: str) -> pathlib.PosixPath:
    """Given multiple potential root directories and a full-path, return parent root.

    Search and return one directory that is the parent of the given path.

    Args:
        root_directories (list): potential root directories
        full_path (str): the full path to search the root directory

    Returns:
        root_directory (pathlib.Path object)

    Raises:
        FileNotFoundError: Full path does not exist
        FileNotFoundError: No valid root directory
    """
    full_path = _to_Path(full_path)

    if not full_path.exists():
        raise FileNotFoundError(f"{full_path} does not exist!")

    # Turn to list if only a single root directory is provided
    if isinstance(root_directories, (str, pathlib.Path)):
        root_directories = [_to_Path(root_directories)]

    try:
        return next(
            _to_Path(root_dir)
            for root_dir in root_directories
            if _to_Path(root_dir) in set(full_path.parents)
        )

    except StopIteration:
        raise FileNotFoundError(
            "No valid root directory found (from {})"
            " for {}".format(root_directories, full_path)
        )

ingest_csv_to_table(csvs, tables, verbose=True, skip_duplicates=True, ignore_extra_fields=True, allow_direct_insert=False)

Inserts data from a series of csvs into their corresponding table:

Example

ingest_csv_to_table(['./lab_data.csv', './proj_data.csv'], [lab.Lab(),lab.Project()]

Parameters:

Name Type Description Default
csvs list

list of paths to CSV files relative to current directory. CSV are delimited by commas.

required
tables list

list of datajoint tables with terminal ()

required
verbose bool

print number inserted (i.e., table length change)

True
skip_duplicates bool

skip duplicate entries. See DataJoint's insert

True
ignore_extra_fields bool

if a csv feeds multiple tables, the subset of columns not applicable to a table will be ignored. See DataJoint's insert

True
allow_direct_insert bool

permit insertion into Imported and Computed tables See DataJoint's insert.

False
Source code in element_interface/utils.py
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def ingest_csv_to_table(
    csvs: list,
    tables: list,
    verbose: bool = True,
    skip_duplicates: bool = True,
    ignore_extra_fields: bool = True,
    allow_direct_insert: bool = False,
):
    """Inserts data from a series of csvs into their corresponding table:

    Example:
        > ingest_csv_to_table(['./lab_data.csv', './proj_data.csv'],
                                 [lab.Lab(),lab.Project()]

    Args:
        csvs (list): list of paths to CSV files relative to current directory.
            CSV are delimited by commas.
        tables (list): list of datajoint tables with terminal `()`
        verbose (bool): print number inserted (i.e., table length change)
        skip_duplicates (bool): skip duplicate entries. See DataJoint's `insert`
        ignore_extra_fields (bool): if a csv feeds multiple tables, the subset of
            columns not applicable to a table will be ignored. See DataJoint's `insert`
        allow_direct_insert (bool): permit insertion into Imported and Computed tables
            See DataJoint's `insert`.
    """
    for csv_filepath, table in zip(csvs, tables):
        with open(csv_filepath, newline="") as f:
            data = list(csv.DictReader(f, delimiter=","))
        if verbose:
            prev_len = len(table)
        table.insert(
            data,
            skip_duplicates=skip_duplicates,
            # Ignore extra fields because some CSVs feed multiple tables
            ignore_extra_fields=ignore_extra_fields,
            # Allow direct bc element-event uses dj.Imported w/o `make` funcs
            allow_direct_insert=allow_direct_insert,
        )
        if verbose:
            insert_len = len(table) - prev_len
            logger.info(
                f"\n---- Inserting {insert_len} entry(s) "
                + f"into {to_camel_case(table.table_name)} ----"
            )

value_to_bool(value)

Return whether the provided value represents true. Otherwise false.

Parameters:

Name Type Description Default
value (str, bool, int)

Any input

required

Returns:

Name Type Description
bool bool

True if value in ("y", "yes", "t", "true", "on", "1")

Source code in element_interface/utils.py
156
157
158
159
160
161
162
163
164
165
166
167
def value_to_bool(value) -> bool:
    """Return whether the provided value represents true. Otherwise false.

    Args:
        value (str, bool, int): Any input

    Returns:
        bool (bool): True if value in ("y", "yes", "t", "true", "on", "1")
    """
    if not value:
        return False
    return str(value).lower() in ("y", "yes", "t", "true", "on", "1")