|
| 1 | +from os import PathLike |
| 2 | +from pathlib import Path |
| 3 | + |
| 4 | +import cv2 |
| 5 | +import numpy as np |
| 6 | +from rdflib import XSD |
| 7 | + |
| 8 | +from tamper.vocabularies import TAMPER |
| 9 | + |
| 10 | +from tamper.core import ImageAsset, Operation, MappedProperty |
| 11 | + |
| 12 | + |
| 13 | +class AddGaussianNoise(Operation): |
| 14 | + __rdf_type__ = TAMPER.AddGaussianNoise |
| 15 | + |
| 16 | + mean: MappedProperty[float] = MappedProperty(TAMPER.gaussianMean, XSD.double) |
| 17 | + std: MappedProperty[float] = MappedProperty(TAMPER.gaussianStd, XSD.double) |
| 18 | + seed: MappedProperty[int] = MappedProperty(TAMPER.noiseSeed, XSD.integer) |
| 19 | + |
| 20 | + def mutate(self, out_dir: PathLike[str] | None = None): |
| 21 | + used = self.get_used() |
| 22 | + if len(used) != 1: |
| 23 | + raise ValueError("Operation requires exactly one image asset") |
| 24 | + |
| 25 | + img_asset = ImageAsset(self.graph, used[0]) |
| 26 | + |
| 27 | + img = cv2.imread(img_asset.file_path) |
| 28 | + rng = np.random.default_rng(self.seed) |
| 29 | + noise = rng.normal(self.mean, self.std, img.shape) |
| 30 | + noisy_img = np.clip(img + noise, 0, 255).astype(np.uint8) |
| 31 | + ext = Path(img_asset.file_path).suffix or ".png" |
| 32 | + ok, buf = cv2.imencode(ext, noisy_img) |
| 33 | + if not ok: |
| 34 | + raise RuntimeError(f"Encoding to {ext} failed") |
| 35 | + |
| 36 | + with self._generates_file(dir=out_dir, suffix=ext) as f: |
| 37 | + Path(f).write_bytes(buf.tobytes()) |
| 38 | + |
| 39 | + |
| 40 | +class AddSaltPepperNoise(Operation): |
| 41 | + __rdf_type__ = TAMPER.AddSaltPepperNoise |
| 42 | + |
| 43 | + amount: MappedProperty[float] = MappedProperty(TAMPER.saltPepperAmount, XSD.double) |
| 44 | + salt_ratio: MappedProperty[float] = MappedProperty( |
| 45 | + TAMPER.saltPepperRatio, XSD.double |
| 46 | + ) |
| 47 | + seed: MappedProperty[int] = MappedProperty(TAMPER.noiseSeed, XSD.integer) |
| 48 | + |
| 49 | + def mutate(self, out_dir: PathLike[str] | None = None): |
| 50 | + used = self.get_used() |
| 51 | + if len(used) != 1: |
| 52 | + raise ValueError("Operation requires exactly one image asset") |
| 53 | + |
| 54 | + img_asset = ImageAsset(self.graph, used[0]) |
| 55 | + |
| 56 | + img = cv2.imread(img_asset.file_path) |
| 57 | + if img is None: |
| 58 | + raise RuntimeError(f"Could not read image: {img_asset.file_path}") |
| 59 | + |
| 60 | + rng = np.random.default_rng(self.seed) |
| 61 | + out = img.copy() |
| 62 | + h, w = img.shape[:2] |
| 63 | + n = int(self.amount * h * w) |
| 64 | + n_salt = int(n * self.salt_ratio) |
| 65 | + |
| 66 | + flat = rng.choice(h * w, size=n, replace=False) |
| 67 | + ys, xs = np.unravel_index(flat, (h, w)) |
| 68 | + out[ys[:n_salt], xs[:n_salt]] = 255 |
| 69 | + out[ys[n_salt:], xs[n_salt:]] = 0 |
| 70 | + |
| 71 | + ext = Path(img_asset.file_path).suffix or ".png" |
| 72 | + ok, buf = cv2.imencode(ext, out) |
| 73 | + if not ok: |
| 74 | + raise RuntimeError(f"Encoding to {ext} failed") |
| 75 | + |
| 76 | + with self._generates_file(dir=out_dir, suffix=ext) as f: |
| 77 | + Path(f).write_bytes(buf.tobytes()) |
0 commit comments