Coverage for valmix/knob.py: 100%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# 

4# SPDX-License-Identifier: Apache-2.0 

5# Copyright 2024 Inria 

6 

7"""Knob interface.""" 

8 

9from multiprocessing.sharedctypes import Synchronized as Value 

10from typing import Sequence, Union 

11 

12T = Union[int, float] 

13 

14 

15class Knob: 

16 """Name and list values for a target quantity.""" 

17 

18 def __init__( 

19 self, 

20 name: str, 

21 value: Value, 

22 values: Sequence[T], 

23 ) -> None: 

24 """Create a new knob. 

25 

26 Args: 

27 name: Display name of the value. 

28 value: Actual multiprocessing synchronized shared object. 

29 values: Values the internal value may take. 

30 """ 

31 nb_values = len(values) 

32 initial_value = value.value 

33 self.__index = nb_values // 2 

34 self.__mp_value = value 

35 self.__name = name 

36 self.__nb_values = nb_values 

37 self.__values = sorted(values) 

38 # 

39 self.__snap_to(initial_value) 

40 self.__update_mp_value() 

41 

42 def __snap_to(self, snap_value: T) -> None: 

43 """Snap to available value closest to a given one. 

44 

45 Args: 

46 snap_value: Quantity to get close to using values from the set. 

47 """ 

48 self.__index = 0 

49 best_dist = abs(self.__values[0] - snap_value) 

50 for i in range(1, self.__nb_values): 

51 cur_dist = abs(self.__values[i] - snap_value) 

52 if cur_dist < best_dist: 

53 self.__index = i 

54 best_dist = cur_dist 

55 

56 def __update_mp_value(self) -> None: 

57 """Update internal multiprocessing value from current index.""" 

58 new_value = self.__values[self.__index] 

59 with self.__mp_value.get_lock(): 

60 self.__mp_value.value = new_value 

61 

62 def advance(self, step: int) -> None: 

63 """Advance to the next value listed when configuring the knob. 

64 

65 Args: 

66 step: Number of steps to add to the list index. 

67 """ 

68 self.__index += step 

69 if self.__index >= self.__nb_values: 

70 self.__index = self.__nb_values - 1 

71 elif self.__index < 0: 

72 self.__index = 0 

73 self.__update_mp_value() 

74 

75 @property 

76 def current_index(self) -> int: 

77 """Get current index in the list of available values. 

78 

79 Returns: 

80 Current index of the knob. 

81 """ 

82 return self.__index 

83 

84 @property 

85 def value(self) -> T: 

86 """Get current value of the knob. 

87 

88 Returns: 

89 Current value of the knob. 

90 """ 

91 return self.__mp_value.value 

92 

93 @property 

94 def name(self) -> str: 

95 """Knob name to display.""" 

96 return self.__name 

97 

98 @property 

99 def nb_values(self) -> int: 

100 """Get the number of different values that the knob can take. 

101 

102 Returns: 

103 Number of different values that the knob can take. 

104 """ 

105 return self.__nb_values 

106 

107 def __repr__(self) -> str: 

108 """Get a string representation of the knob's value. 

109 

110 Returns: 

111 String representation of the current value of the knob. 

112 """ 

113 v = self.value 

114 if isinstance(v, float): 

115 return f"{v:.2f}" if 1e-2 <= abs(v) < 100.0 else f"{v:.1e}" 

116 return str(self.value)