-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathutils.py
More file actions
8115 lines (7375 loc) · 373 KB
/
Copy pathutils.py
File metadata and controls
8115 lines (7375 loc) · 373 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
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
46
47
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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
from __future__ import annotations
import os
import polars as pl
import signal
import subprocess
import sys
from contextlib import contextmanager
from functools import cache, reduce, wraps
from typing import Any, Iterable
pl.enable_string_cache()
###############################################################################
# [1] General utilities
###############################################################################
@contextmanager
def Timer(message=None, verbose=True):
"""
Use "with Timer(message):" to time the code inside the with block. Based on
preshing.com/20110924/timing-your-code-using-pythons-with-statement
Args:
message: a message to print when starting the with block (with "..."
after) and ending the with block (with the time after)
verbose: if False, disables the Timer. This is useful to conditionally
run the Timer based on the value of a boolean variable.
"""
if verbose:
from timeit import default_timer
if message is not None:
print(f'{message}...')
start = default_timer()
aborted = False
try:
yield
except Exception as e:
aborted = True
raise e
finally:
end = default_timer()
duration = end - start
days = int(duration // 86400)
hours = int((duration % 86400) // 3600)
minutes = int((duration % 3600) // 60)
seconds = int(duration % 60)
milliseconds = int((duration * 1000) % 1000)
microseconds = int((duration * 1000000) % 1000)
nanoseconds = int((duration * 1000000000) % 1000)
time_parts = []
if days > 0:
time_parts.append(f'{days} {plural("day", days)}')
if hours > 0:
time_parts.append(f'{hours}h')
if minutes > 0:
time_parts.append(f'{minutes}m')
if seconds > 0:
time_parts.append(f'{seconds}s')
if milliseconds > 0:
time_parts.append(f'{milliseconds}ms')
if microseconds > 0:
time_parts.append(f'{microseconds}µs')
if nanoseconds > 0:
time_parts.append(f'{nanoseconds}ns')
time_str = \
' '.join(time_parts[:2]) if time_parts else 'less than 1ns'
print(f'{message if message is not None else "Command"} '
f'{"aborted after" if aborted else "took"} '
f'{time_str}')
else:
yield # no-op
@contextmanager
def cd(path, *, create_if_missing=False):
"""
Use "with cd(path):" to temporarly change directory inside the with block
Args:
path: the directory to change to temporarily
create_if_missing: whether to create the directory if missing
"""
if create_if_missing:
os.makedirs(path, exist_ok=True)
original_dir = os.getcwd()
try:
os.chdir(path)
yield
finally:
os.chdir(original_dir)
@contextmanager
def SuppressMessages():
"""
Use "with SuppressMessages:" to suppress stdout inside the with block
"""
try:
sys.stdout = open(os.devnull, "w")
yield
finally:
sys.stdout = sys.__stdout__
def raise_error_if_on_compute_node(message=None):
"""
Raises an error if the user is on a compute node
Args:
message: Message to print when user is not on a login node; if None,
prints a default message
"""
import re
import socket
if re.search('(nia|nc|nl|ng)[0-9]', socket.gethostname()):
# nia matches the Niagara compute nodes; nc/nl/ng match the Narval ones
import inspect
calling_function = inspect.currentframe().f_back.f_code.co_name
raise RuntimeError(message if message is not None else
f'{calling_function}() needs internet access! Run '
f'once on the login node to download required '
f'files, then re-run here on this compute node')
def check_cluster(cluster):
"""
Check the cluster the user is on.
Returns:
The cluster: "narval" or "niagara". Raises an error if $CLUSTER is not
set to one of those two.
"""
if cluster is None:
raise RuntimeError('The environment variable $CLUSTER is not set; it '
'must be set to "narval" or "niagara"')
if cluster != 'narval' and cluster != 'niagara':
raise RuntimeError(f"The environment variable $CLUSTER is set to "
f"{cluster!r}, but must be set to 'narval' or "
f"'niagara'")
@cache
def get_base_data_directory():
"""
Get the location of the "base" data directory, where data will be stored.
Returns:
The path of the base data directory
"""
cluster = os.environ.get('CLUSTER')
return '/home/wainberg/projects/def-wainberg' if cluster == 'narval' else \
'/scratch/w/wainberg/wainberg' if cluster == 'niagara' else '.'
def print_large_objects(N=20):
"""
Print the N largest objects in the user's Python session.
Args:
N: the number of largest objects to print
"""
# There are 2 main options for accurately calculating Python object sizes:
# - objsize.get_deep_size from pypistats.org/packages/objsize
# - pympler.asizeof.asizeof from pypistats.org/packages/pympler
from pympler.asizeof import asizeof
mem = pl.DataFrame({'Variable': list(globals()),
'Size': [asizeof(eval(key)) for key in globals()]})\
.with_columns(pl.all().sort_by('Size', descending=True))
with pl.Config(tbl_hide_dataframe_shape=True):
print(mem.head(N))
def escape(x):
"""
Escapes a string or polars Series or expression by removing leading and
trailing whitespace and converting internal groups of 1+ whitespace
characters to a single underscore.
Args:
x: the string, Series of expression to escape
Returns:
The escaped version of x.
"""
if isinstance(x, str):
import re
return re.sub(r'\W+', '_', x).strip('_')
else:
return x.str.replace(r'\W+', '_').str.strip_chars('_')
def plural(string, count):
"""
Adds an s to the end of string, unless count is 1 or -1.
Args:
string: a string
count: a count
Returns:
string, with an s at the end fo count is 1 or -1
"""
return string if abs(count) == 1 else f'{string}s'
def check_type(variable: Any, variable_name: str,
expected_types: type | tuple[type, ...],
expected_type_name: str):
"""
Raise a TypeError if `variable` is not of the expected type.
Args:
variable: the variable to be checked
variable_name: the name of the variable, used in the error message
expected_types: the expected type or types (specifying int, float, or
bool also implicitly includes their NumPy equivalents)
expected_type_name: the name of the expected type, used in the error
message (e.g. 'a polars DataFrame')
"""
if isinstance(variable, expected_types):
return
if not isinstance(expected_types, tuple):
expected_types = expected_types,
for t in expected_types:
if t is int:
import numpy as np
if isinstance(variable, np.integer):
return
elif t is float:
import numpy as np
if isinstance(variable, np.floating):
return
elif t is bool:
import numpy as np
if isinstance(variable, np.bool_):
return
error_message = (
f'{variable_name} must be {expected_type_name}, but has type '
f'{type(variable).__name__!r}')
raise TypeError(error_message)
def check_types(variable: Iterable[Any], variable_name: str,
expected_types: type | tuple[type, ...],
expected_type_name: str):
"""
Raise a TypeError if not all elements of `variable` are of the expected
type(s).
Args:
variable: the variable to be checked
variable_name: the name of the variable, used in the error message
expected_types: the expected type or types
expected_type_name: the name of the expected type, used in the error
message (e.g. 'polars DataFrames')
"""
if not isinstance(expected_types, tuple):
expected_types = expected_types,
for element in variable:
if not isinstance(element, expected_types):
for t in expected_types:
if t is int:
import numpy as np
if isinstance(variable, np.integer):
break
elif t is float:
import numpy as np
if isinstance(variable, np.floating):
break
elif t is bool:
import numpy as np
if isinstance(variable, np.bool_):
break
else:
error_message = (
f'all elements of {variable_name} must be '
f'{expected_type_name}, but it contains an element of '
f'type {type(element).__name__!r}')
raise TypeError(error_message)
def check_dtype(series: pl.Series, series_name: str,
expected_dtypes: pl.datatypes.classes.DataTypeClass | str |
tuple[pl.datatypes.classes.DataTypeClass |
str, ...]):
"""
Raise a TypeError if series is not of the expected polars dtype.
Args:
series: the polars Series to be checked
series_name: the name of the variable, used in the error message
expected_dtypes: the expected dtype or dtypes. Specify the string
'integer' to include all integer dtypes, and
'floating-point' to include all floating-point dtypes.
"""
base_type = series.dtype.base_type()
if not isinstance(expected_dtypes, tuple):
expected_dtypes = expected_dtypes,
for expected_type in expected_dtypes:
if base_type == expected_type or expected_type == 'integer' and \
base_type in pl.INTEGER_DTYPES or \
expected_type == 'floating-point' and \
base_type in pl.FLOAT_DTYPES:
return
if len(expected_dtypes) == 1:
expected_dtypes = str(expected_dtypes[0])
elif len(expected_dtypes) == 2:
expected_dtypes = ' or '.join(map(str, expected_dtypes))
else:
expected_dtypes = ', '.join(map(str, expected_dtypes[:-1])) + \
', or ' + str(expected_dtypes[-1])
error_message = (
f'{series_name} must be {expected_dtypes}, but has data type '
f'{base_type!r}')
raise TypeError(error_message)
def check_bounds(variable, variable_name, lower_bound=None, upper_bound=None,
*, left_open=False, right_open=False):
"""
Check whether variable is between lower bound and upper bound, inclusive.
Args:
variable: the variable to be checked
variable_name: the name of the variable, used in the error message
lower_bound: the smallest allowed value for variable, or None to have
no lower bound
upper_bound: the largest allowed value for variable, or None to have no
upper bound
left_open: if True, require variable to be strictly greater than
lower_bound, rather than >= lower_bound; has no effect if
lower_bound is None
right_open: if True, require variable to be strictly less than
upper_bound, rather than <= upper_bound; has no effect if
upper_bound is None
"""
if lower_bound is None and upper_bound is None:
error_message = 'lower_bound and upper_bound cannot both be None'
raise ValueError(error_message)
if lower_bound is not None and (variable <= lower_bound if left_open
else variable < lower_bound) or \
upper_bound is not None and (variable >= upper_bound if right_open
else variable > upper_bound):
error_message = f'{variable_name} is {variable:,}, but must be'
if lower_bound is not None:
error_message += f' {">" if left_open else "≥"} {lower_bound:,}'
if upper_bound is not None:
error_message += ' and'
if upper_bound is not None:
error_message += f' {"<" if right_open else "≤"} {upper_bound:,}'
raise ValueError(error_message)
def check_R_variable_name(
R_variable_name: str,
variable_name: str,
R_keywords: set[str] = {
'if', 'else', 'repeat', 'while', 'function', 'for', 'in', 'next',
'break', 'TRUE', 'FALSE', 'NULL', 'Inf', 'NaN', 'NA',
'NA_integer_', 'NA_real_', 'NA_complex_', 'NA_character_',
'...'}) -> None:
"""
Raise an error if R_variable_name is not a valid variable name in R.
Args:
R_variable_name: the R variable name to be checked
variable_name: the name of the Python variable the R variable name
`R_variable_name` is stored in
R_keywords: the set of R reserved keywords to check against
"""
import re
if not R_variable_name:
error_message = f'{variable_name} is an empty string'
raise ValueError(error_message)
if R_variable_name[0] == '.':
if len(R_variable_name) > 1 and R_variable_name[1].isdigit():
error_message = (
f'{variable_name} {R_variable_name!r} starts with a period '
f'followed by a digit, which is not a valid R variable name')
raise ValueError(error_message)
elif not R_variable_name[0].isidentifier():
error_message = (
f'{variable_name} {R_variable_name!r} must start with a letter, '
f'number, period or underscore')
raise ValueError(error_message)
if not re.fullmatch(r'[\w.]*', R_variable_name[1:]):
invalid_characters = \
sorted(set(re.findall(r'[^\w.]',
''.join(dict.fromkeys(R_variable_name)))))
if len(invalid_characters) == 1:
description = f"the character '{invalid_characters[0]}'"
else:
description = f"the characters " + ", ".join(
f"'{character}'" for character in invalid_characters) + \
f" and '{invalid_characters[-1]}'"
error_message = (
f'{variable_name} {R_variable_name!r} contains {description}, but '
f'must contain only letters, numbers, periods and underscores')
raise ValueError(error_message)
if R_variable_name in R_keywords or (R_variable_name.startswith('..') and
R_variable_name[2:].isdigit()):
error_message = (
f'{variable_name} {R_variable_name!r} is a reserved keyword in R, '
f'and cannot be used as a variable name')
raise ValueError(error_message)
class ProcessPool(object):
"""
Like multiprocessing.Pool but 1) child processes ignore KeyboardInterrupts
and 2) apply_async() is called submit() and takes actual *args and **kwargs
rather than a list of args and a dictionary of kwargs (like
ProcessPoolExecutor.submit() from the concurrent.futures module)
Attributes:
pool (multiprocessing.Pool): the underlying multiprocessing.Pool object
"""
def __init__(self, max_concurrent, start_method='forkserver'):
"""
Sets up the multiprocessing.Pool object underlying this ProcessPool.
Args:
max_concurrent: The number of worker processes to be spawned by the
Pool object, i.e. the maximum number of concurrent
processes that will be allowed to run at once.
start_method: How worker processes will be started. Possible
values are 'fork', 'spawn', 'forkserver'. For
details, see docs.python.org/3/library/
multiprocessing.html#contexts-and-start-methods.
"""
import multiprocessing
self.pool = multiprocessing.get_context(start_method)\
.Pool(max_concurrent, initializer=self.ignore_keyboard_interrupts)
@staticmethod
def ignore_keyboard_interrupts():
"""
When provided as an initializer to a multiprocessing.Pool object,
this function tells the Pool to ignore KeyboardInterrupts (i.e. SIGINT)
so that pressing Ctrl + C doesn't kill all your background processes.
"""
signal.signal(signal.SIGINT, signal.SIG_IGN)
def submit(self, func, *args, **kwargs):
"""
Submits a function to the process pool. A thin wrapper over
Pool.apply_async() that allows the user to pass actual *args and
**kwargs rather than a list of args and a dictionary of kwargs.
Args:
func: the function to be submitted
*args: positional arguments to be passed to function
**kwargs: keyword arguments to be passed to function
Returns:
The multiprocessing.pool.AsyncResult object returned by
Pool.apply_async().
"""
return self.pool.apply_async(func, args, kwargs)
@cache
def get_process_pool(max_concurrent, start_method='forkserver'):
"""
Creates a ProcessPool for a given value of max_concurrent and start_method,
which due to the @cache decorator will persist across multiple calls to
functions like run_background() or run_function_background(), so long as
they call this function with the same max_concurrent and start_method.
Args:
max_concurrent: The number of worker processes to be spawned by the
ProcessPool, i.e. the maximum number of concurrent
processes that will be allowed to run at once.
start_method: How worker processes will be started. Possible
values are 'fork', 'spawn', 'forkserver'. For
details, see docs.python.org/3/library/
multiprocessing.html#contexts-and-start-methods.
Returns:
A ProcessPool for the given values of max_concurrent and start_method.
A new ProcessPool will be created the first time this function is run
for a given value of max_concurrent and start_method, which will then
be cached for all subsequent calls with the same max_concurrent and
start_method.
"""
return ProcessPool(max_concurrent, start_method=start_method)
@cache
def cython_inline(code, boundscheck=False, cdivision=True,
initializedcheck=False, wraparound=False,
warn_undeclared=True, debug_symbols=False, include_dirs=None,
libraries=None, verbose=False, **other_cython_settings):
"""
A drop-in replacement for cython.inline that supports cimports. It turns on
the major Cython optimizations (boundscheck=False, cdivision=True,
initializedcheck=False, wraparound=False), sets quiet=True for quiet
compilation, and sets language_level=3 for full Python 3 compatibility.
Args:
code: a string of Cython code to compile
boundscheck: whether to perform array bounds checking when indexing;
always affects array/memoryview indexing, but also affects
list, tuple, and string indexing when wraparound=False
cdivision: whether to use C-style rather than Python-style division and
remainder operations; disabling leads to a ~35% speed
penalty for these operations
initializedcheck: whether to check whether memoryviews and C++ classes
are initialized before using them
wraparound: whether to support Python-style negative indexing
warn_undeclared: whether to warn about undeclared variables (i.e. those
without a cdef type declaration)
debug_symbols: whether to add debug symbols, so you can run tools like
gdb or valgrind; slows down your code
include_dirs: an optional tuple of include directories of libraries to
link against; `np.get_include()` will always be included
libraries: an optional tuple of libraries to link against,
e.g. ('hdf5',)
verbose: if True, print Cython's compilation logs
**other_cython_settings: other Cython settings, which will be written
into the source code as #cython compiler
directives
Returns:
The {function_name: function} dictionary of compiled functions that
would be returned by cython.inline().
"""
from hashlib import md5
from inspect import getmembers
from textwrap import dedent
# ~ is read-only on Niagara compute nodes, so build in CYTHON_CACHE_DIR in
# scratch instead
cython_cache_dir = os.path.abspath(os.environ.get(
'CYTHON_CACHE_DIR', os.path.expanduser('~/.cython')))
os.makedirs(cython_cache_dir, exist_ok=True)
# Remove extra levels of indentation from the code string (since it's
# usually defined inside a function, so there's at least one extra level of
# indentation that would cause a syntax error if not removed) and remove
# any leading newlines if present (since when users define code strings,
# they usually use triple-quoted strings, and the code usually doesn't
# start until the line after the three opening quotes, leading to a single
# leading newline)
settings = dict(language_level=3, boundscheck=boundscheck,
cdivision=cdivision, initializedcheck=initializedcheck,
wraparound=wraparound,
**{'warn.undeclared': warn_undeclared})
settings.update(other_cython_settings)
code = ''.join(f'#cython: {setting_name}={setting}\n'
for setting_name, setting in settings.items()) + \
f'#distutils: define_macros=NPY_NO_DEPRECATED_API=' \
f'NPY_1_7_API_VERSION ' + dedent(code)
# Make a short alphabetic module name by taking the code string's MD5 hash
# and converting the hexadegimal digits to letters (0 -> a, 1 -> b, ...,
# 9 --> j, a --> k, ..., f --> p)
module_name = ''.join(chr(ord(c) + (49 if c <= '9' else 10))
for c in md5(code.encode('utf-8')).hexdigest())
code_file = os.path.join(cython_cache_dir, f'{module_name}.pyx')
# Try to import the module; build it if it does not exist
sys.path.append(cython_cache_dir)
try:
module = __import__(module_name)
except ModuleNotFoundError:
# Create the code file
with open(code_file, 'w') as f:
# noinspection PyTypeChecker
print(code, file=f)
# Write a build script to a temp file based on the module name
build_file = os.path.join(cython_cache_dir, f'{module_name}_build.py')
with open(build_file, 'w') as f:
import numpy as np
if include_dirs is not None:
include_dirs = (np.get_include(),) + include_dirs
else:
include_dirs = np.get_include(),
include_dirs = \
'[' + ', '.join(f'{include_dir!r}'
for include_dir in include_dirs) + ']'
if libraries is not None:
libraries = \
f'[' + ', '.join(f'{library!r}'
for library in libraries) + ']'
# noinspection PyTypeChecker
print(dedent(f'''
from setuptools import Extension, setup
from Cython.Build import cythonize
setup(name='{module_name}', ext_modules=cythonize([
Extension('{module_name}', ['{code_file}'],
language='c++',
include_dirs={include_dirs},
libraries={libraries},
extra_compile_args=['-Ofast', '-march=native',
'-funroll-loops',
'-fopenmp', '-Werror'],
extra_link_args=['-Ofast', '-fopenmp'])],
build_dir='{cython_cache_dir}'))'''), file=f)
# Build the code (note: `sys.executable` is the location of Python)
run(f'cd {cython_cache_dir} && '
f'{"CFLAGS=-g " if debug_symbols else ""}'
f'{sys.executable} {build_file} build_ext --inplace'
f'{"" if verbose else " > /dev/null"}')
# Remove the temp file
os.unlink(build_file)
# Try again
module = __import__(module_name)
finally:
sys.path = sys.path[:-1]
# Create a dict of all the Cython functions defined in the module
function_dict = {function_name: function
for function_name, function in getmembers(module)
if repr(function).startswith('<cyfunction')}
# Return the dict of Cython functions
return function_dict
@cache
def cython_type(dtype):
"""
Converts a NumPy dtype or string representation of a dtype to its
corresponding Cython type. Raises a TypeError if dtype isn't recognized or
is not a Cython type
Args:
dtype: An NumPy dtype object (e.g. np.float32) or string representation
of a dtype (e.g. 'float32').
Returns:
str: Corresponding Cython type as a string.
"""
import numpy as np
cython_types = {
'i1': 'char', 'u1': 'unsigned char', 'i2': 'short',
'u2': 'unsigned short', 'i4': 'int', 'u4': 'unsigned int',
'i8': 'long', 'u8': 'unsigned long', 'i16': 'long long',
'u16': 'unsigned long long', 'f2': 'float16', 'f4': 'float',
'f8': 'double', 'f16': 'long double', 'c8': 'complex float',
'c16': 'complex double', 'c32': 'complex long double', 'b1': 'bint'}
try:
dtype_string = np.dtype(dtype).str[1:]
except TypeError:
raise TypeError(f'{dtype!r} is not a valid NumPy dtype')
try:
cython_type = cython_types[dtype_string]
except KeyError:
raise TypeError(f'{dtype!r} does not correspond to any Cython type')
return cython_type
def debug(turn_on=True, *, third_party=False):
"""
Turns on "debug mode", or turns it off if turn_on=False.
In debug mode, whenever you get an error inside a function, local variables
from inside the function are automatically copied into the global
namespace, instead of just being discarded.
Of course, the error may happen many layers of functions deep, so do this
for every stack frame (nested function call)! Go from the outermost stack
frame to the innermost, so that variables in inner stack frames overwrite
variables with the same name from outer stack frames.
However, do not include variables from third-party library code (i.e. code
files in your miniforge/mambaforge directory), unless
include_library_variables=True. utils.py is not considered library code!
Implementation details:
- Uses sys.modules['__main__'].__dict__ (or get_ipython().user_global_ns
for IPython) instead of globals(): globals() is a module-level variable
and each module has its own globals(), so we'd only be modifying
utils.py's globals() and not the REPL's globals()!
- Imported modules are in f_globals but not f_locals, but functions' local
variables are in f_locals but not f_globals, so include both.
Args:
turn_on: whether to turn on (if True) or turn off (if False) debug mode
third_party: if True, copies variables from third-party library code,
not just those from your code
"""
def add_variables_to_globals(traceback, global_namespace):
while traceback is not None:
if third_party or ('miniforge3' not in
traceback.tb_frame.f_code.co_filename and 'mambaforge'
not in traceback.tb_frame.f_code.co_filename):
global_namespace.update(traceback.tb_frame.f_globals)
global_namespace.update(traceback.tb_frame.f_locals)
traceback = traceback.tb_next
# Inside polars, the module object "pl" is sometimes reassigned to the
# module "polars._reexport"; reset it here
if third_party and 'pl' in global_namespace:
import polars as pl
global_namespace['pl'] = pl
try:
# noinspection PyUnresolvedReferences
ipython = get_ipython()
except NameError:
# Not IPython
if turn_on:
def excepthook(exception_class, exception, traceback):
global_namespace = sys.modules['__main__'].__dict__
add_variables_to_globals(traceback, global_namespace)
sys.__excepthook__(exception_class, exception, traceback)
sys.excepthook = excepthook
else:
sys.excepthook = sys.__excepthook__ # reset
else:
# IPython
if turn_on:
def excepthook(shell, etype, evalue, tb, tb_offset=None):
global_namespace = ipython.user_global_ns
add_variables_to_globals(tb, global_namespace)
shell.showtraceback((etype, evalue, tb), tb_offset=tb_offset)
ipython.set_custom_exc((Exception,), excepthook)
else:
ipython.set_custom_exc((), None) # reset
def to_tuple(variable):
"""
Cast Iterables (except str/bytes) to tuple, but box non-Iterables (and
str/bytes) in a length-1 tuple.
Args:
variable: a variable
Returns:
`variable` as a tuple
"""
from collections.abc import Iterable
return tuple(variable) if isinstance(variable, Iterable) and not \
isinstance(variable, (str, bytes)) else (variable,)
def to_tuple_checked(variable: Any | Iterable[Any], variable_name: str,
expected_types: type | tuple[type, ...],
expected_type_name: str) -> tuple[Any, ...]:
"""
Like `to_tuple`, but check that `variable` or its elements are of the
expected type(s) and that it is non-empty.
Args:
variable: the variable to be checked and expanded
variable_name: the name of the variable, used in error messages
expected_types: the expected type or types
expected_type_name: the name of the expected type, used in error
messages (e.g. 'polars DataFrames')
Returns:
`variable` as a tuple.
"""
from collections.abc import Iterable
if isinstance(variable, Iterable) and \
not isinstance(variable, (str, bytes)):
variable = tuple(variable)
if len(variable) == 0:
error_message = f'{variable_name} is empty'
raise ValueError(error_message)
check_types(variable, variable_name, expected_types,
expected_type_name)
else:
check_type(variable, variable_name, expected_types,
f'{expected_type_name} (or a sequence thereof)')
variable = variable,
return variable
def reload(module):
"""
A drop-in replacement for `importlib.reload()` that also updates existing
objects' methods (though not class or instance variables).
After reloading the given module, it updates:
1. All type objects in the caller's global namespace that have the same
name as a class defined in the reloaded module. The update modifies the
existing type objects' methods to match those of the newly reloaded
classes.
2. All functions, submodules and variables in the caller's global namespace
that have the same name as a function, submodule or variable defined in
the reloaded module.
It then updates the `__class__` attribute of all objects defined anywhere
(according to `gc.get_objects()`) so `isinstance()` checks don't break.
This function is mainly meant to be called interactively, in which case it
will update all objects defined in your current Python session.
Args:
module: the module to reload
Returns:
The newly reloaded module.
"""
import gc
import importlib
from types import FunctionType
def deleted(key):
error_message = \
f'method {key!r} no longer exists after running reload()'
raise AttributeError(error_message)
# Sometimes the reload doesn't work on the first try (due to caching?) so
# try twice
for _ in range(2):
# Perform the standard reload
module = importlib.reload(module)
module_name = module.__name__
module_dict = module.__dict__
# Get the calling frame's globals
# noinspection PyUnresolvedReferences
calling_frame_globals = sys._getframe(1).f_globals
# Construct the set of classes to update
types_to_update = {
cls for cls in map(type, calling_frame_globals.values())
if cls.__module__ == module_name and cls.__name__ in module_dict}
# Update each class
for cls in types_to_update:
class_dict = cls.__dict__
class_name = cls.__name__
new_class = module_dict[class_name]
new_class_dict = new_class.__dict__
# Add/update methods
for key, value in new_class_dict.items():
if isinstance(value, FunctionType):
setattr(cls, key, value)
# Remove methods that no longer exist
# (note: it's not possible to fully remove the method because of
# the way Python caches method lookups for performance)
for key in class_dict.keys() - new_class_dict.keys():
if isinstance(class_dict[key], FunctionType):
setattr(cls, key, (
lambda key: lambda *args, **kwargs: deleted(key))(key))
# Update __class__ attribute of all instances of the class to point
# to the new class, so that `isinstance()` checks don't break
for obj in gc.get_objects():
if type(obj).__name__ == class_name:
obj.__class__ = new_class
# Update functions, sub-modules, and variables
for key, value in module_dict.items():
if key in calling_frame_globals and not key.startswith('__'):
calling_frame_globals[key] = value
return module
###############################################################################
# [2] Polars
###############################################################################
def save_npy(df, filename):
# noinspection GrazieInspection
"""
Saves df to filename in NumPy's .npy binary format.
Args:
df: a polars DataFrame; all columns must have the same numeric dtype
filename: a filename to save to. df's data will be saved to
f'{filename.removesuffix(".npy")}.npy', and its columns to
f'{filename.removesuffix(".npy")}.columns'.
"""
if df.is_empty():
raise ValueError('df is empty!')
dtypes = set(df.dtypes)
if len(dtypes) > 1 or dtypes.pop() not in pl.NUMERIC_DTYPES:
raise ValueError('All columns of df must have the same numeric dtype '
'to save with save_npy()')
import numpy as np
prefix = filename.removesuffix('.npy')
np.save(f'{prefix}.npy', df.to_numpy())
pl.DataFrame(df.columns).write_csv(f'{prefix}.columns',
include_header=False)
def load_npy(filename):
# noinspection GrazieInspection
"""
Loads a polars DataFrame saved via save_npy().
Args:
filename: a filename to load from. The DataFrame's data will be loaded
from f'{filename.removesuffix(".npy")}.npy', and its columns
from f'{filename.removesuffix(".npy")}.columns'.
Returns:
The polars DataFrame.
"""
import numpy as np
prefix = filename.removesuffix('.npy')
return pl.from_numpy(np.load(f'{prefix}.npy'), pl.read_csv(
f'{prefix}.columns', has_header=False).to_series().to_list())
def print_df(df, num_rows=-1, num_columns=-1):
"""
Prints the entirety of a polars DataFrame without truncating.
Args:
df: the DataFrame to print
num_rows: the number of rows to print (-1 to print all rows)
num_columns: the number of columns to print (-1 to print all columns)
"""
with pl.Config(tbl_rows=num_rows, tbl_cols=num_columns,
set_tbl_width_chars=900):
print(df)
def print_row(df, row_number=0):
"""
Prints a row of a polars DataFrame with each column's value on its own line
alongside its column header. Similar to df.glimpse() but less cluttered and
only showing one row.
Args:
df: the DataFrame to print the row of
row_number: which row to print (by default, the first)
"""
print_df(df[row_number].unpivot(variable_name='column'))
def filter_columns(df, predicates, *more_predicates):
"""
Selects columns from a polars DataFrame where all the boolean expressions
in predicates evaluate to True, like filter() but for columns instead of
rows. Use it in method chains, e.g. df.pipe(filter_columns,
pl.all().n_unique() > 1). See github.com/pola-rs/polars/issues/11254.
Args:
df: a polars DataFrame
predicates: the boolean expressions to filter on
*more_predicates: additional boolean expressions, specified as
positional arguments
Returns:
df, filtered to the columns where all the boolean expressions in
predicates evaluate to True.
"""
predicates = to_tuple(predicates) + more_predicates
boolean_expression = reduce(lambda a, b: a & b, predicates)
return df.pipe(lambda df: df.select(df.select(boolean_expression)
.unpivot()
.filter(pl.col.value)
['variable']
.to_list()))
def map_df(df, map_col, other_df, key_col, value_col, *,
retain_missing=False):
"""
Maps df[map_col] based on the mapping other_df[key_col] ->
other_df[value_col].
In other words, for each element of df[map_col], check if it's in
other_df[key_col], and if so, replace it with the corresponding entry of
other_df[value_col].
Equivalent to df.with_columns(pl.col(map_col).replace_strict(dict(zip(
other_df[key_col], other_df[value_col])), default=pl.first() if
retain_missing else None)).
Implementation detail: uses a join, but prefixes other_df's key_col and
value_col with "__MAP_DF_" to handle the possibility that map_col might
have the same name as key_col or value_col.
Args:
df: a polars DataFrame
map_col: a column in df
other_df: another polars DataFrame
key_col: a column in other_df with the mapping keys; all values must be
unique, although this is not checked, for speed
value_col: a column in other_df with the mapping values
retain_missing: if False, sets elements of map_col that don't appear in
key_col to null; if True, leaves them unchanged
Returns:
df with map_col transformed so that each of its values that are in
key_col are transformed to the corresponding value in value_col.
"""
if key_col == value_col:
raise ValueError(f'Both key_col and value_col are set to the column '
f'name "{key_col}"')
if isinstance(df, pl.LazyFrame):
other_df = other_df.lazy()
if isinstance(other_df, pl.LazyFrame):
df = df.lazy()
prefix = '__MAP_DF_'
df = df.join(other_df.select(pl.col(key_col, value_col)
.name.prefix(prefix)),
left_on=map_col, right_on=prefix + key_col, how='left')
if retain_missing:
df = df.with_columns(pl.col(prefix + value_col)
.fill_null(pl.col(map_col)))