Skip to content

pipeline

ChangeDetPipeline

Basic pipeline for running change detection algorithms.

Parameters:

Name Type Description Default
algo str

Change detection algorithm to be used

 required

Attributes:

Name Type Description
algo_name str

Name of change detection algorithm
algo_obj str

Change detection algorithm object
logger Logger

Logger object
Source code in changedet/pipeline.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class ChangeDetPipeline:
    """
    Basic pipeline for running change detection algorithms.

    Args:
        algo (str): Change detection algorithm to be used

    Attributes:
        algo_name (str): Name of change detection algorithm
        algo_obj (str): Change detection algorithm object
        logger (logging.Logger): Logger object

    """

    def __init__(self, algo: str):
        """Initialise Pipeline

        Args:
            algo (str): Change detection algorithm to be used
        """
        self.algo_name = algo
        self.algo_obj = AlgoCatalog.get(algo)
        self.logger = init_logger("changedet")

    # Image loading and sanity checks should be done here
    def read(self, im1: str, im2: str, band: int) -> tuple[np.ndarray, np.ndarray]:
        """Read and prepare images

        Args:
            im1 (str): Path to image 1
            im2 (str): Path to image 2
            band (int): Band selection

        Raises:
            AssertionError: If images are not in the same projection system
            AssertionError: If images are not of same shape

        Returns:
            tuple:
                - arr1 (numpy.ndarray): Image 1 array of shape (B, H, W)
                - arr2 (numpy.ndarray): Image 2 array of shape (B, H, W)
        """
        try:
            assert Path(im1).exists() and Path(im2).exists()
        except AssertionError:
            self.logger.critical("Images not found")
            raise

        arr1, crs1, self.meta1 = self._read(im1, band)
        arr2, crs2, self.meta2 = self._read(im2, band)

        try:
            assert crs1 == crs2
        except AssertionError:
            self.logger.critical("Images are not in the same projection system")
            raise

        try:
            assert arr1.shape == arr2.shape
        except AssertionError:
            self.logger.critical("Image array shapes do not match")
            raise

        return arr1, arr2

    def _read(self, im: str, band: int) -> tuple[np.ndarray, CRS, Profile]:
        with rio.open(im) as raster:
            profile = raster.profile
            crs = raster.crs

            if band == -1:
                arr = raster.read()
            else:
                arr = np.expand_dims(raster.read(band), axis=0)
        return arr, crs, profile

    def run(self, im1: str, im2: str, band: int = -1, **kwargs: Any) -> None:
        """
        Run change detection on images

        Args:
            im1 (str): Path to image 1
            im2 (str): Path to image 2
            band (int): Band selection

        Raises:
            AssertionError: If no algorithm is specified
        """
        if not self.algo_obj:
            raise AssertionError("Algorithm not specified")
        im1a, im2a = self.read(im1, im2, band)
        # TODO: Decide whether algos should have their own loggers
        kwargs.update({"logger": self.logger, "band": band})
        cmap = self.algo_obj.run(im1a, im2a, **kwargs)
        self.write(cmap)

    def write(self, cmap: np.ndarray) -> None:
        """Write change map to disk

        Args:
            cmap (numpy.ndarray): Change map of shape (B, H, W)

        """

        profile = self.meta1
        outfile = f"{self.algo_name}_cmap.tif"

        # Bandwise change or Single band change
        cmap = np.expand_dims(cmap, axis=0) if len(cmap.shape) == 2 else cmap

        profile["count"] = cmap.shape[0]

        with rio.Env():
            with rio.open(outfile, "w", **profile) as dst:
                for i in range(profile["count"]):
                    dst.write(cmap[i], i + 1)
        self.logger.info("Change map written to %s", outfile)

    @classmethod
    def list(cls) -> None:
        """List available algorithms"""
        print(AlgoCatalog.list())

__init__(algo)

Initialise Pipeline

Parameters:

Name Type Description Default
algo str

Change detection algorithm to be used

 required
