class TestMetaGridLevelMetaTilesGeodetic(object):
    def __init__(self):
        self.meta_grid = MetaGrid(TileGrid(is_geodetic=True), meta_size=(2, 2))

    def test_full_grid_2(self):
        bbox = (-180.0, -90.0, 180.0, 90)
        abbox, tile_grid, meta_tiles = \
            self.meta_grid.get_affected_level_tiles(bbox, 2)
        meta_tiles = list(meta_tiles)
        assert_almost_equal_bbox(bbox, abbox)

        eq_(tile_grid, (2, 1))
        eq_(len(meta_tiles), 2)
        eq_(meta_tiles[0], (0, 0, 2))
        eq_(meta_tiles[1], (2, 0, 2))

    def test_partial_grid_3(self):
        bbox = (0.0, 5.0, 45, 40)
        abbox, tile_grid, meta_tiles = \
            self.meta_grid.get_affected_level_tiles(bbox, 3)
        meta_tiles = list(meta_tiles)
        assert_almost_equal_bbox((0.0, 0.0, 90.0, 90.0), abbox)

        eq_(tile_grid, (1, 1))
        eq_(len(meta_tiles), 1)
        eq_(meta_tiles[0], (4, 2, 3))

class TestMetaGridLevelMetaTiles(object):
    def __init__(self):
        self.meta_grid = MetaGrid(TileGrid(), meta_size=(2, 2))

    def test_full_grid_0(self):
        bbox = (-20037508.34, -20037508.34, 20037508.34, 20037508.34)
        abbox, tile_grid, meta_tiles = \
            self.meta_grid.get_affected_level_tiles(bbox, 0)
        meta_tiles = list(meta_tiles)
        assert_almost_equal_bbox(bbox, abbox)

        eq_(len(meta_tiles), 1)
        eq_(meta_tiles[0], (0, 0, 0))

    def test_full_grid_2(self):
        bbox = (-20037508.34, -20037508.34, 20037508.34, 20037508.34)
        abbox, tile_grid, meta_tiles = \
            self.meta_grid.get_affected_level_tiles(bbox, 2)
        meta_tiles = list(meta_tiles)
        assert_almost_equal_bbox(bbox, abbox)

        eq_(tile_grid, (2, 2))
        eq_(len(meta_tiles), 4)
        eq_(meta_tiles[0], (0, 2, 2))
        eq_(meta_tiles[1], (2, 2, 2))
        eq_(meta_tiles[2], (0, 0, 2))
        eq_(meta_tiles[3], (2, 0, 2))

