def process()

in granule_ingester/granule_ingester/processors/ElevationOffset.py [0:0]

• 33 lines of code
• 9 McCabe index (conditional complexity)

    def process(self, tile, dataset):
        slice_dims = {}

        tile_type = tile.tile.WhichOneof("tile_type")
        tile_data = getattr(tile.tile, tile_type)

        tile_summary = tile.summary

        spec_list = tile_summary.section_spec.split(',')

        height_index = None

        for spec in spec_list:
            v = spec.split(':')

            if v[0] == self.offset_dimension:
                height_index = int(v[1])
            elif v[0] in dataset[self.base_dimension].dims:
                slice_dims[v[0]] = slice(int(v[1]), int(v[2]))

        if height_index is None:
            logger.warning(f"Cannot compute heights for tile {str(tile.summary.tile_id)}. Unable to determine height index from spec")

            return tile

        height = dataset[self.offset_dimension][height_index].item()
        base_height = dataset[self.base_dimension].isel(slice_dims).data

        try:
            flip_lat, lat_axis = dataset.attrs.pop('_FlippedLat')
        except KeyError:
            flip_lat, lat_axis = (False, None)

        if flip_lat:
            base_height = np.flip(base_height, axis=lat_axis)

        computed_height = base_height + height

        # if tile_type in ['GridTile', 'GridMultiVariableTile']:
        #     elev_shape = (len(from_shaped_array(tile_data.latitude)), len(from_shaped_array(tile_data.longitude)))
        # else:
        #     elev_shape = from_shaped_array(tile_data.latitude).shape

        if self.flip_lat:
            computed_height = np.flip(computed_height, axis=0)

        tile_data.elevation.CopyFrom(
            to_shaped_array(computed_height)
        )

        tile_data.max_elevation = np.nanmax(computed_height).item()
        tile_data.min_elevation = np.nanmin(computed_height).item()

        return tile