8139141456

Committed 04 Mar 2024 11:03AM UTC coverage: 37.923% (-60.6%) from 98.477%

Build # 8139141456

Build Type

Pull #722

github

Committed by

web-flow

Commit Message

Merge 5663238db into 17e0e9b31

Pull Request Pull Request #722: Fix guide compilation

Run Details

1344 of 3544 relevant lines covered (37.92%)

0.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

24.64

/foolbox/attacks/dataset_attack.py

from typing import Union, Optional, Any, List
import numpy as np
import eagerpy as ep

from ..devutils import atleast_kd

from ..models import Model

from ..distances import Distance

from ..criteria import Criterion

from .base import FlexibleDistanceMinimizationAttack
from .base import T
from .base import get_criterion
from .base import raise_if_kwargs
from .base import verify_input_bounds


class DatasetAttack(FlexibleDistanceMinimizationAttack):
    """Draws randomly from the given dataset until adversarial examples for all
    inputs have been found.

    To pass data form the dataset to this attack, call :meth:`feed()`.
    :meth:`feed()` can be called several times and should only be called with
    batches that are small enough that they can be passed through the model.

    Args:
        distance : Distance measure for which minimal adversarial examples are searched.
    """

    def __init__(self, *, distance: Optional[Distance] = None):
        super().__init__(distance=distance)
        self.raw_inputs: List[ep.Tensor] = []
        self.raw_outputs: List[ep.Tensor] = []
        self.inputs: Optional[ep.Tensor] = None
        self.outputs: Optional[ep.Tensor] = None

    def feed(self, model: Model, inputs: Any) -> None:
        x = ep.astensor(inputs)
        del inputs

        self.raw_inputs.append(x)
        self.raw_outputs.append(model(x))

    def process_raw(self) -> None:
        raw_inputs = self.raw_inputs
        raw_outputs = self.raw_outputs
        assert len(raw_inputs) == len(raw_outputs)
        assert (self.inputs is None) == (self.outputs is None)

        if self.inputs is None:
            if len(raw_inputs) == 0:
                raise ValueError(
                    "DatasetAttack can only be called after data has been provided using 'feed()'"
                )
        elif self.inputs is not None:
            assert self.outputs is not None
            raw_inputs = [self.inputs] + raw_inputs
            raw_outputs = [self.outputs] + raw_outputs

        self.inputs = ep.concatenate(raw_inputs, axis=0)
        self.outputs = ep.concatenate(raw_outputs, axis=0)
        self.raw_inputs = []
        self.raw_outputs = []

    def run(
        self,
        model: Model,
        inputs: T,
        criterion: Union[Criterion, T],
        *,
        early_stop: Optional[float] = None,
        **kwargs: Any,
    ) -> T:
        raise_if_kwargs(kwargs)
        self.process_raw()
        assert self.inputs is not None
        assert self.outputs is not None
        x, restore_type = ep.astensor_(inputs)
        del inputs, kwargs

        verify_input_bounds(x, model)

        criterion = get_criterion(criterion)

        result = x
        found = criterion(x, model(x))

        batch_size = len(x)

        # for every sample try every other sample
        index_pools: List[List[int]] = []
        for i in range(batch_size):
            indices = list(range(batch_size))
            indices.remove(i)
            np.random.shuffle(indices)
            index_pools.append(indices)

        for i in range(batch_size - 1):
            if found.all():
                break

            indices_np = np.array([pool[i] for pool in index_pools])

            xp = self.inputs[indices_np]
            yp = self.outputs[indices_np]
            is_adv = criterion(xp, yp)

            new_found = ep.logical_and(is_adv, found.logical_not())
            result = ep.where(atleast_kd(new_found, result.ndim), xp, result)
            found = ep.logical_or(found, new_found)

        return restore_type(result)