Source code in changedet/pipeline.py 
27
28
29
30
31
32
33
34
35
def __init__(self, algo: str):
    """Initialise Pipeline

    Args:
        algo (str): Change detection algorithm to be used
    """
    self.algo_name = algo
    self.algo_obj = AlgoCatalog.get(algo)
    self.logger = init_logger("changedet")

list() classmethod

List available algorithms

Source code in changedet/pipeline.py 
131
132
133
134
@classmethod
def list(cls) -> None:
    """List available algorithms"""
    print(AlgoCatalog.list())

read(im1, im2, band)

Read and prepare images

Parameters:

Name Type Description Default
im1 str

Path to image 1

 required
im2 str

Path to image 2

 required
band int

Band selection

 required

Raises:

Type Description
AssertionError

If images are not in the same projection system
AssertionError

If images are not of same shape

Returns:

Name Type Description
tuple tuple[ndarray, ndarray]
  • arr1 (numpy.ndarray): Image 1 array of shape (B, H, W)
  • arr2 (numpy.ndarray): Image 2 array of shape (B, H, W)
Source code in changedet/pipeline.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def read(self, im1: str, im2: str, band: int) -> tuple[np.ndarray, np.ndarray]:
    """Read and prepare images

    Args:
        im1 (str): Path to image 1
        im2 (str): Path to image 2
        band (int): Band selection

    Raises:
        AssertionError: If images are not in the same projection system
        AssertionError: If images are not of same shape

    Returns:
        tuple:
            - arr1 (numpy.ndarray): Image 1 array of shape (B, H, W)
            - arr2 (numpy.ndarray): Image 2 array of shape (B, H, W)
    """
    try:
        assert Path(im1).exists() and Path(im2).exists()
    except AssertionError:
        self.logger.critical("Images not found")
        raise

    arr1, crs1, self.meta1 = self._read(im1, band)
    arr2, crs2, self.meta2 = self._read(im2, band)

    try:
        assert crs1 == crs2
    except AssertionError:
        self.logger.critical("Images are not in the same projection system")
        raise

    try:
        assert arr1.shape == arr2.shape
    except AssertionError:
        self.logger.critical("Image array shapes do not match")
        raise

    return arr1, arr2

run(im1, im2, band=-1, **kwargs)

Run change detection on images

Parameters:

Name Type Description Default
im1 str

Path to image 1

 required
im2 str

Path to image 2

 required
band int

Band selection

 -1

Raises:

Type Description
AssertionError

If no algorithm is specified
Source code in changedet/pipeline.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def run(self, im1: str, im2: str, band: int = -1, **kwargs: Any) -> None:
    """
    Run change detection on images

    Args:
        im1 (str): Path to image 1
        im2 (str): Path to image 2
        band (int): Band selection

    Raises:
        AssertionError: If no algorithm is specified
    """
    if not self.algo_obj:
        raise AssertionError("Algorithm not specified")
    im1a, im2a = self.read(im1, im2, band)
    # TODO: Decide whether algos should have their own loggers
    kwargs.update({"logger": self.logger, "band": band})
    cmap = self.algo_obj.run(im1a, im2a, **kwargs)
    self.write(cmap)

write(cmap)

Write change map to disk

Parameters:

Name Type Description Default
cmap ndarray

Change map of shape (B, H, W)

 required
Source code in changedet/pipeline.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def write(self, cmap: np.ndarray) -> None:
    """Write change map to disk

    Args:
        cmap (numpy.ndarray): Change map of shape (B, H, W)

    """

    profile = self.meta1
    outfile = f"{self.algo_name}_cmap.tif"

    # Bandwise change or Single band change
    cmap = np.expand_dims(cmap, axis=0) if len(cmap.shape) == 2 else cmap

    profile["count"] = cmap.shape[0]

    with rio.Env():
        with rio.open(outfile, "w", **profile) as dst:
            for i in range(profile["count"]):
                dst.write(cmap[i], i + 1)
    self.logger.info("Change map written to %s", outfile)