Coverage for mindsdb / integrations / handlers / lightwood_handler / utils.py: 0%

1import gc 

2import json 

3import sys 

4 

5import dill 

6from lightwood.api.high_level import _module_from_code 

7 

8 

9def unpack_jsonai_old_args(json_ai_override): 

10 while '.' in str(list(json_ai_override.keys())): 

11 for k in list(json_ai_override.keys()): 

12 if '.' in k: 

13 nks = k.split('.') 

14 obj = json_ai_override 

15 for nk in nks[:-1]: 

16 if nk not in obj: 

17 obj[nk] = {} 

18 obj = obj[nk] 

19 obj[nks[-1]] = json_ai_override[k] 

20 del json_ai_override[k] 

21 

22 

23def load_predictor(predictor_dict, name): 

24 try: 

25 return dill.loads(predictor_dict['predictor']) 

26 except Exception as e: 

27 module_name = str(e).lstrip("No module named '").split("'")[0] 

28 

29 try: 

30 del sys.modules[module_name] 

31 except Exception: 

32 pass 

33 

34 gc.collect() 

35 _module_from_code(predictor_dict['code'], module_name) 

36 return dill.loads(predictor_dict['predictor']) 

37 

38 

39def rep_recur(org: dict, ovr: dict): 

40 for k in ovr: 

41 if k in org: 

42 if isinstance(org[k], dict) and isinstance(ovr[k], dict): 

43 rep_recur(org[k], ovr[k]) 

44 else: 

45 org[k] = ovr[k] 

46 else: 

47 org[k] = ovr[k] 

48 

49 

50def brack_to_mod(ovr): 

51 if not isinstance(ovr, dict): 

52 if isinstance(ovr, list): 

53 for i in range(len(ovr)): 

54 ovr[i] = brack_to_mod(ovr[i]) 

55 elif isinstance(ovr, str): 

56 if '(' in ovr and ')' in ovr: 

57 mod = ovr.split('(')[0] 

58 args = {} 

59 if '()' not in ovr: 

60 for str_pair in ovr.split('(')[1].split(')')[0].split(','): 

61 k = str_pair.split('=')[0].strip(' ') 

62 v = str_pair.split('=')[1].strip(' ') 

63 args[k] = v 

64 

65 ovr = {'module': mod, 'args': args} 

66 elif '{' in ovr and '}' in ovr: 

67 try: 

68 ovr = json.loads(ovr) 

69 except Exception: 

70 pass 

71 return ovr 

72 else: 

73 for k in ovr.keys(): 

74 ovr[k] = brack_to_mod(ovr[k]) 

75 

76 return ovr