1	from typing import Union, Optional, Any, List	1✔
2	import numpy as np	1✔
3	import eagerpy as ep	1✔
4
5	from ..devutils import atleast_kd	1✔
6
7	from ..models import Model	1✔
8
9	from ..distances import Distance	1✔
10
11	from ..criteria import Criterion	1✔
12
13	from .base import FlexibleDistanceMinimizationAttack	1✔
14	from .base import T	1✔
15	from .base import get_criterion	1✔
16	from .base import raise_if_kwargs	1✔
17	from .base import verify_input_bounds	1✔
18
19
20	class DatasetAttack(FlexibleDistanceMinimizationAttack):	1✔
21	"""Draws randomly from the given dataset until adversarial examples for all
22	inputs have been found.
23
24	To pass data form the dataset to this attack, call :meth:`feed()`.
25	:meth:`feed()` can be called several times and should only be called with
26	batches that are small enough that they can be passed through the model.
27
28	Args:
29	distance : Distance measure for which minimal adversarial examples are searched.
30	"""
31
32	def __init__(self, *, distance: Optional[Distance] = None):	1✔
33	super().__init__(distance=distance)	×
34	self.raw_inputs: List[ep.Tensor] = []	×
35	self.raw_outputs: List[ep.Tensor] = []	×
36	self.inputs: Optional[ep.Tensor] = None	×
37	self.outputs: Optional[ep.Tensor] = None	×
38
39	def feed(self, model: Model, inputs: Any) -> None:	1✔
40	x = ep.astensor(inputs)	×
41	del inputs	×
42
43	self.raw_inputs.append(x)	×
44	self.raw_outputs.append(model(x))	×
45
46	def process_raw(self) -> None:	1✔
47	raw_inputs = self.raw_inputs	×
48	raw_outputs = self.raw_outputs	×
49	assert len(raw_inputs) == len(raw_outputs)	×
50	assert (self.inputs is None) == (self.outputs is None)	×
51
52	if self.inputs is None:	×
53	if len(raw_inputs) == 0:	×
54	raise ValueError(	×
55	"DatasetAttack can only be called after data has been provided using 'feed()'"
56	)
57	elif self.inputs is not None:	×
58	assert self.outputs is not None	×
59	raw_inputs = [self.inputs] + raw_inputs	×
60	raw_outputs = [self.outputs] + raw_outputs	×
61
62	self.inputs = ep.concatenate(raw_inputs, axis=0)	×
63	self.outputs = ep.concatenate(raw_outputs, axis=0)	×
64	self.raw_inputs = []	×
65	self.raw_outputs = []	×
66
67	def run(	1✔
68	self,
69	model: Model,
70	inputs: T,
71	criterion: Union[Criterion, T],
72	*,
73	early_stop: Optional[float] = None,
74	**kwargs: Any,
75	) -> T:
76	raise_if_kwargs(kwargs)	×
77	self.process_raw()	×
78	assert self.inputs is not None	×
79	assert self.outputs is not None	×
80	x, restore_type = ep.astensor_(inputs)	×
81	del inputs, kwargs	×
82
83	verify_input_bounds(x, model)	×
84
85	criterion = get_criterion(criterion)	×
86
87	result = x	×
88	found = criterion(x, model(x))	×
89
90	batch_size = len(x)	×
91
92	# for every sample try every other sample
93	index_pools: List[List[int]] = []	×
94	for i in range(batch_size):	×
95	indices = list(range(batch_size))	×
96	indices.remove(i)	×
97	np.random.shuffle(indices)	×
98	index_pools.append(indices)	×
99
100	for i in range(batch_size - 1):	×
101	if found.all():	×
102	break	×
103
104	indices_np = np.array([pool[i] for pool in index_pools])	×
105
106	xp = self.inputs[indices_np]	×
107	yp = self.outputs[indices_np]	×
108	is_adv = criterion(xp, yp)	×
109
110	new_found = ep.logical_and(is_adv, found.logical_not())	×
111	result = ep.where(atleast_kd(new_found, result.ndim), xp, result)	×
112	found = ep.logical_or(found, new_found)	×
113
114	return restore_type(result)	×

bethgelab / foolbox / 8139141456

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous