Coverage for mddb_workflow / utils / arg_cksum.py: 87%

1import json 

2 

3from mddb_workflow.utils.auxiliar import safe_hasattr 

4from mddb_workflow.utils.file import File 

5from mddb_workflow.utils.selections import Selection 

6from mddb_workflow.utils.structures import Structure 

7from mddb_workflow.utils.register import Register 

8from mddb_workflow.utils.cache import Cache 

9from mddb_workflow.utils.database import Database 

10from mddb_workflow.utils.mda_spells import get_mda_universe_cksum 

11from mddb_workflow.utils.type_hints import * 

12 

13from MDAnalysis.core.universe import Universe 

14 

15# Note that it is more convinient not to have this function among auxiliar functions 

16# We depend on class types which depend on the auxiliar module 

17 

18# Given a value of any type, generate a 'cksum' like code which is reproducible 

19# This is used to compare values between different runs without having to store the whole value 

20def get_cksum_id (value) -> Optional[int | float | str]: 

21 # Nones remain as they are 

22 if value == None: return None 

23 # Ge the value type 

24 value_type = type(value) 

25 # For numbers simply use the number itself 

26 if value_type == int or value_type == float or value_type == bool: return value 

27 # For strings, sum the ordinal numbers of every letter 

28 if value_type == str: return f'{sum(map(ord, value))} -> {len(value)}' 

29 # For exceptions simply keep the excepction message 

30 if value_type == Exception: return str(value) 

31 # For objects, stringify them and then do the same that with strings 

32 if value_type in { list, dict, tuple }: 

33 stringifyed = json.dumps(value, default = lambda o: o.__repr__()) 

34 return get_cksum_id(stringifyed) 

35 # If we have a set then make a list and sort it alphabetically 

36 # Thus we make sure the order is coherent between different 

37 if value_type == set: 

38 standard = list(value).sort() 

39 return get_cksum_id(standard) 

40 # For files use file last modification time and size 

41 if isinstance(value, File): return value.get_cksum(unsafe=True) 

42 # For the parsed structure 

43 if isinstance(value, Structure): 

44 pdb_content = value.generate_pdb() 

45 return get_cksum_id(pdb_content) 

46 # For a MDAnalysis universe 

47 if isinstance(value, Universe): return get_mda_universe_cksum(value) 

48 # For the parsed structure 

49 if isinstance(value, Selection): return f'{len(value.atom_indices)}-{sum(value.atom_indices)}' 

50 # For handler class instances it makes not sense making a comparision 

51 # WARNING: If they are used as input then make sure it is only being "used" and not "read" 

52 # Otherwise, the content will be not compared between runs (e.g. warnings in the register) 

53 if isinstance(value, Register): return True 

54 if isinstance(value, Cache): return True 

55 if isinstance(value, Database): return str(value) 

56 if safe_hasattr(value, '__class__'): 

57 if value.__class__.__name__ == 'Project': return True 

58 if value.__class__.__name__ == 'MD': return True 

59 if value.__class__.__name__ == 'Task': return True 

60 # For functions get some parsed code variables and get thier cksums 

61 # We check for functions in the last place because some classes may be callable as well (e.g. Taks) 

62 if callable(value): 

63 return get_cksum_id(f'{value.__code__.co_code} {value.__code__.co_consts}') 

64 # If the value has non of previous types then we complain 

65 raise TypeError(f'Non supported type "{value_type}" for cksum id: {value}')