## ZooUI - Zooming User Interface
## Copyright (C) 2009 David Roberts <d@vidr.cc>
##
## This program is free software; you can redistribute it and/or
## modify it under the terms of the GNU General Public License
## as published by the Free Software Foundation; either version 3
## of the License, or (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, see <https://www.gnu.org/licenses/>.
"""The TileManager is responsible for requesting tiles from TileProviders,
caching them in memory, and providing them to MediaObjects when requested
to do so.
It is also responsible for creating new tiles from available ones when
no tiles of the requested resolution are available.
"""
import math
from typing import TYPE_CHECKING, Any, Optional
from zooui.logger import get_logger
from . import tilestore as TileStore
from .tileproviders import FernTileProvider, StaticTileProvider
from .tilestore import TileCache
## Performance optimization note:
## Phase 2 optimizations replace 2**x with math.exp2(x) (1.85x faster)
## and math.log(x, 2) with math.log2(x) (2x faster) throughout the codebase.
## These changes are performance-critical for zoom operations.
if TYPE_CHECKING:
from logging import Logger
from .tile import Tile
from .tileproviders import FernTileProvider, StaticTileProvider
from .tilestore import TileCache
TileID = tuple[str, int, int, int]
# Module-level global variables (initialized by init())
__tilecache: Optional["TileCache"] = None
__temptilecache: Optional["TileCache"] = None
__tp_static: Optional["StaticTileProvider"] = None
__tp_dynamic: dict[str, "FernTileProvider"] = {}
__logger: Optional["Logger"] = None
# Cleanup parameters for shutdown execution
__cleanup_enabled: bool = False
__cleanup_max_age_days: int = 3
__cleanup_executed: bool = False
__atexit_registered: bool = False
[docs]
def init(
total_cache_size: int = 1024,
auto_cleanup: bool = True,
cleanup_max_age_days: int = 3,
collect_cleanup_stats: bool = False,
) -> None:
"""
Function :
init(total_cache_size, auto_cleanup, cleanup_max_age_days, collect_cleanup_stats)
Parameters :
total_cache_size : int
- Total cache size: number of total cached tiles (default: 1024)
auto_cleanup : bool
- Enable automatic cleanup of old tiles (default: True)
cleanup_max_age_days : int
- Maximum age in days for tiles (default: 3)
collect_cleanup_stats : bool
- Collect detailed before/after cleanup statistics (default: False)
- Setting to False improves startup performance on large tilestores
init(total_cache_size, auto_cleanup, cleanup_max_age_days, collect_cleanup_stats) --> None
Initialise the TileManager. This **must** be called before any other
functions are called.
The central coordinator that:
- Routes tile requests to appropriate providers
- Manages two-tier caching (80% permanent / 20% temporary)
- Synthesizes missing tiles from available ones via cut_tile()
- Key methods: load_tile(), get_tile(), get_tile_robust()
"""
global __tilecache, __temptilecache, __tp_static, __tp_dynamic, __logger
global __cleanup_enabled, __cleanup_max_age_days, __cleanup_executed, __atexit_registered
# Shut down any previous threads before re-initializing to prevent
# thread leaks when init() is called multiple times (e.g. in tests).
if __tilecache is not None or __tp_static is not None:
_shutdown_threads()
# Store cleanup parameters for shutdown execution
__cleanup_enabled = auto_cleanup
__cleanup_max_age_days = cleanup_max_age_days
__cleanup_executed = False
# tile cache reserved for static tile provider.
__tilecache = TileCache(int(0.8 * total_cache_size))
# tile cache reserved for dynamic tile provider
__temptilecache = TileCache(int(0.2 * total_cache_size))
# creates a tileproviders.tileprovider(tilecache: Any) thread instance
# and starts it
__tp_static = StaticTileProvider(__tilecache)
__tp_static.start()
# define dynamic tile providers instances list
__tp_dynamic = {
"dynamic:fern": FernTileProvider(__tilecache),
}
# Starts dynamic tile providers thread instances
for tp in list(__tp_dynamic.values()):
tp.start()
# set up TileManager logger
__logger = get_logger("TileManager")
import atexit
# Register thread shutdown at exit — ensures background threads are
# stopped before Python destroys Qt internals. Registered only once
# even if init() is called multiple times (e.g. in tests).
global __atexit_registered
if not __atexit_registered:
atexit.register(_shutdown_threads)
__atexit_registered = True
# Register shutdown cleanup if enabled
if auto_cleanup:
atexit.register(_shutdown_cleanup)
__logger.info("Tilestore cleanup registered for shutdown execution")
else:
__logger.debug("Tilestore auto cleanup disabled")
[docs]
def shutdown() -> None:
"""Stop all tile provider threads and tile cache threads.
This should be called during application shutdown (via aboutToQuit)
to ensure background threads are joined before Qt begins
destroying its internals.
"""
_shutdown_threads()
_shutdown_cleanup()
[docs]
def _shutdown_threads() -> None:
"""Stop and join all TileProvider and TileCache background threads."""
global __tp_static, __tp_dynamic, __tilecache, __temptilecache, __logger
# Signal all TileProvider threads to stop
if __tp_static and getattr(__tp_static, "is_alive", lambda: False)():
__tp_static.stop()
for tp in list(__tp_dynamic.values()):
if tp and getattr(tp, "is_alive", lambda: False)():
tp.stop()
# Signal TileCache periodic clean threads to stop
if __tilecache:
getattr(__tilecache, "shutdown", lambda: None)()
if __temptilecache:
getattr(__temptilecache, "shutdown", lambda: None)()
# Wait for TileProvider threads to finish (with timeout)
if __tp_static and getattr(__tp_static, "is_alive", lambda: False)():
__tp_static.join(timeout=2.0)
for tp in list(__tp_dynamic.values()):
if tp and getattr(tp, "is_alive", lambda: False)():
tp.join(timeout=2.0)
if __logger:
__logger.debug("tile provider threads shut down")
[docs]
def _shutdown_cleanup() -> None:
"""
Function :
_shutdown_cleanup()
Parameters :
None
_shutdown_cleanup() --> None
Execute tilestore cleanup on application shutdown.
This function is registered with atexit and connected to Qt's
aboutToQuit signal. It runs cleanup with stored parameters
and prevents duplicate execution.
"""
global __cleanup_executed, __cleanup_enabled, __cleanup_max_age_days, __logger
# Check if cleanup should run and hasn't already run
if __cleanup_enabled and not __cleanup_executed:
__cleanup_executed = True # Prevent duplicate execution
if __logger:
__logger.info("Running tilestore cleanup on shutdown")
try:
# Run cleanup with fast mode (skip detailed statistics)
TileStore.auto_cleanup(
max_age_days=__cleanup_max_age_days,
enable=True,
collect_stats=False, # Skip detailed stats for faster cleanup
)
except Exception as e:
if __logger:
__logger.error(f"Error during shutdown cleanup: {e}")
# Don't propagate exception - cleanup shouldn't prevent shutdown
[docs]
def load_tile(tile_id: TileID) -> None:
"""
Function :
load_tile(tile_id)
Parameters :
tile_id : Tuple[str, int, int, int]
load_tile(tile_id) --> None
Request that the tile identified by `tile_id` be loaded into the
tilecache.
"""
media_id = tile_id[0]
if media_id in __tp_dynamic:
__tp_dynamic[media_id].request(tile_id)
else:
if __tp_static:
__tp_static.request(tile_id)
[docs]
def get_tile(tile_id: TileID) -> "Tile":
"""
Function :
get_tile(tile_id)
Parameters :
tile_id : Tuple[str, int, int, int]
get_tile(tile_id) --> Tile
Return the requested tile identified by `tile_id`.
If the tile is not available in the tilecache, one of three errors will be
raised: :class:`MediaNotTiled`, :class:`TileNotLoaded`, or :class:`TileNotAvailable`
"""
if tile_id[1] < 0:
## negative tilelevel
raise TileNotAvailable
try:
if __tilecache is None:
raise KeyError
tile = __tilecache[tile_id]
except KeyError:
media_id = tile_id[0]
if tiled(media_id):
load_tile(tile_id)
raise TileNotLoaded from None
else:
raise MediaNotTiled from None
if tile:
return tile
else:
raise TileNotAvailable
[docs]
def cut_tile(tile_id: TileID, tempcache: int = 0) -> tuple["Tile", bool]:
"""
Function :
cut_tile(tile_id, tempcache)
Parameters :
tile_id : Tuple[str, int, int, int]
tempcache : int
cut_tile(tile_id, tempcache) --> Tuple[Tile, bool]
Create a tile from resizing and cropping those loaded into the tile
cache. Returns a tuple containing the tile, and a bool `final` which is
False iff the tile is not the greatest resolution possible and should
therefore not be cached indefinitely.
If `tempcache` > 0, then tiles with `final`=False will cached in the
:class:`TileCache`, but will expire after they have been accessed `tempcache` times.
This function should only be called if a :class:`TileNotLoaded` or
:class:`TileNotAvailable` error has been encountered.
Precondition: the (0,0,0) tile exists for the given media
Precondition: the requested tile doesn't fall outside the bounds of the
image
"""
media_id, tilelevel, row, col = tile_id
tilesize_val = get_metadata(media_id, "tilesize")
if tilesize_val is None:
raise ValueError(f"Tilesize not found for media_id: {media_id}")
tilesize = int(tilesize_val)
if tempcache <= 0:
## purge temporary tiles
if __temptilecache:
__temptilecache.purge()
if tilelevel < 0:
## resize the (0,0,0) tile
if __tilecache is None:
raise KeyError
tile000 = __tilecache[media_id, 0, 0, 0]
scale = math.exp2(tilelevel)
tile = tile000.resize(int(tile000.size[0] * scale), int(tile000.size[1] * scale))
final = True
else:
big_tile_id = (media_id, tilelevel - 1, row // 2, col // 2)
try:
return get_tile(tile_id), True
except TileNotLoaded:
final = False
try:
## check if there is a temporary cut tile in the cache
if __temptilecache is None:
raise KeyError
return __temptilecache[tile_id], False
except KeyError:
## don't worry if there isn't
pass
big_tile = cut_tile(big_tile_id)[0]
except TileNotAvailable:
big_tile, final = cut_tile(big_tile_id)
if col % 2 == 0:
x1 = 0
x2 = min(tilesize // 2, big_tile.size[0])
else:
x1 = tilesize // 2
x2 = big_tile.size[0]
if row % 2 == 0:
y1 = 0
y2 = min(tilesize // 2, big_tile.size[1])
else:
y1 = tilesize // 2
y2 = big_tile.size[1]
tile = big_tile.crop((x1, y1, x2, y2))
tile = tile.resize(2 * tile.size[0], 2 * tile.size[1])
if final:
if __tilecache:
__tilecache[tile_id] = tile
elif tempcache > 0 and __temptilecache:
__temptilecache.insert(tile_id, tile, tempcache)
return tile, final
[docs]
def get_tile_robust(tile_id: TileID) -> "Tile":
"""
Function :
get_tile_robust(tile_id)
Parameters :
tile_id : Tuple[str, int, int, int]
get_tile_robust(tile_id) --> Tile
Will try returning the result of :func:`get_tile`, and if that fails will
return the result of :func:`cut_tile`.
This function will not raise :class:`TileNotLoaded` or :class:`TileNotAvailable`, but may
raise :class:`MediaNotTiled`.
"""
try:
return get_tile(tile_id)
except (TileNotLoaded, TileNotAvailable):
return cut_tile(tile_id)[0]
[docs]
def tiled(media_id: str) -> bool:
"""
Function :
tiled(media_id)
Parameters :
media_id : str
tiled(media_id) --> bool
Returns True iff the media identified by `media_id` has been tiled.
Will always return True for dynamic media.
"""
return media_id.startswith("dynamic:") or TileStore.tiled(media_id)
[docs]
def purge(media_id: str | None = None) -> None:
"""
Function :
purge(media_id)
Parameters :
media_id : Optional[str]
purge(media_id) --> None
Purge the specified *media_id* from the *TileProviders*. If *media_id*
is omitted then all media will be purged.
Precondition: the media to be purged should not be active (i.e. no
*MediaObjects* for the media should exist).
"""
if __tp_static:
__tp_static.purge(media_id)
for tp in list(__tp_dynamic.values()):
tp.purge(media_id)
[docs]
def pause() -> None:
"""
Function :
pause()
Parameters :
None
pause() --> None
Pause all TileProvider threads. This should be called before running
converter processes to avoid conflicts between pyvips and tile loading.
"""
if __tp_static:
__tp_static.pause()
for tp in list(__tp_dynamic.values()):
tp.pause()
if __logger:
__logger.debug("all tile providers paused")
[docs]
def resume() -> None:
"""
Function :
resume()
Parameters :
None
resume() --> None
Resume all TileProvider threads after they were paused.
"""
if __tp_static:
__tp_static.resume()
for tp in list(__tp_dynamic.values()):
tp.resume()
if __logger:
__logger.debug("all tile providers resumed")
[docs]
class TileNotLoaded(Exception):
"""Exception for when tiles are requested before they have been loaded into
the tile cache."""
pass
[docs]
class TileNotAvailable(Exception):
"""Exception for when an attempt to load the requested tile has previously
failed."""
pass