def test_metagrid_tiles_w_meta_size():
    mgrid = MetaGrid(grid=TileGrid(), meta_size=(4, 2))
    assert list(mgrid.meta_tile((1, 2, 2)).tile_patterns) == \
        [((0, 3, 2), (0, 0)), ((1, 3, 2), (256, 0)),
         ((2, 3, 2), (512, 0)), ((3, 3, 2), (768, 0)),
         ((0, 2, 2), (0, 256)), ((1, 2, 2), (256, 256)),
         ((2, 2, 2), (512, 256)), ((3, 2, 2), (768, 256))]

    def __init__(self, grid, cache, sources, format, locker, image_opts=None, request_format=None,
        meta_buffer=None, meta_size=None, minimize_meta_requests=False, identifier=None,
        pre_store_filter=None, concurrent_tile_creators=1, tile_creator_class=None,
        bulk_meta_tiles=False,
        ):
        self.grid = grid
        self.cache = cache
        self.locker = locker
        self.identifier = identifier
        self.meta_grid = None
        self.format = format
        self.image_opts = image_opts
        self.request_format = request_format or format
        self.sources = sources
        self.minimize_meta_requests = minimize_meta_requests
        self._expire_timestamp = None
        self.pre_store_filter = pre_store_filter or []
        self.concurrent_tile_creators = concurrent_tile_creators
        self.tile_creator_class = tile_creator_class or TileCreator

        if meta_buffer or (meta_size and not meta_size == [1, 1]):
            if all(source.supports_meta_tiles for source in sources):
                self.meta_grid = MetaGrid(grid, meta_size=meta_size, meta_buffer=meta_buffer)
            elif any(source.supports_meta_tiles for source in sources):
                raise ValueError('meta tiling configured but not supported by all sources')
            elif meta_size and not meta_size == [1, 1] and bulk_meta_tiles:
                # meta tiles configured but all sources are tiled
                # use bulk_meta_tile mode that download tiles in parallel
                self.meta_grid = MetaGrid(grid, meta_size=meta_size, meta_buffer=0)
                self.tile_creator_class = partial(self.tile_creator_class, bulk_meta_tiles=True)

 def test_metatile_bbox(self):
     mgrid = MetaGrid(grid=TileGrid(), meta_size=(2, 2))
     meta_tile = mgrid.meta_tile((0, 0, 2))
     assert meta_tile.bbox == (-20037508.342789244, -20037508.342789244, 0.0, 0.0)
     meta_tile = mgrid.meta_tile((1, 1, 2))
     assert meta_tile.bbox == (-20037508.342789244, -20037508.342789244, 0.0, 0.0)
     meta_tile = mgrid.meta_tile((4, 5, 3))
     assert_almost_equal_bbox(meta_tile.bbox, (0.0, 0.0, 10018754.171394622, 10018754.171394622))

 def test_metatile_non_default_meta_size(self):
     mgrid = MetaGrid(grid=self.grid, meta_size=(4, 2), meta_buffer=0)
     meta_tile = mgrid.meta_tile((4, 3, 6))
     eq_(meta_tile.bbox, (0.0, 0.0, 180.0, 90.0))
     eq_(meta_tile.size, (1024, 512))
     eq_(meta_tile.grid_size, (4, 2))
     eq_(meta_tile.tile_patterns, [((4, 3, 6), (0, 0)), ((5, 3, 6), (256, 0)),
         ((6, 3, 6), (512, 0)), ((7, 3, 6), (768, 0)), ((4, 2, 6), (0, 256)),
         ((5, 2, 6), (256, 256)), ((6, 2, 6), (512, 256)), ((7, 2, 6), (768, 256))])

def test_metagrid_tiles():
    mgrid = MetaGrid(grid=TileGrid(), meta_size=(2, 2))
    assert list(mgrid.meta_tile((0, 0, 0)).tile_patterns) == \
        [((0, 0, 0), (0, 0))]
    assert list(mgrid.meta_tile((0, 1, 1)).tile_patterns) == \
        [((0, 1, 1), (0, 0)), ((1, 1, 1), (256, 0)),
         ((0, 0, 1), (0, 256)), ((1, 0, 1), (256, 256))]

    assert list(mgrid.meta_tile((1, 2, 2)).tile_patterns) == \
        [((0, 3, 2), (0, 0)), ((1, 3, 2), (256, 0)),
         ((0, 2, 2), (0, 256)), ((1, 2, 2), (256, 256))]

class TestMinimalMetaTile(object):
    def setup(self):
        self.mgrid = MetaGrid(grid=tile_grid('EPSG:4326'), meta_size=(2, 2), meta_buffer=10)

    def test_minimal_tiles(self):
        sgrid = self.mgrid.minimal_meta_tile([(0, 0, 2), (1, 0, 2)])
        eq_(sgrid.grid_size, (2, 1))
        eq_(list(sgrid.tile_patterns),
            [
                ((0, 0, 2), (0, 10)),
                ((1, 0, 2), (256, 10)),
            ]
        )
        eq_(sgrid.bbox, (-180.0, -90.0, 3.515625, 3.515625))

    def test_minimal_tiles_fragmented(self):
        sgrid = self.mgrid.minimal_meta_tile(
            [
                           (2, 3, 3),
                (1, 2, 3),
                           (2, 1, 3),
            ])

        eq_(sgrid.grid_size, (2, 3))
        eq_(list(sgrid.tile_patterns),
            [
                ((1, 3, 3), (10, 0)), ((2, 3, 3), (266, 0)),
                ((1, 2, 3), (10, 256)), ((2, 2, 3), (266, 256)),
                ((1, 1, 3), (10, 512)), ((2, 1, 3), (266, 512)),
            ]
        )
        eq_(sgrid.bbox, (-136.7578125, -46.7578125, -43.2421875, 90.0))

    def test_minimal_tiles_fragmented_ul(self):
        self.mgrid = MetaGrid(grid=tile_grid('EPSG:4326', origin='ul'),
            meta_size=(2, 2), meta_buffer=10)
        sgrid = self.mgrid.minimal_meta_tile(
            [
                           (2, 0, 3),
                (1, 1, 3),
                           (2, 2, 3),
            ])

        eq_(sgrid.grid_size, (2, 3))
        eq_(list(sgrid.tile_patterns),
            [
                ((1, 0, 3), (10, 0)), ((2, 0, 3), (266, 0)),
                ((1, 1, 3), (10, 256)), ((2, 1, 3), (266, 256)),
                ((1, 2, 3), (10, 512)), ((2, 2, 3), (266, 512)),
            ]
        )
        eq_(sgrid.bbox, (-136.7578125, -46.7578125, -43.2421875, 90.0))

