from abc import ABC, abstractmethod
from collections.abc import Generator
from pathlib import Path
from typing import Any, Generic, Optional, TypeVar
from pyproj import CRS
from .shared_metadata import SharedMetadata
from .tiler_worker import TilerWorker
_SharedMetadataT = TypeVar("_SharedMetadataT", bound=SharedMetadata)
_TilerWorkerT = TypeVar("_TilerWorkerT", bound=TilerWorker[Any])
[docs]
class Tiler(ABC, Generic[_SharedMetadataT, _TilerWorkerT]):
"""
Tiler abstract class, this list of attributes and methods is used by convert.
This class will organize the different tasks and their order of dispatch to the TilerWorker instances.
You must set a name as class attribute and overwrite all abstract methods.
Some methods are not required, overwrite them only if needed (like `validate_binary_data` or `memory_control`)
"""
name = b""
shared_metadata: _SharedMetadataT
[docs]
@abstractmethod
def initialization(
self,
crs_out: Optional[CRS],
working_dir: Path,
number_of_jobs: int,
) -> None:
"""
The __init__ method must only set attributes without any action.
It is in this method that this work must be done (and the initialization of shared_metadata).
The method will be called before all others.
"""
[docs]
@abstractmethod
def get_worker(self) -> _TilerWorkerT:
"""
Returns an instantiated tiler worker.
"""
[docs]
@abstractmethod
def get_tasks(
self, startup: float
) -> Generator[tuple[bytes, list[bytes]], None, None]:
"""
Yields tasks to be sent to workers.
This methods will get called by the main convert function each time it wants new tasks to be fed to workers.
Implementors should each time returns the task that has the biggest priority.
"""
[docs]
@abstractmethod
def process_message(self, return_type: bytes, content: list[bytes]) -> bool:
"""
Updates the state of the tiler in function of the return type and the returned data
"""
[docs]
@abstractmethod
def write_tileset(self) -> None:
"""
Writes the tileset file once the binary data written
"""
[docs]
def validate_binary_data(self) -> None:
"""
Checks if the state of the tiler or the binary data written is correct.
This method is called after the end of the conversion of this tiler (but before write_tileset)
"""
[docs]
def memory_control(self) -> None:
"""
Method called at the end of each loop of the convert method.
Checks if there is no too much memory used by the tiler and do actions in function
"""
[docs]
@abstractmethod
def print_summary(self) -> None:
"""
Prints the summary of the tiler before the start of the conversion.
"""
[docs]
def benchmark(self, benchmark_id: str, startup: float) -> None:
"""
Prints benchmark info at the end of the conversion of this tiler and the writing of the tileset.
"""
[docs]
@abstractmethod
def print_debug(
self, now: float, number_of_jobs: int, number_of_idle_clients: int
) -> None:
"""
Prints info about the progression of the conversion. Called everytime a tiler worker task is finished.
"""