Coverage for lasso/dimred/svd/pod_functions.py: 97%

1from typing import Tuple, Union 

2 

3import numpy as np 

4from scipy.sparse import csc_matrix 

5from scipy.sparse.linalg import svds 

6from rich.progress import Progress, TaskID 

7 

8from lasso.utils.rich_progress_bars import PlaceHolderBar 

9 

10 

11def svd_step_and_dim(s_mat: np.ndarray, k=10) -> np.ndarray: 

12 """ 

13 Performs a svds operation on the two dimensional s_mat 

14 

15 Parameters 

16 ---------- 

17 s_mat: ndarray 

18 2D array on which the svds operation shall be performed 

19 k: int, 10, optinal. 

20 The size of the POD 

21 

22 Returns 

23 ------- 

24 v: ndarray 

25 Array containing the right reduced order basis 

26 """ 

27 small_mat = csc_matrix(s_mat.astype(np.float64)) 

28 

29 _, _, v = svds(small_mat, k=k) 

30 

31 v = v[::-1, :] 

32 

33 return v 

34 

35 

36def calculate_v_and_betas( 

37 stacked_sub_displ: np.ndarray, 

38 progress_bar: Union[None, Progress, PlaceHolderBar] = None, 

39 task_id: Union[None, TaskID] = None, 

40) -> Union[str, Tuple[np.ndarray, np.ndarray]]: 

41 """ 

42 Calculates the right reduced order Basis V and up to 10 eigenvalues of the subsamples 

43 

44 Parameters 

45 ---------- 

46 stacked_sub_displ: np.ndarray 

47 np.ndarray containing all subsampled displacements 

48 shape must be (samples, timesteps, nodes, dims) 

49 

50 Returns 

51 ------- 

52 v_big: np.ndarray 

53 Reduced order basis to transform betas bag into subsamples 

54 betas: np.ndarray 

55 Projected simulation runs 

56 err_msg: str 

57 Error message if not enough samples where provided 

58 """ 

59 

60 big_mat = stacked_sub_displ.reshape( 

61 ( 

62 stacked_sub_displ.shape[0], 

63 stacked_sub_displ.shape[1], 

64 stacked_sub_displ.shape[2] * stacked_sub_displ.shape[3], 

65 ) 

66 ) 

67 

68 diff_mat = np.stack([big_mat[:, 0, :] for _ in range(big_mat.shape[1])]).reshape( 

69 (big_mat.shape[0], big_mat.shape[1], big_mat.shape[2]) 

70 ) 

71 

72 # We only want the difference in displacement 

73 big_mat = big_mat - diff_mat 

74 

75 k = min(10, big_mat.shape[0] - 1) 

76 if k < 1: 

77 return "Must provide more than 1 sample" 

78 

79 if task_id is None and progress_bar: 

80 return "Progress requires a task ID" 

81 

82 v_big = np.zeros((k, big_mat.shape[1], big_mat.shape[2])) 

83 if progress_bar: 

84 progress_bar.advance(task_id) # type: ignore 

85 for step in range(big_mat.shape[1] - 1): 

86 v_big[:, step + 1] = svd_step_and_dim(big_mat[:, step + 1], k) 

87 progress_bar.advance(task_id) # type: ignore 

88 else: 

89 for step in range(big_mat.shape[1] - 1): 

90 v_big[:, step + 1] = svd_step_and_dim(big_mat[:, step + 1], k) 

91 

92 betas_big = np.einsum("stn, ktn -> stk", big_mat, v_big) 

93 

94 return v_big, betas_big