class TestMetaTileSQRT2(object):
    def setup(self):
        self.grid = tile_grid('EPSG:4326', res_factor='sqrt2')
        self.mgrid = MetaGrid(grid=self.grid, meta_size=(4, 4), meta_buffer=10)
    def test_meta_tile(self):
        meta_tile = self.mgrid.meta_tile((0, 0, 8))
        eq_(meta_tile.size, (1034, 1034))

    def test_metatile_bbox(self):
        meta_tile = self.mgrid.meta_tile((0, 0, 2))
        eq_(meta_tile.bbox,  (-180, -90, 180, 90))
        eq_(meta_tile.size,  (512, 256))
        eq_(meta_tile.grid_size,  (2, 1))
        eq_(meta_tile.tile_patterns, [((0, 0, 2), (0, 0)), ((1, 0, 2), (256, 0))])

        meta_tile = self.mgrid.meta_tile((1, 0, 2))
        eq_(meta_tile.bbox, (-180.0, -90, 180.0, 90.0))
        eq_(meta_tile.size,  (512, 256))
        eq_(meta_tile.grid_size,  (2, 1))

        meta_tile = self.mgrid.meta_tile((0, 0, 3))
        eq_(meta_tile.bbox, (-180.0, -90, 180.0, 90.0))
        eq_(meta_tile.size,  (724, 362))
        eq_(meta_tile.tile_patterns, [((0, 1, 3), (0, -149)), ((1, 1, 3), (256, -149)),
            ((2, 1, 3), (512, -149)), ((0, 0, 3), (0, 107)), ((1, 0, 3), (256, 107)),
            ((2, 0, 3), (512, 107))])

    def test_metatile_non_default_meta_size(self):
        mgrid = MetaGrid(grid=self.grid, meta_size=(4, 2), meta_buffer=0)
        meta_tile = mgrid.meta_tile((4, 3, 6))
        eq_(meta_tile.bbox, (0.0, 0.0, 180.0, 90.0))
        eq_(meta_tile.size, (1024, 512))
        eq_(meta_tile.grid_size, (4, 2))
        eq_(meta_tile.tile_patterns, [((4, 3, 6), (0, 0)), ((5, 3, 6), (256, 0)),
            ((6, 3, 6), (512, 0)), ((7, 3, 6), (768, 0)), ((4, 2, 6), (0, 256)),
            ((5, 2, 6), (256, 256)), ((6, 2, 6), (512, 256)), ((7, 2, 6), (768, 256))])

    def test_minimal_tiles_fragmented_ul(self):
        self.mgrid = MetaGrid(grid=tile_grid('EPSG:4326', origin='ul'),
            meta_size=(2, 2), meta_buffer=10)
        sgrid = self.mgrid.minimal_meta_tile(
            [
                           (2, 0, 3),
                (1, 1, 3),
                           (2, 2, 3),
            ])

        eq_(sgrid.grid_size, (2, 3))
        eq_(list(sgrid.tile_patterns),
            [
                ((1, 0, 3), (10, 0)), ((2, 0, 3), (266, 0)),
                ((1, 1, 3), (10, 256)), ((2, 1, 3), (266, 256)),
                ((1, 2, 3), (10, 512)), ((2, 2, 3), (266, 512)),
            ]
        )
        eq_(sgrid.bbox, (-136.7578125, -46.7578125, -43.2421875, 90.0))

    def __init__(self, task, worker_pool, handle_stale=False, handle_uncached=False,
                 work_on_metatiles=True, skip_geoms_for_last_levels=0, progress_logger=None,
                 seed_progress=None):
        self.tile_mgr = task.tile_manager
        self.task = task
        self.worker_pool = worker_pool
        self.handle_stale = handle_stale
        self.handle_uncached = handle_uncached
        self.work_on_metatiles = work_on_metatiles
        self.skip_geoms_for_last_levels = skip_geoms_for_last_levels
        self.progress_logger = progress_logger

        num_seed_levels = len(task.levels)
        self.report_till_level = task.levels[int(num_seed_levels * 0.8)]
        meta_size = self.tile_mgr.meta_grid.meta_size if self.tile_mgr.meta_grid else (1, 1)
        self.tiles_per_metatile = meta_size[0] * meta_size[1]
        self.grid = MetaGrid(self.tile_mgr.grid, meta_size=meta_size, meta_buffer=0)
        self.count = 0
        self.seed_progress = seed_progress or SeedProgress()

    def __init__(self, grid, cache, sources, format, image_opts=None, request_format=None,
        meta_buffer=None, meta_size=None, minimize_meta_requests=False,
        pre_store_filter=None, concurrent_tile_creators=1):
        self.grid = grid
        self.cache = cache
        self.meta_grid = None
        self.format = format
        self.image_opts = image_opts
        self.request_format = request_format or format
        self.sources = sources
        self.minimize_meta_requests = minimize_meta_requests
        self._expire_timestamp = None
        self.transparent = self.sources[0].transparent
        self.pre_store_filter = pre_store_filter or []
        self.concurrent_tile_creators = concurrent_tile_creators

        if meta_buffer or (meta_size and not meta_size == [1, 1]):
            if all(source.supports_meta_tiles for source in sources):
                self.meta_grid = MetaGrid(grid, meta_size=meta_size, meta_buffer=meta_buffer)
            elif any(source.supports_meta_tiles for source in sources):
                raise ValueError('meta tiling configured but not supported by all sources')

    def __init__(self, task, worker_pool, handle_stale=False, handle_uncached=False,
                 work_on_metatiles=True, skip_geoms_for_last_levels=0, progress_logger=None,
                 seed_progress=None):
        self.tile_mgr = task.tile_manager
        self.task = task
        self.worker_pool = worker_pool
        self.handle_stale = handle_stale
        self.handle_uncached = handle_uncached
        self.work_on_metatiles = work_on_metatiles
        self.skip_geoms_for_last_levels = skip_geoms_for_last_levels
        self.progress_logger = progress_logger

        num_seed_levels = len(task.levels)
        if num_seed_levels >= 4:
            self.report_till_level = task.levels[num_seed_levels-2]
        else:
            self.report_till_level = task.levels[num_seed_levels-1]
        meta_size = self.tile_mgr.meta_grid.meta_size if self.tile_mgr.meta_grid else (1, 1)
        self.tiles_per_metatile = meta_size[0] * meta_size[1]
        self.grid = MetaGrid(self.tile_mgr.grid, meta_size=meta_size, meta_buffer=0)
        self.count = 0
        self.seed_progress = seed_progress or SeedProgress()

        # It is possible that we 'walk' through the same tile multiple times
        # when seeding irregular tile grids[0]. limit_sub_bbox prevents that we
        # recurse into the same area multiple times, but it is still possible
        # that a tile is processed multiple times. Locking prevents that a tile
        # is seeded multiple times, but it is possible that we count the same tile
        # multiple times (in dry-mode, or while the tile is in the process queue).

        # Tile counts can be off by 280% with sqrt2 grids.
        # We keep a small cache of already processed tiles to skip most duplicates.
        # A simple cache of 64 tile coordinates for each level already brings the
        # difference down to ~8%, which is good enough and faster than a more
        # sophisticated FIFO cache with O(1) lookup, or even caching all tiles.

        # [0] irregular tile grids: where one tile does not have exactly 4 subtiles
        # Typically when you use res_factor, or a custom res list.
        self.seeded_tiles = {l: deque(maxlen=64) for l in task.levels}

