class RV:
def make_noise_threshold(
self,
input_space: tuple[MapDomain[AtomDomain[TK], AtomDomain[IBig]], MI],
threshold: IBig,
) -> Measurement[
MapDomain[AtomDomain[TK], AtomDomain[IBig]], HashMap[TK, IBig], MI, MO
]:
input_domain, input_metric = input_space
output_measure = MO.default()
threshold_magnitude = threshold.into_parts()[1] privacy_map = self.noise_threshold_privacy_map( input_metric, output_measure, threshold_magnitude
)
match threshold.sign():
case Sign.Positive:
inner = Ordering.Less
case Sign.Negative:
inner = Ordering.Greater
def function(data: HashMap[TK, IBig]) -> HashMap[TK, IBig]:
out = []
for k, v in data.items():
v = self.sample(v)
if v.cmp(threshold) != inner:
out.append((k, v))
random.shuffle(out) return dict(out)
return Measurement.new(
input_domain,
Function.new_fallible(function),
input_metric,
output_measure,
privacy_map,
)