-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathnotebook_utils_old.py
7313 lines (5973 loc) · 263 KB
/
notebook_utils_old.py
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
#!/usr/bin/env python
# Utility Functions to run Jupyter notebooks.
# Dave Babbitt <[email protected]>
# Author: Dave Babbitt, Machine Learning Engineer
# coding: utf-8
# Soli Deo gloria
"""
Run this in a Git Bash terminal if you push anything:
cd ~/OneDrive/Documents/GitHub/notebooks/sh
./update_share_submodules.sh
"""
from bs4 import BeautifulSoup as bs
from datetime import timedelta
from numpy import nan
from os import (
listdir as listdir, makedirs as makedirs, path as osp, remove as remove,
walk as walk
)
from pandas import (
DataFrame, Series, concat, get_dummies, notnull, read_csv, read_pickle,
to_datetime, read_html
)
from re import (
IGNORECASE, Pattern, split, sub
)
import humanize
import inspect
import math
import matplotlib.pyplot as plt
import numpy as np
import os
import pandas as pd
import pkgutil
import re
import seaborn as sns
import subprocess
import sys
import time
import urllib
import warnings
try:
import dill as pickle
except Exception:
import pickle
warnings.filterwarnings('ignore')
# Check for presence of 'get_ipython' function (exists in Jupyter)
try:
get_ipython()
from IPython.display import display
except NameError:
def display(message):
"""
Display a message. If IPython's display is unavailable, fall back to
printing.
Parameters:
message (str): The message to display.
"""
print(message)
# Check if pandas is installed and import relevant functions
try:
from pandas.core.arrays.numeric import is_integer_dtype, is_float_dtype
def is_integer(srs):
"""
Check if the given pandas Series has an integer data type.
Parameters:
srs (pd.Series): A pandas Series to check.
Returns:
bool: True if the Series contains integers, False otherwise.
"""
return is_integer_dtype(srs)
def is_float(srs):
"""
Check if the given pandas Series has a float data type.
Parameters:
srs (pd.Series): A pandas Series to check.
Returns:
bool: True if the Series contains floats, False otherwise.
"""
return is_float_dtype(srs)
except Exception:
def is_integer(srs):
"""
Check if the given list-like object contains integer values.
Parameters:
srs (pd.Series or list): A pandas Series or list-like object to
check.
Returns:
bool: True if any of the values are of integer type, False
otherwise.
"""
return any(map(
lambda value: np.issubdtype(type(value), np.integer), srs.tolist()
))
def is_float(srs):
"""
Check if the given list-like object contains float values.
Parameters:
srs (pd.Series or list): A pandas Series or list-like object to
check.
Returns:
bool: True if any of the values are of float type, False otherwise.
"""
return any(map(lambda value: np.issubdtype(
type(value), np.floating
), srs.tolist()))
class NotebookUtilities(object):
"""
This class implements the core of the utility
functions needed to install and run GPTs and
also what is common to running Jupyter notebooks.
Example:
# Add the path to the shared utilities directory
import os.path as osp, os as os
# Define the shared folder path using join for better compatibility
shared_folder = osp.abspath(osp.join(
osp.dirname(__file__), os.pardir, os.pardir, os.pardir, 'share'
))
# Add the shared folder to system path if it's not already included
import sys
if shared_folder not in sys.path:
sys.path.insert(1, shared_folder)
# Attempt to import the Storage object
try:
from notebook_utils import NotebookUtilities
except ImportError as e:
print(f"Error importing NotebookUtilities: {e}")
# Initialize with data and saves folder paths
nu = NotebookUtilities(
data_folder_path=osp.abspath(osp.join(
osp.dirname(__file__), os.pardir, 'data'
)),
saves_folder_path=osp.abspath(osp.join(
osp.dirname(__file__), os.pardir, 'saves'
))
)
"""
def __init__(
self, data_folder_path=None, saves_folder_path=None, verbose=False
):
# Create the data folder if it doesn't exist
if data_folder_path is None:
self.data_folder = osp.join(os.pardir, 'data')
else:
self.data_folder = data_folder_path
makedirs(self.data_folder, exist_ok=True)
if verbose:
print(
'data_folder: {}'.format(osp.abspath(self.data_folder)),
flush=True
)
# Create the saves folder if it doesn't exist
if saves_folder_path is None:
self.saves_folder = osp.join(os.pardir, 'saves')
else:
self.saves_folder = saves_folder_path
makedirs(self.saves_folder, exist_ok=True)
if verbose:
print(
'saves_folder: {}'.format(osp.abspath(self.saves_folder)),
flush=True
)
self.pip_command_str = f'{sys.executable} -m pip'
# Assume this is instantiated in a subfolder one below the main
self.github_folder = osp.dirname(osp.abspath(osp.curdir))
# Create the assumed directories
self.data_csv_folder = osp.join(self.data_folder, 'csv')
makedirs(name=self.data_csv_folder, exist_ok=True)
self.saves_csv_folder = osp.join(self.saves_folder, 'csv')
makedirs(name=self.saves_csv_folder, exist_ok=True)
self.saves_mp3_folder = osp.join(self.saves_folder, 'mp3')
makedirs(name=self.saves_mp3_folder, exist_ok=True)
self.saves_pickle_folder = osp.join(self.saves_folder, 'pkl')
makedirs(name=self.saves_pickle_folder, exist_ok=True)
self.saves_text_folder = osp.join(self.saves_folder, 'txt')
makedirs(name=self.saves_text_folder, exist_ok=True)
self.saves_wav_folder = osp.join(self.saves_folder, 'wav')
makedirs(name=self.saves_wav_folder, exist_ok=True)
self.saves_png_folder = osp.join(self.saves_folder, 'png')
makedirs(name=self.saves_png_folder, exist_ok=True)
self.txt_folder = osp.join(self.data_folder, 'txt')
makedirs(self.txt_folder, exist_ok=True)
# Create the model directories
self.bin_folder = osp.join(self.data_folder, 'bin')
makedirs(self.bin_folder, exist_ok=True)
self.cache_folder = osp.join(self.data_folder, 'cache')
makedirs(self.cache_folder, exist_ok=True)
self.data_models_folder = osp.join(self.data_folder, 'models')
makedirs(name=self.data_models_folder, exist_ok=True)
self.db_folder = osp.join(self.data_folder, 'db')
makedirs(self.db_folder, exist_ok=True)
self.graphs_folder = osp.join(self.saves_folder, 'graphs')
makedirs(self.graphs_folder, exist_ok=True)
self.indices_folder = osp.join(self.saves_folder, 'indices')
makedirs(self.indices_folder, exist_ok=True)
# Ensure the Scripts folder is in PATH
self.anaconda_folder = osp.dirname(sys.executable)
self.scripts_folder = osp.join(self.anaconda_folder, 'Scripts')
if self.scripts_folder not in sys.path:
sys.path.insert(1, self.scripts_folder)
# Handy list of the different types of encodings
self.encoding_types_list = ['utf-8', 'latin1', 'iso8859-1']
self.encoding_type = self.encoding_types_list[0]
self.encoding_errors_list = ['ignore', 'replace', 'xmlcharrefreplace']
self.encoding_error = self.encoding_errors_list[2]
self.decoding_types_list = [
'ascii', 'cp037', 'cp437', 'cp863', 'utf_32', 'utf_32_be',
'utf_32_le', 'utf_16', 'utf_16_be', 'utf_16_le', 'utf_7',
'utf_8', 'utf_8_sig', 'latin1', 'iso8859-1'
]
self.decoding_type = self.decoding_types_list[11]
self.decoding_errors_list = self.encoding_errors_list.copy()
self.decoding_error = self.decoding_errors_list[2]
# Regular expressions to determine URL from file path
s = r'\b(https?|file)://[-A-Z0-9+&@#/%?=~_|$!:,.;]*[A-Z0-9+&@#/%=~_|$]'
self.url_regex = re.compile(s, IGNORECASE)
s = r'\b[c-d]:\\(?:[^\\/:*?"<>|\x00-\x1F]{0,254}'
s += r'[^.\\/:*?"<>|\x00-\x1F]\\)*(?:[^\\/:*?"<>'
s += r'|\x00-\x1F]{0,254}[^.\\/:*?"<>|\x00-\x1F])'
self.filepath_regex = re.compile(s, IGNORECASE)
# Compile the pattern for identifying function definitions
self.simple_defs_regex = re.compile(r'\bdef ([a-z0-9_]+)\(')
# Create a pattern to match function definitions
self.ipynb_defs_regex = re.compile('\\s+"def ([a-z0-9_]+)\\(')
# Create a regex for finding instance methods and self usage
s = '^ def ([a-z]+[a-z_]+)\\(\\s*self,\\s+(?:[^\\)]+)\\):'
self.instance_defs_regex = re.compile(s, re.MULTILINE)
# Create a regex to search for self references within function bodies
self.self_regex = re.compile('\\bself\\b')
# Compile regex to find all unprefixed comments in the source code
self.comment_regex = re.compile('^( *)# ([^\r\n]+)', re.MULTILINE)
# Compile a regex pattern to match non-alphanumeric characters
self.lower_ascii_regex = re.compile('[^a-z0-9]+')
# Various aspect ratios
self.facebook_aspect_ratio = 1.91
self.twitter_aspect_ratio = 16/9
try:
from pysan.elements import get_alphabet
self.get_alphabet = get_alphabet
except Exception:
self.get_alphabet = lambda sequence: set(sequence)
# Module lists
self.object_evaluators = [
fn for fn in dir(inspect) if fn.startswith('is')
]
module_paths = sorted([
path
for path in sys.path
if path and not path.startswith(osp.dirname(__file__))
])
self.standard_lib_modules = sorted([
module_info.name
for module_info in pkgutil.iter_modules(path=module_paths)
])
# -------------------
# Numeric Functions
# -------------------
@staticmethod
def float_to_ratio(value, tolerance=0.01):
"""
Convert a float to a ratio of two integers, approximating the float
within a specified tolerance.
Parameters:
value (float): The float to convert.
tolerance (float, optional):
The acceptable difference between the float and the ratio.
Default is 0.01.
Returns:
tuple:
A tuple of two integers (numerator, denominator) representing
the ratio that approximates the float.
"""
# Use helper function to check if the fraction is within tolerance
from fractions import Fraction
def is_within_tolerance(numerator, denominator, value, tolerance):
"""
Check if the ratio (numerator/denominator) is within the given
tolerance of the target value.
"""
return abs(numerator / denominator - value) <= tolerance
# Validate input
if not isinstance(value, (float, int)):
raise TypeError('Input value must be a float or an integer.')
if not isinstance(tolerance, float) or tolerance <= 0:
raise ValueError('Tolerance must be a positive float.')
# Use the Fraction module to find an approximate fraction
fraction = Fraction(value).limit_denominator()
numerator, denominator = (fraction.numerator, fraction.denominator)
# Adjust the fraction only if needed
while not is_within_tolerance(
numerator, denominator, value, tolerance
):
numerator += 1
if is_within_tolerance(numerator, denominator, value, tolerance):
break
denominator += 1
return (numerator, denominator)
# -------------------
# String Functions
# -------------------
@staticmethod
def compute_similarity(a, b):
"""
Calculate the similarity between two strings.
Parameters:
a (str): The first string.
b (str): The second string.
Returns:
float
The similarity between the two strings, as a float between 0
and 1.
"""
from difflib import SequenceMatcher
return SequenceMatcher(None, str(a), str(b)).ratio()
@staticmethod
def get_first_year_element(x):
"""
Extract the first year element from a given string, potentially
containing multiple date or year formats.
Parameters:
x (str): The input string containing potential year information.
Returns:
int or float
The extracted first year element, or NaN if no valid year
element is found.
"""
# Split the input string using various separators
stripped_list = split('( |/|\\x96|\\u2009|-|\\[)', str(x), 0)
# Remove non-numeric characters from each element in the stripped list
stripped_list = [sub('\\D+', '', x) for x in stripped_list]
# Filter elements with lengths between 3 and 4, as likely to be years
stripped_list = [
x for x in stripped_list if len(x) >= 3 and len(x) <= 4
]
try:
# Identify the index of the 1st numeric in the stripped list
numeric_list = [x.isnumeric() for x in stripped_list]
# If a numeric substring is found, extract the first numeric value
if True in numeric_list:
idx = numeric_list.index(True, 0)
first_numeric = int(stripped_list[idx])
# If no numeric substring is found, raise an exception
else:
raise Exception('No numeric year element found')
# Handle exceptions and return the first substring if no numeric
except Exception:
# If there are any substrings, return the 1st one as the year
if stripped_list:
first_numeric = int(stripped_list[0])
# If there are no substrings, return NaN
else:
first_numeric = np.nan
return first_numeric
@staticmethod
def format_timedelta(time_delta):
"""
Format a time delta object to a string in the
format '0 sec', '30 sec', '1 min', '1:30', '2 min', etc.
Parameters:
time_delta: A time delta object representing a duration.
Returns:
A string representing the formatted time delta in a human-readable
format: '0 sec', '30 sec', '1 min', '1:30', '2 min', etc.
"""
# Extract total seconds from the time delta object
seconds = time_delta.total_seconds()
# Calculate the number of whole minutes
minutes = int(seconds // 60)
# Calculate the remaining seconds after accounting for minutes
seconds = int(seconds % 60)
# Format the output string for zero minutes, showing only seconds
if minutes == 0:
return f'{seconds} sec'
# If there are minutes and seconds, return in 'min:sec' format
elif seconds > 0:
return f'{minutes}:{seconds:02}' # noqa: E231
# If there are only minutes, return in 'min' format
else:
return f'{minutes} min'
@staticmethod
def outline_chars(text_str, verbose=False):
ord_list = []
for char in list(text_str):
i = ord(char)
if i >= ord('a'):
i += (ord('𝕒') - ord('a'))
elif i >= ord('A'):
i += (ord('𝔸') - ord('A'))
if verbose:
print(f'{char} or {ord(char)}: {i} or {chr(i)}')
ord_list.append(i)
return ''.join([chr(i) for i in ord_list])
# -------------------
# List Functions
# -------------------
@staticmethod
def conjunctify_nouns(noun_list=None, and_or='and', verbose=False):
"""
Concatenate a list of nouns into a grammatically correct string with
specified conjunctions.
Parameters:
noun_list (list or str): A list of nouns to be concatenated.
and_or (str, optional): The conjunction used to join the nouns.
Default is 'and'.
verbose (bool, optional):
Whether to print debug or status messages. Defaults to False.
Returns:
str
A string containing the concatenated nouns with appropriate
conjunctions.
Example:
noun_list = ['apples', 'oranges', 'bananas']
conjunction = 'and'
result = conjunctify_nouns(noun_list, and_or=conjunction)
print(result)
Output: 'apples, oranges, and bananas'
"""
# Handle special cases where noun_list is None or not a list
if noun_list is None:
return ''
if not isinstance(noun_list, list):
noun_list = list(noun_list)
# If there are more than two nouns in the list
if len(noun_list) > 2:
# Create a noun string of the last element in the list
last_noun_str = noun_list[-1]
# Create comma-delimited but-last string out of the rest
but_last_nouns_str = ', '.join(noun_list[:-1])
# Join the but-last string and the last noun string with `and_or`
list_str = f', {and_or} '.join(
[but_last_nouns_str, last_noun_str]
)
# If just two nouns in the list, join the nouns with `and_or`
elif len(noun_list) == 2:
list_str = f' {and_or} '.join(noun_list)
# If there is just one noun in the list, make that the returned string
elif len(noun_list) == 1:
list_str = noun_list[0]
# Otherwise, make a blank the returned string
else:
list_str = ''
# Print debug output if verbose
if verbose:
print(
f'noun_list="{noun_list}", and_or="{and_or}", '
f'list_str="{list_str}"'
)
# Return the conjuncted noun list
return list_str
def get_jitter_list(self, ages_list):
"""
Generate a list of jitter values for plotting age data points with a
scattered plot.
Parameters:
ages_list (list): A list of ages for which jitter values are
generated.
Returns:
list of float
A list of jitter values corresponding to the input ages.
"""
# Initialize an empty list to store jitter values
jitter_list = []
# Iterate over the list of age groups
for splits_list in self.split_list_by_gap(ages_list):
# If multiple ages in group, calculate jitter values for each age
if len(splits_list) > 1:
# Generate jitter values using cut and extend
jitter_list.extend(
pd.cut(
np.array(
[min(splits_list) - 0.99, max(splits_list) + 0.99]
),
len(splits_list) - 1,
retbins=True
)[1]
)
# If only one age in a group, add that age as the jitter value
else:
jitter_list.extend(splits_list)
# Return the list of jitter values
return jitter_list
@staticmethod
def split_list_by_gap(ages_list, value_difference=1, verbose=False):
"""
Divide a list of ages into sublists based on gaps in the age sequence.
Parameters:
ages_list (list of int or float): A list of ages to be split into
sublists.
Returns:
list of lists of int or float
A list of sublists, each containing consecutive ages.
"""
# List to store sublists of consecutive ages
splits_list = []
# Temporary list to store the current consecutive ages
current_list = []
# Initialize with a value lower than the first age
previous_age = ages_list[0] - value_difference
# Iterate over the list of ages
for age in ages_list:
# Check if there is a gap between current age and previous age
if age - previous_age > value_difference:
# Append the current_list to splits_list
splits_list.append(current_list)
# Reset the current_list
current_list = []
# Add the current age to the current_list
current_list.append(age)
# Update the previous_age
previous_age = age
# Append the last current_list to splits_list
splits_list.append(current_list)
# Return the list of sublists of ages
return splits_list
@staticmethod
def count_ngrams(actions_list, highlighted_ngrams):
"""
Count the occurrences of a sequence of elements (n-grams) in a list.
This static method traverses through the `actions_list` and counts
how many times the sequence `highlighted_ngrams` appears. It is
useful for analyzing the frequency of specific patterns within a
list of actions or tokens.
Parameters:
actions_list (list):
A list of elements in which to count the occurrences of the
n-gram.
highlighted_ngrams (list):
The sequence of elements (n-gram) to count occurrences of.
Returns:
int:
The count of how many times `highlighted_ngrams` occurs in
`actions_list`.
Examples:
actions = ['jump', 'run', 'jump', 'run', 'jump']
ngrams = ['jump', 'run']
nu.count_ngrams(actions, ngrams) # 2
"""
# Initialize the count of n-gram occurrences
count = 0
# Calculate the range for the loop to avoid IndexErrors
range_limit = len(actions_list) - len(highlighted_ngrams) + 1
# Loop over the actions_list at that window size to find occurrences
for i in range(range_limit):
# Check if the current slice matches the highlighted_ngrams
if actions_list[
i:i + len(highlighted_ngrams)
] == highlighted_ngrams:
# Increment the count if a match is found
count += 1
return count
@staticmethod
def get_sequences_by_count(tg_dict, count=4):
"""
Get sequences from the input dictionary based on a specific sequence
length.
Parameters:
tg_dict (dict): Dictionary containing sequences.
count (int, optional): Desired length of sequences to filter.
Default is 4.
Returns:
list: List of sequences with the specified length.
Raises:
AssertionError: If no sequences of the specified length are found
in the dictionary.
"""
# Convert the lengths in the dictionary values into value counts
value_counts = Series(
[len(actions_list) for actions_list in tg_dict.values()]
).value_counts()
# Get desired sequence length of exactly count sequences
value_counts_list = value_counts[value_counts == count].index.tolist()
s = f"You don't have exactly {count} sequences of the same length in"
s += " the dictionary"
assert value_counts_list, s
sequences = [
actions_list
for actions_list in tg_dict.values()
if (len(actions_list) == value_counts_list[0])
]
return sequences
@staticmethod
def get_shape(list_of_lists):
"""
Return the shape of a list of lists, assuming the sublists are all of
the same length.
Parameters:
list_of_lists: A list of lists.
Returns:
A tuple representing the shape of the list of lists.
"""
# Check if the list of lists is empty
if not list_of_lists:
return ()
# Get the length of the first sublist
num_cols = len(list_of_lists[0])
# Check if all of the sublists are the same length
for sublist in list_of_lists:
if len(sublist) != num_cols:
raise ValueError(
'All of the sublists must be the same length.'
)
# Return a tuple representing the shape of the list of lists
return (len(list_of_lists), num_cols)
@staticmethod
def split_list_by_exclusion(
splitting_indices_list, excluded_indices_list=[]
):
"""
Split a list of row indices into a list of lists, where each inner list
contains a contiguous sequence of indices that are not in the excluded
indices list.
Parameters:
splitting_indices_list: A list of row indices to split.
excluded_indices_list:
A list of row indices that should be considered excluded.
Empty by default.
Returns:
A list of lists, where each inner list contains a contiguous
sequence of indices that are not in the excluded indices.
"""
# Initialize the output list
split_list = []
# Initialize the current list
current_list = []
# Iterate over the splitting indices list
for current_idx in range(
int(min(splitting_indices_list)),
int(max(splitting_indices_list)) + 1
):
# Check that the current index is in the splitting indices list
if (
current_idx in splitting_indices_list
):
# Check that the current index is not in the excluded list
if (
current_idx not in excluded_indices_list
):
# If so, add it to the current list
current_list.append(current_idx)
# Otherwise
else:
# If the current list is not empty
if current_list:
# Add it to the split list
split_list.append(current_list)
# And start a new current list
current_list = []
# If the current list is not empty
if current_list:
# Add it to the split list
split_list.append(current_list)
# Return the split list
return split_list
def check_4_doubles(self, item_list, verbose=False):
"""
Find and compare items within a list to identify similar pairs.
This method compares each item in the list with every other item
to find the most similar item based on a computed similarity
metric. The results are returned in a DataFrame containing pairs
of items and their similarity scores and byte representations.
Parameters:
item_list (list):
The list of items to be compared for similarity.
verbose (bool, optional):
Whether to print debug or status messages. Defaults to False.
Returns:
pandas.DataFrame:
A DataFrame containing columns for the first item, second item,
their byte representations, and the maximum similarity score
found for each pair.
"""
# Start the timer if verbose is enabled
if verbose:
t0 = time.time()
rows_list = []
n = len(item_list)
# Iterate over each item in the list
for i in range(n-1):
# Get the current item to compare with others
first_item = item_list[i]
# Initialize the maximum similarity score
max_similarity = 0.0
# Initialize the item with the highest similarity to the current
max_item = first_item
# Compare the current item with the rest of the items in the list
for j in range(i+1, n):
# Get the item to compare against
second_item = item_list[j]
# Ensure items are not identical before similarity calculation
if first_item != second_item:
# Compute the similarity between the two items
this_similarity = self.compute_similarity(
str(first_item), str(second_item)
)
# Has a higher similarity been found?
if this_similarity > max_similarity:
# Update max_similarity and max_item
max_similarity = this_similarity
max_item = second_item
# Create row dict to store information for each similar item pair
row_dict = {'first_item': first_item, 'second_item': max_item}
# Convert items to byte arrays for string representation
row_dict['first_bytes'] = '-'.join([str(x) for x in bytearray(
str(first_item), encoding=self.encoding_type, errors='replace'
)])
row_dict['second_bytes'] = '-'.join([str(x) for x in bytearray(
str(max_item), encoding=self.encoding_type, errors='replace'
)])
row_dict['max_similarity'] = max_similarity
# Add the row dictionary to the list of rows
rows_list.append(row_dict)
# Define the column names for the resulting DataFrame
column_list = [
'first_item', 'second_item', 'first_bytes', 'second_bytes',
'max_similarity'
]
# Create the DataFrame from the list of row dictionaries
item_similarities_df = DataFrame(rows_list, columns=column_list)
# Display end time for performance measurement (if verbose)
if verbose:
t1 = time.time()
print(
f'Finished in {t1 - t0:.2f} ' # noqa: E231
f'seconds ({time.ctime(t1)})'
)
return item_similarities_df
def check_for_typos(
self, left_list, right_list,
rename_dict={'left_item': 'left_item', 'right_item': 'right_item'},
verbose=False
):
"""
Check the closest names for typos by comparing items from left_list
with items from right_list and computing their similarities.
Parameters:
left_list (list): List containing items to be compared (left side).
right_list (list):
List containing items to be compared (right side).
rename_dict (dict, optional):
Dictionary specifying custom column names in the output
DataFrame. Default is {'left_item': 'left_item', 'right_item':
'right_item'}.
verbose (bool, optional):
Whether to print debug or status messages. Defaults to False.
Returns:
pandas.DataFrame: DataFrame containing columns: 'left_item',
'right_item', and 'max_similarity'.
Example:
commonly_misspelled_words = [
"absence", "consensus", "definitely", "broccoli", "necessary"
]
common_misspellings = [
"absense", "concensus", "definately", "brocolli", "neccessary"
]
typos_df = nu.check_for_typos(
commonly_misspelled_words,
common_misspellings,
rename_dict={
'left_item': 'commonly_misspelled',
'right_item': 'common_misspelling'
}
).sort_values(
[
'max_similarity', 'commonly_misspelled',
'common_misspelling'
],
ascending=[False, True, True]
)
display(typos_df)
"""
# Initialize the time taken for the computation if verbose is True
if verbose:
t0 = time.time()
# Initialize an empty list to store rows of the output data frame
rows_list = []
# Iterate through items in the left list
for left_item in left_list:
max_similarity = 0.0
max_item = left_item
# Iterate through items in the right list
for right_item in right_list:
this_similarity = self.compute_similarity(
left_item, right_item
)
# Find the most similar item
if this_similarity > max_similarity:
max_similarity = this_similarity
max_item = right_item
# Create a dictionary representing a row in the output data frame
row_dict = {
'left_item': left_item,
'right_item': max_item,
'max_similarity': max_similarity
}
# Add the row dictionary to the list of rows
rows_list.append(row_dict)
# Define the column names for the output data frame
column_list = ['left_item', 'right_item', 'max_similarity']