class TestMetaTile(object):
    def setup(self):
        self.mgrid = MetaGrid(grid=tile_grid('EPSG:4326'), meta_size=(2, 2), meta_buffer=10)
    def test_meta_tile(self):
        meta_tile = self.mgrid.meta_tile((2, 0, 2))
        eq_(meta_tile.size, (522, 512))

    def test_metatile_bbox(self):
        mgrid = MetaGrid(grid=TileGrid(), meta_size=(2, 2))
        meta_tile = mgrid.meta_tile((0, 0, 2))
        assert meta_tile.bbox == (-20037508.342789244, -20037508.342789244, 0.0, 0.0)
        meta_tile = mgrid.meta_tile((1, 1, 2))
        assert meta_tile.bbox == (-20037508.342789244, -20037508.342789244, 0.0, 0.0)
        meta_tile = mgrid.meta_tile((4, 5, 3))
        assert_almost_equal_bbox(meta_tile.bbox, (0.0, 0.0, 10018754.171394622, 10018754.171394622))

    def test_metatile_non_default_meta_size(self):
        mgrid = MetaGrid(grid=TileGrid(), meta_size=(4, 2))
        meta_tile = mgrid.meta_tile((4, 5, 3))
        assert_almost_equal_bbox(meta_tile.bbox, (0.0, 0.0, 20037508.342789244, 10018754.171394622))
        eq_(meta_tile.size, (1024, 512))
        eq_(meta_tile.grid_size, (4, 2))

class TileWalker(object):
    def __init__(self, task, worker_pool, handle_stale=False, handle_uncached=False,
                 work_on_metatiles=True, skip_geoms_for_last_levels=0, progress_logger=None,
                 seed_progress=None):
        self.tile_mgr = task.tile_manager
        self.task = task
        self.worker_pool = worker_pool
        self.handle_stale = handle_stale
        self.handle_uncached = handle_uncached
        self.work_on_metatiles = work_on_metatiles
        self.skip_geoms_for_last_levels = skip_geoms_for_last_levels
        self.progress_logger = progress_logger

        num_seed_levels = len(task.levels)
        self.report_till_level = task.levels[int(num_seed_levels * 0.8)]
        meta_size = self.tile_mgr.meta_grid.meta_size if self.tile_mgr.meta_grid else (1, 1)
        self.tiles_per_metatile = meta_size[0] * meta_size[1]
        self.grid = MetaGrid(self.tile_mgr.grid, meta_size=meta_size, meta_buffer=0)
        self.count = 0
        self.seed_progress = seed_progress or SeedProgress()

    def walk(self):
        assert self.handle_stale or self.handle_uncached
        bbox = self.task.coverage.extent.bbox_for(self.tile_mgr.grid.srs)
        if self.seed_progress.already_processed():
            # nothing to seed
            self.seed_progress.step_forward()
        else:
            try:
                self._walk(bbox, self.task.levels)
            except StopProcess:
                pass
        self.report_progress(self.task.levels[0], self.task.coverage.bbox)

    def _walk(self, cur_bbox, levels, all_subtiles=False):
        """
        :param cur_bbox: the bbox to seed in this call
        :param levels: list of levels to seed
        :param all_subtiles: seed all subtiles and do not check for
                             intersections with bbox/geom
        """
        current_level, levels = levels[0], levels[1:]
        bbox_, tiles, subtiles = self.grid.get_affected_level_tiles(cur_bbox, current_level)
        total_subtiles = tiles[0] * tiles[1]

        if len(levels) < self.skip_geoms_for_last_levels:
            # do not filter in last levels
            all_subtiles = True
        subtiles = self._filter_subtiles(subtiles, all_subtiles)

        if current_level <= self.report_till_level:
            self.report_progress(current_level, cur_bbox)

        if not self.seed_progress.running():
            self.report_progress(current_level, cur_bbox)
            self.tile_mgr.cleanup()
            raise StopProcess()

        for i, (subtile, sub_bbox, intersection) in enumerate(subtiles):
            if subtile is None: # no intersection
                self.seed_progress.step_forward(total_subtiles)
                continue
            if levels: # recurse to next level
                sub_bbox = limit_sub_bbox(cur_bbox, sub_bbox)
                if intersection == CONTAINS:
                    all_subtiles = True
                else:
                    all_subtiles = False

                with self.seed_progress.step_down(i, total_subtiles):
                    if self.seed_progress.already_processed():
                        self.seed_progress.step_forward()
                    else:
                        self._walk(sub_bbox, levels,
                            all_subtiles=all_subtiles)

            if not self.work_on_metatiles:
                # collect actual tiles
                handle_tiles = self.grid.tile_list(subtile)
            else:
                handle_tiles = [subtile]

            if self.handle_uncached:
                handle_tiles = [t for t in handle_tiles if
                                    t is not None and
                                    not self.tile_mgr.is_cached(t)]
            elif self.handle_stale:
                handle_tiles = [t for t in handle_tiles if
                                    t is not None and
                                    self.tile_mgr.is_stale(t)]
            if handle_tiles:
                self.count += 1
                self.worker_pool.process(handle_tiles, self.seed_progress)

            if not levels:
                self.seed_progress.step_forward(total_subtiles)

        if len(levels) >= 4:
            # call cleanup to close open caches
            # for connection based caches
#.........这里部分代码省略.........

 def __init__(self):
     self.meta_grid = MetaGrid(TileGrid(is_geodetic=True), meta_size=(2, 2))

class TestMetaGridGeodetic(object):
    def setup(self):
        self.mgrid = MetaGrid(grid=tile_grid('EPSG:4326'), meta_size=(2, 2), meta_buffer=10)

    def test_meta_bbox_level_0(self):
        eq_(self.mgrid._meta_bbox((0, 0, 0)), ((-180, -90, 180, 90), (0, 0, 0, -128)))
        eq_(self.mgrid._meta_bbox((0, 0, 0), limit_to_bbox=False),
            ((-194.0625, -104.0625, 194.0625, 284.0625), (10, 10, 10, 10)))

        eq_(self.mgrid.meta_tile((0, 0, 0)).size, (256, 128))

    def test_tiles_level_0(self):
        meta_tile = self.mgrid.meta_tile((0, 0, 0))
        eq_(meta_tile.size, (256, 128))
        eq_(meta_tile.grid_size, (1, 1))
        eq_(meta_tile.tile_patterns, [((0, 0, 0), (0, -128))])

    def test_meta_bbox_level_1(self):
        eq_(self.mgrid._meta_bbox((0, 0, 1)), ((-180, -90, 180, 90), (0, 0, 0, 0)))
        eq_(self.mgrid._meta_bbox((0, 0, 1), limit_to_bbox=False),
            ((-187.03125, -97.03125, 187.03125, 97.03125), (10, 10, 10, 10)))
        eq_(self.mgrid.meta_tile((0, 0, 1)).size, (512, 256))

    def test_tiles_level_1(self):
        eq_(list(self.mgrid.meta_tile((0, 0, 1)).tile_patterns),
            [
                ((0, 0, 1), (0, 0)),
                ((1, 0, 1), (256, 0))
            ])
    def test_tile_list_level_1(self):
        eq_(list(self.mgrid.tile_list((0, 0, 1))),
            [(0, 0, 1), (1, 0, 1)])

    def test_meta_bbox_level_2(self):
        eq_(self.mgrid._meta_bbox((0, 0, 2)), ((-180, -90, 3.515625, 90), (0, 0, 10, 0)))
        eq_(self.mgrid._meta_bbox((0, 0, 2), limit_to_bbox=False),
            ((-183.515625, -93.515625, 3.515625, 93.515625), (10, 10, 10, 10)))
        eq_(self.mgrid.meta_tile((0, 0, 2)).size, (522, 512))

        eq_(self.mgrid._meta_bbox((2, 0, 2)), ((-3.515625, -90, 180, 90), (10, 0, 0, 0)))
        meta_tile = self.mgrid.meta_tile((2, 0, 2))
        eq_(meta_tile.size, (522, 512))
        eq_(meta_tile.grid_size, (2, 2))

    def test_tiles_level_2(self):
        eq_(list(self.mgrid.meta_tile((0, 0, 2)).tile_patterns),
            [
                ((0, 1, 2), (0, 0)),
                ((1, 1, 2), (256, 0)),
                ((0, 0, 2), (0, 256)),
                ((1, 0, 2), (256, 256)),
            ])
        eq_(list(self.mgrid.meta_tile((2, 0, 2)).tile_patterns),
            [
                ((2, 1, 2), (10, 0)),
                ((3, 1, 2), (266, 0)),
                ((2, 0, 2), (10, 256)),
                ((3, 0, 2), (266, 256)),
            ])

    def test_tile_list_level_2(self):
        eq_(list(self.mgrid.tile_list((0, 0, 2))),
            [(0, 1, 2), (1, 1, 2), (0, 0, 2), (1, 0, 2)])
        eq_(list(self.mgrid.tile_list((1, 1, 2))),
            [(0, 1, 2), (1, 1, 2), (0, 0, 2), (1, 0, 2)])

    def test_tiles_level_3(self):
        eq_(list(self.mgrid.meta_tile((2, 0, 3)).tile_patterns),
            [
                ((2, 1, 3), (10, 10)),
                ((3, 1, 3), (266, 10)),
                ((2, 0, 3), (10, 266)),
                ((3, 0, 3), (266, 266)),
            ])
        eq_(list(self.mgrid.meta_tile((2, 2, 3)).tile_patterns),
            [
                ((2, 3, 3), (10, 0)),
                ((3, 3, 3), (266, 0)),
                ((2, 2, 3), (10, 256)),
                ((3, 2, 3), (266, 256)),
            ])

