We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
RANSAC.max_samples_by_conf returns negative numbers
### Describe the bug

This code fails

`
from kornia.geometry import RANSAC

x = RANSAC.max_samples_by_conf(n_inl=100, num_tc=1000, sample_size=7, conf=0.99)
assert x > 0.0
`


### Reproduction steps

```bash
from kornia.geometry import RANSAC

x = RANSAC.max_samples_by_conf(n_inl=100, num_tc=1000, sample_size=7, conf=0.99)
assert x > 0.0
```


### Expected behavior

The returned value should always be positive

### Environment

```shell
Environment agnostic
```


### Additional context

a one-line bugfix will shortly follow

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `kornia/geometry/ransac.py`
Content:
```
1 """Module containing RANSAC modules."""
2 
3 import math
4 from functools import partial
5 from typing import Callable, Optional, Tuple
6 
7 import torch
8 
9 from kornia.core import Device, Module, Tensor, zeros
10 from kornia.core.check import KORNIA_CHECK_SHAPE
11 from kornia.geometry import (
12     find_fundamental,
13     find_homography_dlt,
14     find_homography_dlt_iterated,
15     find_homography_lines_dlt,
16     find_homography_lines_dlt_iterated,
17     symmetrical_epipolar_distance,
18 )
19 from kornia.geometry.homography import (
20     line_segment_transfer_error_one_way,
21     oneway_transfer_error,
22     sample_is_valid_for_homography,
23 )
24 
25 __all__ = ["RANSAC"]
26 
27 
28 class RANSAC(Module):
29     """Module for robust geometry estimation with RANSAC. https://en.wikipedia.org/wiki/Random_sample_consensus.
30 
31     Args:
32         model_type: type of model to estimate: "homography", "fundamental", "fundamental_7pt",
33             "homography_from_linesegments".
34         inliers_threshold: threshold for the correspondence to be an inlier.
35         batch_size: number of generated samples at once.
36         max_iterations: maximum batches to generate. Actual number of models to try is ``batch_size * max_iterations``.
37         confidence: desired confidence of the result, used for the early stopping.
38         max_local_iterations: number of local optimization (polishing) iterations.
39     """
40 
41     def __init__(
42         self,
43         model_type: str = "homography",
44         inl_th: float = 2.0,
45         batch_size: int = 2048,
46         max_iter: int = 10,
47         confidence: float = 0.99,
48         max_lo_iters: int = 5,
49     ) -> None:
50         super().__init__()
51         self.supported_models = ["homography", "fundamental", "fundamental_7pt", "homography_from_linesegments"]
52         self.inl_th = inl_th
53         self.max_iter = max_iter
54         self.batch_size = batch_size
55         self.model_type = model_type
56         self.confidence = confidence
57         self.max_lo_iters = max_lo_iters
58         self.model_type = model_type
59 
60         self.error_fn: Callable[..., Tensor]
61         self.minimal_solver: Callable[..., Tensor]
62         self.polisher_solver: Callable[..., Tensor]
63 
64         if model_type == "homography":
65             self.error_fn = oneway_transfer_error
66             self.minimal_solver = find_homography_dlt
67             self.polisher_solver = find_homography_dlt_iterated
68             self.minimal_sample_size = 4
69         elif model_type == "homography_from_linesegments":
70             self.error_fn = line_segment_transfer_error_one_way
71             self.minimal_solver = find_homography_lines_dlt
72             self.polisher_solver = find_homography_lines_dlt_iterated
73             self.minimal_sample_size = 4
74         elif model_type == "fundamental":
75             self.error_fn = symmetrical_epipolar_distance
76             self.minimal_solver = find_fundamental
77             self.minimal_sample_size = 8
78             self.polisher_solver = find_fundamental
79         elif model_type == "fundamental_7pt":
80             self.error_fn = symmetrical_epipolar_distance
81             self.minimal_solver = partial(find_fundamental, method="7POINT")
82             self.minimal_sample_size = 7
83             self.polisher_solver = find_fundamental
84         else:
85             raise NotImplementedError(f"{model_type} is unknown. Try one of {self.supported_models}")
86 
87     def sample(self, sample_size: int, pop_size: int, batch_size: int, device: Device = torch.device("cpu")) -> Tensor:
88         """Minimal sampler, but unlike traditional RANSAC we sample in batches to get benefit of the parallel
89         processing, esp.
90 
91         on GPU.
92         """
93         rand = torch.rand(batch_size, pop_size, device=device)
94         _, out = rand.topk(k=sample_size, dim=1)
95         return out
96 
97     @staticmethod
98     def max_samples_by_conf(n_inl: int, num_tc: int, sample_size: int, conf: float) -> float:
99         """Formula to update max_iter in order to stop iterations earlier
100         https://en.wikipedia.org/wiki/Random_sample_consensus."""
101         eps = 1e-9
102         if num_tc <= sample_size:
103             return 1.0
104         if n_inl == num_tc:
105             return 1.0
106         return math.log(1.0 - conf) / max(eps, math.log(max(eps, 1.0 - math.pow(n_inl / num_tc, sample_size))))
107 
108     def estimate_model_from_minsample(self, kp1: Tensor, kp2: Tensor) -> Tensor:
109         batch_size, sample_size = kp1.shape[:2]
110         H = self.minimal_solver(kp1, kp2, torch.ones(batch_size, sample_size, dtype=kp1.dtype, device=kp1.device))
111         return H
112 
113     def verify(self, kp1: Tensor, kp2: Tensor, models: Tensor, inl_th: float) -> Tuple[Tensor, Tensor, float]:
114         if len(kp1.shape) == 2:
115             kp1 = kp1[None]
116         if len(kp2.shape) == 2:
117             kp2 = kp2[None]
118         batch_size = models.shape[0]
119         if self.model_type == "homography_from_linesegments":
120             errors = self.error_fn(kp1.expand(batch_size, -1, 2, 2), kp2.expand(batch_size, -1, 2, 2), models)
121         else:
122             errors = self.error_fn(kp1.expand(batch_size, -1, 2), kp2.expand(batch_size, -1, 2), models)
123         inl = errors <= inl_th
124         models_score = inl.to(kp1).sum(dim=1)
125         best_model_idx = models_score.argmax()
126         best_model_score = models_score[best_model_idx].item()
127         model_best = models[best_model_idx].clone()
128         inliers_best = inl[best_model_idx]
129         return model_best, inliers_best, best_model_score
130 
131     def remove_bad_samples(self, kp1: Tensor, kp2: Tensor) -> Tuple[Tensor, Tensor]:
132         """"""
133         # ToDo: add (model-specific) verification of the samples,
134         # E.g. constraints on not to be a degenerate sample
135         if self.model_type == "homography":
136             mask = sample_is_valid_for_homography(kp1, kp2)
137             return kp1[mask], kp2[mask]
138         return kp1, kp2
139 
140     def remove_bad_models(self, models: Tensor) -> Tensor:
141         # ToDo: add more and better degenerate model rejection
142         # For now it is simple and hardcoded
143         main_diagonal = torch.diagonal(models, dim1=1, dim2=2)
144         mask = main_diagonal.abs().min(dim=1)[0] > 1e-4
145         return models[mask]
146 
147     def polish_model(self, kp1: Tensor, kp2: Tensor, inliers: Tensor) -> Tensor:
148         # TODO: Replace this with MAGSAC++ polisher
149         kp1_inl = kp1[inliers][None]
150         kp2_inl = kp2[inliers][None]
151         num_inl = kp1_inl.size(1)
152         model = self.polisher_solver(
153             kp1_inl, kp2_inl, torch.ones(1, num_inl, dtype=kp1_inl.dtype, device=kp1_inl.device)
154         )
155         return model
156 
157     def validate_inputs(self, kp1: Tensor, kp2: Tensor, weights: Optional[Tensor] = None) -> None:
158         if self.model_type in ["homography", "fundamental"]:
159             KORNIA_CHECK_SHAPE(kp1, ["N", "2"])
160             KORNIA_CHECK_SHAPE(kp2, ["N", "2"])
161             if not (kp1.shape[0] == kp2.shape[0]) or (kp1.shape[0] < self.minimal_sample_size):
162                 raise ValueError(
163                     "kp1 and kp2 should be                                  equal shape at at least"
164                     f" [{self.minimal_sample_size}, 2],                                  got {kp1.shape}, {kp2.shape}"
165                 )
166         if self.model_type == "homography_from_linesegments":
167             KORNIA_CHECK_SHAPE(kp1, ["N", "2", "2"])
168             KORNIA_CHECK_SHAPE(kp2, ["N", "2", "2"])
169             if not (kp1.shape[0] == kp2.shape[0]) or (kp1.shape[0] < self.minimal_sample_size):
170                 raise ValueError(
171                     "kp1 and kp2 should be                                  equal shape at at least"
172                     f" [{self.minimal_sample_size}, 2, 2],                                  got {kp1.shape},"
173                     f" {kp2.shape}"
174                 )
175 
176     def forward(self, kp1: Tensor, kp2: Tensor, weights: Optional[Tensor] = None) -> Tuple[Tensor, Tensor]:
177         r"""Main forward method to execute the RANSAC algorithm.
178 
179         Args:
180             kp1: source image keypoints :math:`(N, 2)`.
181             kp2: distance image keypoints :math:`(N, 2)`.
182             weights: optional correspondences weights. Not used now.
183 
184         Returns:
185             - Estimated model, shape of :math:`(1, 3, 3)`.
186             - The inlier/outlier mask, shape of :math:`(1, N)`, where N is number of input correspondences.
187         """
188         self.validate_inputs(kp1, kp2, weights)
189         best_score_total: float = float(self.minimal_sample_size)
190         num_tc: int = len(kp1)
191         best_model_total = zeros(3, 3, dtype=kp1.dtype, device=kp1.device)
192         inliers_best_total: Tensor = zeros(num_tc, 1, device=kp1.device, dtype=torch.bool)
193         for i in range(self.max_iter):
194             # Sample minimal samples in batch to estimate models
195             idxs = self.sample(self.minimal_sample_size, num_tc, self.batch_size, kp1.device)
196             kp1_sampled = kp1[idxs]
197             kp2_sampled = kp2[idxs]
198 
199             kp1_sampled, kp2_sampled = self.remove_bad_samples(kp1_sampled, kp2_sampled)
200             if len(kp1_sampled) == 0:
201                 continue
202             # Estimate models
203             models = self.estimate_model_from_minsample(kp1_sampled, kp2_sampled)
204             models = self.remove_bad_models(models)
205             if (models is None) or (len(models) == 0):
206                 continue
207             # Score the models and select the best one
208             model, inliers, model_score = self.verify(kp1, kp2, models, self.inl_th)
209             # Store far-the-best model and (optionally) do a local optimization
210             if model_score > best_score_total:
211                 # Local optimization
212                 for lo_step in range(self.max_lo_iters):
213                     model_lo = self.polish_model(kp1, kp2, inliers)
214                     if (model_lo is None) or (len(model_lo) == 0):
215                         continue
216                     _, inliers_lo, score_lo = self.verify(kp1, kp2, model_lo, self.inl_th)
217                     # print (f"Orig score = {best_model_score}, LO score = {score_lo} TC={num_tc}")
218                     if score_lo > model_score:
219                         model = model_lo.clone()[0]
220                         inliers = inliers_lo.clone()
221                         model_score = score_lo
222                     else:
223                         break
224                 # Now storing the best model
225                 best_model_total = model.clone()
226                 inliers_best_total = inliers.clone()
227                 best_score_total = model_score
228 
229                 # Should we already stop?
230                 new_max_iter = int(
231                     self.max_samples_by_conf(int(best_score_total), num_tc, self.minimal_sample_size, self.confidence)
232                 )
233                 # print (f"New max_iter = {new_max_iter}")
234                 # Stop estimation, if the model is very good
235                 if (i + 1) * self.batch_size >= new_max_iter:
236                     break
237         # local optimization with all inliers for better precision
238         return best_model_total, inliers_best_total
239
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/kornia/geometry/ransac.py b/kornia/geometry/ransac.py
--- a/kornia/geometry/ransac.py
+++ b/kornia/geometry/ransac.py
@@ -103,7 +103,7 @@
             return 1.0
         if n_inl == num_tc:
             return 1.0
-        return math.log(1.0 - conf) / max(eps, math.log(max(eps, 1.0 - math.pow(n_inl / num_tc, sample_size))))
+        return math.log(1.0 - conf) / min(-eps, math.log(max(eps, 1.0 - math.pow(n_inl / num_tc, sample_size))))
 
     def estimate_model_from_minsample(self, kp1: Tensor, kp2: Tensor) -> Tensor:
         batch_size, sample_size = kp1.shape[:2]

{"golden_diff": "diff --git a/kornia/geometry/ransac.py b/kornia/geometry/ransac.py\n--- a/kornia/geometry/ransac.py\n+++ b/kornia/geometry/ransac.py\n@@ -103,7 +103,7 @@\n             return 1.0\n         if n_inl == num_tc:\n             return 1.0\n-        return math.log(1.0 - conf) / max(eps, math.log(max(eps, 1.0 - math.pow(n_inl / num_tc, sample_size))))\n+        return math.log(1.0 - conf) / min(-eps, math.log(max(eps, 1.0 - math.pow(n_inl / num_tc, sample_size))))\n \n     def estimate_model_from_minsample(self, kp1: Tensor, kp2: Tensor) -> Tensor:\n         batch_size, sample_size = kp1.shape[:2]\n", "issue": "RANSAC.max_samples_by_conf returns negative numbers\n### Describe the bug\n\nThis code fails\r\n\r\n`\r\nfrom kornia.geometry import RANSAC\r\n\r\nx = RANSAC.max_samples_by_conf(n_inl=100, num_tc=1000, sample_size=7, conf=0.99)\r\nassert x > 0.0\r\n`\r\n\n\n### Reproduction steps\n\n```bash\nfrom kornia.geometry import RANSAC\r\n\r\nx = RANSAC.max_samples_by_conf(n_inl=100, num_tc=1000, sample_size=7, conf=0.99)\r\nassert x > 0.0\n```\n\n\n### Expected behavior\n\nThe returned value should always be positive\n\n### Environment\n\n```shell\nEnvironment agnostic\n```\n\n\n### Additional context\n\na one-line bugfix will shortly follow\n", "before_files": [{"content": "\"\"\"Module containing RANSAC modules.\"\"\"\n\nimport math\nfrom functools import partial\nfrom typing import Callable, Optional, Tuple\n\nimport torch\n\nfrom kornia.core import Device, Module, Tensor, zeros\nfrom kornia.core.check import KORNIA_CHECK_SHAPE\nfrom kornia.geometry import (\n    find_fundamental,\n    find_homography_dlt,\n    find_homography_dlt_iterated,\n    find_homography_lines_dlt,\n    find_homography_lines_dlt_iterated,\n    symmetrical_epipolar_distance,\n)\nfrom kornia.geometry.homography import (\n    line_segment_transfer_error_one_way,\n    oneway_transfer_error,\n    sample_is_valid_for_homography,\n)\n\n__all__ = [\"RANSAC\"]\n\n\nclass RANSAC(Module):\n    \"\"\"Module for robust geometry estimation with RANSAC. https://en.wikipedia.org/wiki/Random_sample_consensus.\n\n    Args:\n        model_type: type of model to estimate: \"homography\", \"fundamental\", \"fundamental_7pt\",\n            \"homography_from_linesegments\".\n        inliers_threshold: threshold for the correspondence to be an inlier.\n        batch_size: number of generated samples at once.\n        max_iterations: maximum batches to generate. Actual number of models to try is ``batch_size * max_iterations``.\n        confidence: desired confidence of the result, used for the early stopping.\n        max_local_iterations: number of local optimization (polishing) iterations.\n    \"\"\"\n\n    def __init__(\n        self,\n        model_type: str = \"homography\",\n        inl_th: float = 2.0,\n        batch_size: int = 2048,\n        max_iter: int = 10,\n        confidence: float = 0.99,\n        max_lo_iters: int = 5,\n    ) -> None:\n        super().__init__()\n        self.supported_models = [\"homography\", \"fundamental\", \"fundamental_7pt\", \"homography_from_linesegments\"]\n        self.inl_th = inl_th\n        self.max_iter = max_iter\n        self.batch_size = batch_size\n        self.model_type = model_type\n        self.confidence = confidence\n        self.max_lo_iters = max_lo_iters\n        self.model_type = model_type\n\n        self.error_fn: Callable[..., Tensor]\n        self.minimal_solver: Callable[..., Tensor]\n        self.polisher_solver: Callable[..., Tensor]\n\n        if model_type == \"homography\":\n            self.error_fn = oneway_transfer_error\n            self.minimal_solver = find_homography_dlt\n            self.polisher_solver = find_homography_dlt_iterated\n            self.minimal_sample_size = 4\n        elif model_type == \"homography_from_linesegments\":\n            self.error_fn = line_segment_transfer_error_one_way\n            self.minimal_solver = find_homography_lines_dlt\n            self.polisher_solver = find_homography_lines_dlt_iterated\n            self.minimal_sample_size = 4\n        elif model_type == \"fundamental\":\n            self.error_fn = symmetrical_epipolar_distance\n            self.minimal_solver = find_fundamental\n            self.minimal_sample_size = 8\n            self.polisher_solver = find_fundamental\n        elif model_type == \"fundamental_7pt\":\n            self.error_fn = symmetrical_epipolar_distance\n            self.minimal_solver = partial(find_fundamental, method=\"7POINT\")\n            self.minimal_sample_size = 7\n            self.polisher_solver = find_fundamental\n        else:\n            raise NotImplementedError(f\"{model_type} is unknown. Try one of {self.supported_models}\")\n\n    def sample(self, sample_size: int, pop_size: int, batch_size: int, device: Device = torch.device(\"cpu\")) -> Tensor:\n        \"\"\"Minimal sampler, but unlike traditional RANSAC we sample in batches to get benefit of the parallel\n        processing, esp.\n\n        on GPU.\n        \"\"\"\n        rand = torch.rand(batch_size, pop_size, device=device)\n        _, out = rand.topk(k=sample_size, dim=1)\n        return out\n\n    @staticmethod\n    def max_samples_by_conf(n_inl: int, num_tc: int, sample_size: int, conf: float) -> float:\n        \"\"\"Formula to update max_iter in order to stop iterations earlier\n        https://en.wikipedia.org/wiki/Random_sample_consensus.\"\"\"\n        eps = 1e-9\n        if num_tc <= sample_size:\n            return 1.0\n        if n_inl == num_tc:\n            return 1.0\n        return math.log(1.0 - conf) / max(eps, math.log(max(eps, 1.0 - math.pow(n_inl / num_tc, sample_size))))\n\n    def estimate_model_from_minsample(self, kp1: Tensor, kp2: Tensor) -> Tensor:\n        batch_size, sample_size = kp1.shape[:2]\n        H = self.minimal_solver(kp1, kp2, torch.ones(batch_size, sample_size, dtype=kp1.dtype, device=kp1.device))\n        return H\n\n    def verify(self, kp1: Tensor, kp2: Tensor, models: Tensor, inl_th: float) -> Tuple[Tensor, Tensor, float]:\n        if len(kp1.shape) == 2:\n            kp1 = kp1[None]\n        if len(kp2.shape) == 2:\n            kp2 = kp2[None]\n        batch_size = models.shape[0]\n        if self.model_type == \"homography_from_linesegments\":\n            errors = self.error_fn(kp1.expand(batch_size, -1, 2, 2), kp2.expand(batch_size, -1, 2, 2), models)\n        else:\n            errors = self.error_fn(kp1.expand(batch_size, -1, 2), kp2.expand(batch_size, -1, 2), models)\n        inl = errors <= inl_th\n        models_score = inl.to(kp1).sum(dim=1)\n        best_model_idx = models_score.argmax()\n        best_model_score = models_score[best_model_idx].item()\n        model_best = models[best_model_idx].clone()\n        inliers_best = inl[best_model_idx]\n        return model_best, inliers_best, best_model_score\n\n    def remove_bad_samples(self, kp1: Tensor, kp2: Tensor) -> Tuple[Tensor, Tensor]:\n        \"\"\"\"\"\"\n        # ToDo: add (model-specific) verification of the samples,\n        # E.g. constraints on not to be a degenerate sample\n        if self.model_type == \"homography\":\n            mask = sample_is_valid_for_homography(kp1, kp2)\n            return kp1[mask], kp2[mask]\n        return kp1, kp2\n\n    def remove_bad_models(self, models: Tensor) -> Tensor:\n        # ToDo: add more and better degenerate model rejection\n        # For now it is simple and hardcoded\n        main_diagonal = torch.diagonal(models, dim1=1, dim2=2)\n        mask = main_diagonal.abs().min(dim=1)[0] > 1e-4\n        return models[mask]\n\n    def polish_model(self, kp1: Tensor, kp2: Tensor, inliers: Tensor) -> Tensor:\n        # TODO: Replace this with MAGSAC++ polisher\n        kp1_inl = kp1[inliers][None]\n        kp2_inl = kp2[inliers][None]\n        num_inl = kp1_inl.size(1)\n        model = self.polisher_solver(\n            kp1_inl, kp2_inl, torch.ones(1, num_inl, dtype=kp1_inl.dtype, device=kp1_inl.device)\n        )\n        return model\n\n    def validate_inputs(self, kp1: Tensor, kp2: Tensor, weights: Optional[Tensor] = None) -> None:\n        if self.model_type in [\"homography\", \"fundamental\"]:\n            KORNIA_CHECK_SHAPE(kp1, [\"N\", \"2\"])\n            KORNIA_CHECK_SHAPE(kp2, [\"N\", \"2\"])\n            if not (kp1.shape[0] == kp2.shape[0]) or (kp1.shape[0] < self.minimal_sample_size):\n                raise ValueError(\n                    \"kp1 and kp2 should be                                  equal shape at at least\"\n                    f\" [{self.minimal_sample_size}, 2],                                  got {kp1.shape}, {kp2.shape}\"\n                )\n        if self.model_type == \"homography_from_linesegments\":\n            KORNIA_CHECK_SHAPE(kp1, [\"N\", \"2\", \"2\"])\n            KORNIA_CHECK_SHAPE(kp2, [\"N\", \"2\", \"2\"])\n            if not (kp1.shape[0] == kp2.shape[0]) or (kp1.shape[0] < self.minimal_sample_size):\n                raise ValueError(\n                    \"kp1 and kp2 should be                                  equal shape at at least\"\n                    f\" [{self.minimal_sample_size}, 2, 2],                                  got {kp1.shape},\"\n                    f\" {kp2.shape}\"\n                )\n\n    def forward(self, kp1: Tensor, kp2: Tensor, weights: Optional[Tensor] = None) -> Tuple[Tensor, Tensor]:\n        r\"\"\"Main forward method to execute the RANSAC algorithm.\n\n        Args:\n            kp1: source image keypoints :math:`(N, 2)`.\n            kp2: distance image keypoints :math:`(N, 2)`.\n            weights: optional correspondences weights. Not used now.\n\n        Returns:\n            - Estimated model, shape of :math:`(1, 3, 3)`.\n            - The inlier/outlier mask, shape of :math:`(1, N)`, where N is number of input correspondences.\n        \"\"\"\n        self.validate_inputs(kp1, kp2, weights)\n        best_score_total: float = float(self.minimal_sample_size)\n        num_tc: int = len(kp1)\n        best_model_total = zeros(3, 3, dtype=kp1.dtype, device=kp1.device)\n        inliers_best_total: Tensor = zeros(num_tc, 1, device=kp1.device, dtype=torch.bool)\n        for i in range(self.max_iter):\n            # Sample minimal samples in batch to estimate models\n            idxs = self.sample(self.minimal_sample_size, num_tc, self.batch_size, kp1.device)\n            kp1_sampled = kp1[idxs]\n            kp2_sampled = kp2[idxs]\n\n            kp1_sampled, kp2_sampled = self.remove_bad_samples(kp1_sampled, kp2_sampled)\n            if len(kp1_sampled) == 0:\n                continue\n            # Estimate models\n            models = self.estimate_model_from_minsample(kp1_sampled, kp2_sampled)\n            models = self.remove_bad_models(models)\n            if (models is None) or (len(models) == 0):\n                continue\n            # Score the models and select the best one\n            model, inliers, model_score = self.verify(kp1, kp2, models, self.inl_th)\n            # Store far-the-best model and (optionally) do a local optimization\n            if model_score > best_score_total:\n                # Local optimization\n                for lo_step in range(self.max_lo_iters):\n                    model_lo = self.polish_model(kp1, kp2, inliers)\n                    if (model_lo is None) or (len(model_lo) == 0):\n                        continue\n                    _, inliers_lo, score_lo = self.verify(kp1, kp2, model_lo, self.inl_th)\n                    # print (f\"Orig score = {best_model_score}, LO score = {score_lo} TC={num_tc}\")\n                    if score_lo > model_score:\n                        model = model_lo.clone()[0]\n                        inliers = inliers_lo.clone()\n                        model_score = score_lo\n                    else:\n                        break\n                # Now storing the best model\n                best_model_total = model.clone()\n                inliers_best_total = inliers.clone()\n                best_score_total = model_score\n\n                # Should we already stop?\n                new_max_iter = int(\n                    self.max_samples_by_conf(int(best_score_total), num_tc, self.minimal_sample_size, self.confidence)\n                )\n                # print (f\"New max_iter = {new_max_iter}\")\n                # Stop estimation, if the model is very good\n                if (i + 1) * self.batch_size >= new_max_iter:\n                    break\n        # local optimization with all inliers for better precision\n        return best_model_total, inliers_best_total\n", "path": "kornia/geometry/ransac.py"}], "after_files": [{"content": "\"\"\"Module containing RANSAC modules.\"\"\"\n\nimport math\nfrom functools import partial\nfrom typing import Callable, Optional, Tuple\n\nimport torch\n\nfrom kornia.core import Device, Module, Tensor, zeros\nfrom kornia.core.check import KORNIA_CHECK_SHAPE\nfrom kornia.geometry import (\n    find_fundamental,\n    find_homography_dlt,\n    find_homography_dlt_iterated,\n    find_homography_lines_dlt,\n    find_homography_lines_dlt_iterated,\n    symmetrical_epipolar_distance,\n)\nfrom kornia.geometry.homography import (\n    line_segment_transfer_error_one_way,\n    oneway_transfer_error,\n    sample_is_valid_for_homography,\n)\n\n__all__ = [\"RANSAC\"]\n\n\nclass RANSAC(Module):\n    \"\"\"Module for robust geometry estimation with RANSAC. https://en.wikipedia.org/wiki/Random_sample_consensus.\n\n    Args:\n        model_type: type of model to estimate: \"homography\", \"fundamental\", \"fundamental_7pt\",\n            \"homography_from_linesegments\".\n        inliers_threshold: threshold for the correspondence to be an inlier.\n        batch_size: number of generated samples at once.\n        max_iterations: maximum batches to generate. Actual number of models to try is ``batch_size * max_iterations``.\n        confidence: desired confidence of the result, used for the early stopping.\n        max_local_iterations: number of local optimization (polishing) iterations.\n    \"\"\"\n\n    def __init__(\n        self,\n        model_type: str = \"homography\",\n        inl_th: float = 2.0,\n        batch_size: int = 2048,\n        max_iter: int = 10,\n        confidence: float = 0.99,\n        max_lo_iters: int = 5,\n    ) -> None:\n        super().__init__()\n        self.supported_models = [\"homography\", \"fundamental\", \"fundamental_7pt\", \"homography_from_linesegments\"]\n        self.inl_th = inl_th\n        self.max_iter = max_iter\n        self.batch_size = batch_size\n        self.model_type = model_type\n        self.confidence = confidence\n        self.max_lo_iters = max_lo_iters\n        self.model_type = model_type\n\n        self.error_fn: Callable[..., Tensor]\n        self.minimal_solver: Callable[..., Tensor]\n        self.polisher_solver: Callable[..., Tensor]\n\n        if model_type == \"homography\":\n            self.error_fn = oneway_transfer_error\n            self.minimal_solver = find_homography_dlt\n            self.polisher_solver = find_homography_dlt_iterated\n            self.minimal_sample_size = 4\n        elif model_type == \"homography_from_linesegments\":\n            self.error_fn = line_segment_transfer_error_one_way\n            self.minimal_solver = find_homography_lines_dlt\n            self.polisher_solver = find_homography_lines_dlt_iterated\n            self.minimal_sample_size = 4\n        elif model_type == \"fundamental\":\n            self.error_fn = symmetrical_epipolar_distance\n            self.minimal_solver = find_fundamental\n            self.minimal_sample_size = 8\n            self.polisher_solver = find_fundamental\n        elif model_type == \"fundamental_7pt\":\n            self.error_fn = symmetrical_epipolar_distance\n            self.minimal_solver = partial(find_fundamental, method=\"7POINT\")\n            self.minimal_sample_size = 7\n            self.polisher_solver = find_fundamental\n        else:\n            raise NotImplementedError(f\"{model_type} is unknown. Try one of {self.supported_models}\")\n\n    def sample(self, sample_size: int, pop_size: int, batch_size: int, device: Device = torch.device(\"cpu\")) -> Tensor:\n        \"\"\"Minimal sampler, but unlike traditional RANSAC we sample in batches to get benefit of the parallel\n        processing, esp.\n\n        on GPU.\n        \"\"\"\n        rand = torch.rand(batch_size, pop_size, device=device)\n        _, out = rand.topk(k=sample_size, dim=1)\n        return out\n\n    @staticmethod\n    def max_samples_by_conf(n_inl: int, num_tc: int, sample_size: int, conf: float) -> float:\n        \"\"\"Formula to update max_iter in order to stop iterations earlier\n        https://en.wikipedia.org/wiki/Random_sample_consensus.\"\"\"\n        eps = 1e-9\n        if num_tc <= sample_size:\n            return 1.0\n        if n_inl == num_tc:\n            return 1.0\n        return math.log(1.0 - conf) / min(-eps, math.log(max(eps, 1.0 - math.pow(n_inl / num_tc, sample_size))))\n\n    def estimate_model_from_minsample(self, kp1: Tensor, kp2: Tensor) -> Tensor:\n        batch_size, sample_size = kp1.shape[:2]\n        H = self.minimal_solver(kp1, kp2, torch.ones(batch_size, sample_size, dtype=kp1.dtype, device=kp1.device))\n        return H\n\n    def verify(self, kp1: Tensor, kp2: Tensor, models: Tensor, inl_th: float) -> Tuple[Tensor, Tensor, float]:\n        if len(kp1.shape) == 2:\n            kp1 = kp1[None]\n        if len(kp2.shape) == 2:\n            kp2 = kp2[None]\n        batch_size = models.shape[0]\n        if self.model_type == \"homography_from_linesegments\":\n            errors = self.error_fn(kp1.expand(batch_size, -1, 2, 2), kp2.expand(batch_size, -1, 2, 2), models)\n        else:\n            errors = self.error_fn(kp1.expand(batch_size, -1, 2), kp2.expand(batch_size, -1, 2), models)\n        inl = errors <= inl_th\n        models_score = inl.to(kp1).sum(dim=1)\n        best_model_idx = models_score.argmax()\n        best_model_score = models_score[best_model_idx].item()\n        model_best = models[best_model_idx].clone()\n        inliers_best = inl[best_model_idx]\n        return model_best, inliers_best, best_model_score\n\n    def remove_bad_samples(self, kp1: Tensor, kp2: Tensor) -> Tuple[Tensor, Tensor]:\n        \"\"\"\"\"\"\n        # ToDo: add (model-specific) verification of the samples,\n        # E.g. constraints on not to be a degenerate sample\n        if self.model_type == \"homography\":\n            mask = sample_is_valid_for_homography(kp1, kp2)\n            return kp1[mask], kp2[mask]\n        return kp1, kp2\n\n    def remove_bad_models(self, models: Tensor) -> Tensor:\n        # ToDo: add more and better degenerate model rejection\n        # For now it is simple and hardcoded\n        main_diagonal = torch.diagonal(models, dim1=1, dim2=2)\n        mask = main_diagonal.abs().min(dim=1)[0] > 1e-4\n        return models[mask]\n\n    def polish_model(self, kp1: Tensor, kp2: Tensor, inliers: Tensor) -> Tensor:\n        # TODO: Replace this with MAGSAC++ polisher\n        kp1_inl = kp1[inliers][None]\n        kp2_inl = kp2[inliers][None]\n        num_inl = kp1_inl.size(1)\n        model = self.polisher_solver(\n            kp1_inl, kp2_inl, torch.ones(1, num_inl, dtype=kp1_inl.dtype, device=kp1_inl.device)\n        )\n        return model\n\n    def validate_inputs(self, kp1: Tensor, kp2: Tensor, weights: Optional[Tensor] = None) -> None:\n        if self.model_type in [\"homography\", \"fundamental\"]:\n            KORNIA_CHECK_SHAPE(kp1, [\"N\", \"2\"])\n            KORNIA_CHECK_SHAPE(kp2, [\"N\", \"2\"])\n            if not (kp1.shape[0] == kp2.shape[0]) or (kp1.shape[0] < self.minimal_sample_size):\n                raise ValueError(\n                    \"kp1 and kp2 should be                                  equal shape at at least\"\n                    f\" [{self.minimal_sample_size}, 2],                                  got {kp1.shape}, {kp2.shape}\"\n                )\n        if self.model_type == \"homography_from_linesegments\":\n            KORNIA_CHECK_SHAPE(kp1, [\"N\", \"2\", \"2\"])\n            KORNIA_CHECK_SHAPE(kp2, [\"N\", \"2\", \"2\"])\n            if not (kp1.shape[0] == kp2.shape[0]) or (kp1.shape[0] < self.minimal_sample_size):\n                raise ValueError(\n                    \"kp1 and kp2 should be                                  equal shape at at least\"\n                    f\" [{self.minimal_sample_size}, 2, 2],                                  got {kp1.shape},\"\n                    f\" {kp2.shape}\"\n                )\n\n    def forward(self, kp1: Tensor, kp2: Tensor, weights: Optional[Tensor] = None) -> Tuple[Tensor, Tensor]:\n        r\"\"\"Main forward method to execute the RANSAC algorithm.\n\n        Args:\n            kp1: source image keypoints :math:`(N, 2)`.\n            kp2: distance image keypoints :math:`(N, 2)`.\n            weights: optional correspondences weights. Not used now.\n\n        Returns:\n            - Estimated model, shape of :math:`(1, 3, 3)`.\n            - The inlier/outlier mask, shape of :math:`(1, N)`, where N is number of input correspondences.\n        \"\"\"\n        self.validate_inputs(kp1, kp2, weights)\n        best_score_total: float = float(self.minimal_sample_size)\n        num_tc: int = len(kp1)\n        best_model_total = zeros(3, 3, dtype=kp1.dtype, device=kp1.device)\n        inliers_best_total: Tensor = zeros(num_tc, 1, device=kp1.device, dtype=torch.bool)\n        for i in range(self.max_iter):\n            # Sample minimal samples in batch to estimate models\n            idxs = self.sample(self.minimal_sample_size, num_tc, self.batch_size, kp1.device)\n            kp1_sampled = kp1[idxs]\n            kp2_sampled = kp2[idxs]\n\n            kp1_sampled, kp2_sampled = self.remove_bad_samples(kp1_sampled, kp2_sampled)\n            if len(kp1_sampled) == 0:\n                continue\n            # Estimate models\n            models = self.estimate_model_from_minsample(kp1_sampled, kp2_sampled)\n            models = self.remove_bad_models(models)\n            if (models is None) or (len(models) == 0):\n                continue\n            # Score the models and select the best one\n            model, inliers, model_score = self.verify(kp1, kp2, models, self.inl_th)\n            # Store far-the-best model and (optionally) do a local optimization\n            if model_score > best_score_total:\n                # Local optimization\n                for lo_step in range(self.max_lo_iters):\n                    model_lo = self.polish_model(kp1, kp2, inliers)\n                    if (model_lo is None) or (len(model_lo) == 0):\n                        continue\n                    _, inliers_lo, score_lo = self.verify(kp1, kp2, model_lo, self.inl_th)\n                    # print (f\"Orig score = {best_model_score}, LO score = {score_lo} TC={num_tc}\")\n                    if score_lo > model_score:\n                        model = model_lo.clone()[0]\n                        inliers = inliers_lo.clone()\n                        model_score = score_lo\n                    else:\n                        break\n                # Now storing the best model\n                best_model_total = model.clone()\n                inliers_best_total = inliers.clone()\n                best_score_total = model_score\n\n                # Should we already stop?\n                new_max_iter = int(\n                    self.max_samples_by_conf(int(best_score_total), num_tc, self.minimal_sample_size, self.confidence)\n                )\n                # print (f\"New max_iter = {new_max_iter}\")\n                # Stop estimation, if the model is very good\n                if (i + 1) * self.batch_size >= new_max_iter:\n                    break\n        # local optimization with all inliers for better precision\n        return best_model_total, inliers_best_total\n", "path": "kornia/geometry/ransac.py"}]}