class TestMetaGridGeodeticUL(object):
    def setup(self):
        self.tile_grid = tile_grid('EPSG:4326', origin='ul')
        self.mgrid = MetaGrid(grid=self.tile_grid, meta_size=(2, 2), meta_buffer=10)

    def test_tiles_level_0(self):
        meta_tile = self.mgrid.meta_tile((0, 0, 0))
        eq_(meta_tile.bbox, (-180, -90, 180, 90))
        eq_(meta_tile.size, (256, 128))
        eq_(meta_tile.grid_size, (1, 1))
        eq_(meta_tile.tile_patterns, [((0, 0, 0), (0, 0))])


    def test_tiles_level_1(self):
        meta_tile = self.mgrid.meta_tile((0, 0, 1))
        eq_(meta_tile.bbox, (-180, -90, 180, 90))
        eq_(meta_tile.size, (512, 256))
        eq_(meta_tile.grid_size, (2, 1))
        eq_(list(meta_tile.tile_patterns),
            [
                ((0, 0, 1), (0, 0)),
                ((1, 0, 1), (256, 0))
            ])

    def test_tile_list_level_1(self):
        eq_(list(self.mgrid.tile_list((0, 0, 1))),
            [(0, 0, 1), (1, 0, 1)])

    def test_tiles_level_2(self):
        meta_tile = self.mgrid.meta_tile((0, 0, 2))
        eq_(meta_tile.bbox, (-180, -90, 3.515625, 90))
        eq_(meta_tile.size, (522, 512))
        eq_(meta_tile.grid_size, (2, 2))
        eq_(meta_tile.tile_patterns,
            [
                ((0, 0, 2), (0, 0)),
                ((1, 0, 2), (256, 0)),
                ((0, 1, 2), (0, 256)),
                ((1, 1, 2), (256, 256)),
            ])
        eq_(list(self.mgrid.meta_tile((2, 0, 2)).tile_patterns),
            [
                ((2, 0, 2), (10, 0)),
                ((3, 0, 2), (266, 0)),
                ((2, 1, 2), (10, 256)),
                ((3, 1, 2), (266, 256)),
            ])

    def test_tile_list_level_2(self):
        eq_(list(self.mgrid.tile_list((0, 0, 2))),
            [(0, 0, 2), (1, 0, 2), (0, 1, 2), (1, 1, 2)])
        eq_(list(self.mgrid.tile_list((1, 1, 2))),
            [(0, 0, 2), (1, 0, 2), (0, 1, 2), (1, 1, 2)])

    def test_tiles_level_3(self):
        meta_tile = self.mgrid.meta_tile((2, 0, 3))
        eq_(meta_tile.bbox, (-91.7578125, -1.7578125, 1.7578125, 90))
        eq_(meta_tile.size, (532, 522))
        eq_(meta_tile.grid_size, (2, 2))
        eq_(list(self.mgrid.meta_tile((2, 0, 3)).tile_patterns),
            [
                ((2, 0, 3), (10, 0)),
                ((3, 0, 3), (266, 0)),
                ((2, 1, 3), (10, 256)),
                ((3, 1, 3), (266, 256)),
            ])
        eq_(list(self.mgrid.meta_tile((2, 2, 3)).tile_patterns),
            [
                ((2, 2, 3), (10, 10)),
                ((3, 2, 3), (266, 10)),
                ((2, 3, 3), (10, 266)),
                ((3, 3, 3), (266, 266)),
            ])

 def setup(self):
     self.mgrid = MetaGrid(grid=tile_grid('EPSG:4326'), meta_size=(2, 2), meta_buffer=10)

class TileManager(object):
    """
    Manages tiles for a single grid.
    Loads tiles from the cache, creates new tiles from sources and stores them
    into the cache, or removes tiles.

    :param pre_store_filter: a list with filter. each filter will be called
        with a tile before it will be stored to disc. the filter should
        return this or a new tile object.
    """
    def __init__(self, grid, cache, sources, format, image_opts=None, request_format=None,
        meta_buffer=None, meta_size=None, minimize_meta_requests=False,
        pre_store_filter=None, concurrent_tile_creators=1):
        self.grid = grid
        self.cache = cache
        self.meta_grid = None
        self.format = format
        self.image_opts = image_opts
        self.request_format = request_format or format
        self.sources = sources
        self.minimize_meta_requests = minimize_meta_requests
        self._expire_timestamp = None
        self.transparent = self.sources[0].transparent
        self.pre_store_filter = pre_store_filter or []
        self.concurrent_tile_creators = concurrent_tile_creators

        if meta_buffer or (meta_size and not meta_size == [1, 1]):
            if all(source.supports_meta_tiles for source in sources):
                self.meta_grid = MetaGrid(grid, meta_size=meta_size, meta_buffer=meta_buffer)
            elif any(source.supports_meta_tiles for source in sources):
                raise ValueError('meta tiling configured but not supported by all sources')

    @contextmanager
    def session(self):
        """
        Context manager for access to the cache. Cleans up after usage
        for connection based caches.

        >>> with tile_manager.session(): #doctest: +SKIP
        ...    tile_manager.load_tile_coords(tile_coords)

        """
        yield
        self.cleanup()

    def cleanup(self):
        if hasattr(self.cache, 'cleanup'):
            self.cache.cleanup()

    def load_tile_coord(self, tile_coord, dimensions=None, with_metadata=False):
        tile = Tile(tile_coord)
        self.cache.load_tile(tile, with_metadata)

        if tile.coord is not None and not self.is_cached(tile, dimensions=dimensions):
            # missing or staled
            creator = self.creator(dimensions=dimensions)
            created_tiles = creator.create_tiles([tile])
            for created_tile in created_tiles:
                if created_tile.coord == tile_coord:
                    return created_tile

        return tile

    def load_tile_coords(self, tile_coords, dimensions=None, with_metadata=False):
        tiles = TileCollection(tile_coords)
        uncached_tiles = []

        # load all in batch
        self.cache.load_tiles(tiles, with_metadata)

        for tile in tiles:
            if tile.coord is not None and not self.is_cached(tile, dimensions=dimensions):
                # missing or staled
                uncached_tiles.append(tile)

        if uncached_tiles:
            creator = self.creator(dimensions=dimensions)
            created_tiles = creator.create_tiles(uncached_tiles)
            for created_tile in created_tiles:
                if created_tile.coord in tiles:
                    tiles[created_tile.coord].source = created_tile.source

        return tiles

    def remove_tile_coords(self, tile_coords, dimensions=None):
        tiles = TileCollection(tile_coords)
        self.cache.remove_tiles(tiles)

    def creator(self, dimensions=None):
        return TileCreator(self.cache, self.sources, self.grid, self.meta_grid,
            self, dimensions=dimensions)

    def lock(self, tile):
        if self.meta_grid:
            tile = Tile(self.meta_grid.main_tile(tile.coord))
        return self.cache.lock(tile)

    def is_cached(self, tile, dimensions=None):
        """
        Return True if the tile is cached.
#.........这里部分代码省略.........