diff --git a/modules/map/api.py b/modules/map/api.py index 19481825..77eba4c6 100644 --- a/modules/map/api.py +++ b/modules/map/api.py @@ -6,6 +6,7 @@ from sanic_ext import validate from sanic_ext.extensions.openapi import openapi from modules.map import lib, schema +from wrolpi import flags from wrolpi.common import wrol_mode_check, get_media_directory, background_task from wrolpi.errors import ValidationError from wrolpi.root_api import get_blueprint, json_response @@ -22,7 +23,7 @@ bp = get_blueprint('Map', '/api/map') @validate(schema.ImportPost) @wrol_mode_check async def import_pbfs(_: Request, body: schema.ImportPost): - if lib.IMPORT_EVENT.is_set(): + if flags.map_importing.is_set(): return response.json({'error': 'Map import already running'}, HTTPStatus.CONFLICT) paths = [i for i in body.files if i] @@ -48,7 +49,7 @@ def get_files_status(_: Request): body = dict( files=paths, pending=pending, - import_running=lib.IMPORT_EVENT.is_set(), + import_running=flags.map_importing.is_set(), dockerized=DOCKERIZED, ) return json_response(body, HTTPStatus.OK) diff --git a/modules/map/lib.py b/modules/map/lib.py index 2e023100..0ee32bc5 100644 --- a/modules/map/lib.py +++ b/modules/map/lib.py @@ -1,21 +1,22 @@ import asyncio import subprocess -from multiprocessing import Event, Manager +from multiprocessing import Manager from pathlib import Path from typing import List from sqlalchemy.orm import Session from modules.map.models import MapFile +from wrolpi import flags from wrolpi.cmd import BASH_BIN from wrolpi.common import get_media_directory, walk, logger, wrol_mode_check -from wrolpi.dates import now, timedelta_to_timestamp +from wrolpi.dates import now, timedelta_to_timestamp, seconds_to_timestamp from wrolpi.db import optional_session, get_db_session +from wrolpi.events import Events from wrolpi.vars import PYTEST, PROJECT_DIR logger = logger.getChild(__name__) -IMPORT_EVENT = Event() IMPORTING = Manager().dict(dict( pending=None, )) @@ -81,87 +82,93 @@ import_logger = logger.getChild('import') async def import_files(paths: List[str]): - if IMPORT_EVENT.is_set(): + if flags.map_importing.is_set(): import_logger.error('Map import already running...') return - paths = [get_media_directory() / i for i in paths] - import_logger.warning(f'Importing: {", ".join(map(str, paths))}') + with flags.map_importing: + paths = [get_media_directory() / i for i in paths] + import_logger.warning(f'Importing: {", ".join(map(str, paths))}') - dumps = [i for i in paths if i.suffix == '.dump'] - pbfs = [i for i in paths if i.suffix == '.pbf'] + dumps = [i for i in paths if i.suffix == '.dump'] + pbfs = [i for i in paths if i.suffix == '.pbf'] - any_success = False - try: - IMPORT_EVENT.set() - if pbfs: - success = False - try: - IMPORTING.update(dict( - pending=list(pbfs), - )) - await run_import_command(*pbfs) - success = True - any_success = True - except Exception as e: - import_logger.warning('Failed to run import', exc_info=e) - finally: - IMPORTING.update(dict( - pending=None, - )) + total_elapsed = 0 - if success: - with get_db_session(commit=True) as session: - # Any previously imported PBFs are no longer imported. - for pbf_path in pbfs: - pbf_file = get_or_create_map_file(pbf_path, session) - pbf_file.imported = True - for pbf_file in session.query(MapFile): - pbf_file.imported = pbf_file.path in pbfs + any_success = False + try: + if pbfs: + success = False + try: + IMPORTING.update(dict( + pending=list(pbfs), + )) + total_elapsed += await run_import_command(*pbfs) + success = True + any_success = True + except Exception as e: + import_logger.warning('Failed to run import', exc_info=e) + finally: + IMPORTING.update(dict( + pending=None, + )) - for path in dumps: - # Import each dump individually. - if not path.is_file(): - import_logger.fatal(f'Map file does not exist! {path}') - continue + if success: + with get_db_session(commit=True) as session: + # Any previously imported PBFs are no longer imported. + for pbf_path in pbfs: + pbf_file = get_or_create_map_file(pbf_path, session) + pbf_file.imported = True + for pbf_file in session.query(MapFile): + pbf_file.imported = pbf_file.path in pbfs - with get_db_session() as session: - map_file = session.query(MapFile).filter_by(path=path).one_or_none() - if map_file and map_file.imported: - # Don't import a map file twice. - import_logger.debug(f'{path} is already imported') + for path in dumps: + # Import each dump individually. + if not path.is_file(): + import_logger.fatal(f'Map file does not exist! {path}') continue - success = False - try: - IMPORTING.update(dict( - pending=str(path), - )) - await run_import_command(path) - success = True - any_success = True - except Exception as e: - import_logger.warning('Failed to run import', exc_info=e) - finally: - IMPORTING.update(dict( - pending=None, - )) + with get_db_session() as session: + map_file = session.query(MapFile).filter_by(path=path).one_or_none() + if map_file and map_file.imported: + # Don't import a map file twice. + import_logger.debug(f'{path} is already imported') + continue - if success: - with get_db_session(commit=True) as session: - map_file = get_or_create_map_file(path, session) - map_file.imported = True - finally: - if any_success: - # A map was imported, remove the tile cache files. - await clear_mod_tile() - IMPORT_EVENT.clear() + success = False + try: + IMPORTING.update(dict( + pending=str(path), + )) + total_elapsed += await run_import_command(path) + success = True + any_success = True + except Exception as e: + import_logger.warning('Failed to run import', exc_info=e) + finally: + IMPORTING.update(dict( + pending=None, + )) + + if success: + with get_db_session(commit=True) as session: + map_file = get_or_create_map_file(path, session) + map_file.imported = True + finally: + if any_success: + # A map was imported, remove the tile cache files. + await clear_mod_tile() + Events.send_map_import_complete(f'Map import completed; took {seconds_to_timestamp(total_elapsed)}') + else: + Events.send_map_import_failed(f'Map import failed! See server logs.') -async def run_import_command(*paths: Path): +async def run_import_command(*paths: Path) -> int: """Run the map import script on the provided paths. Can only import a single *.dump file, or a list of *.osm.pbf files. They cannot be mixed. + + @return: The seconds elapsed during import. """ paths = [i.absolute() for i in paths] dumps = [i for i in paths if i.suffix == '.dump'] @@ -203,6 +210,8 @@ async def run_import_command(*paths: Path): import_logger.error(line[:500]) raise ValueError(f'Importing map file failed with return code {proc.returncode}') + return int(elapsed.total_seconds()) + @optional_session def get_import_status(session: Session = None) -> List[MapFile]: @@ -227,3 +236,12 @@ async def clear_mod_tile(): logger.warning('Clearing map tile cache files') await asyncio.create_subprocess_shell(f'sudo /opt/wrolpi/scripts/clear_map_cache.sh') + + +# Bps calculated using many tests on a well-cooled RPi4. +RPI4_PBF_BYTES_PER_SECOND = 61879 + + +def seconds_to_import(size_in_bytes: int) -> int: + """Attempt to predict how long it will take an RPi4 to import a given PBF file.""" + return max(int(size_in_bytes // RPI4_PBF_BYTES_PER_SECOND), 0) diff --git a/modules/map/models.py b/modules/map/models.py index 8a166ea3..1259906d 100644 --- a/modules/map/models.py +++ b/modules/map/models.py @@ -1,23 +1,10 @@ -from pathlib import Path - from sqlalchemy import Column, Integer, Boolean, BigInteger +from modules.map import lib from wrolpi import dates from wrolpi.common import Base, ModelHelper from wrolpi.media_path import MediaPathType -# Bps calculated using many tests on a well-cooled RPi4. -RPI4_PBF_BYTES_PER_SECOND = 3943 - - -def seconds_to_import(path: Path, size: int) -> int: - """Attempt to predict how long it will take an RPi4 to import a given PBF file.""" - if str(path).endswith('.osm.pbf'): - return max(int(size // RPI4_PBF_BYTES_PER_SECOND), 0) - - # Can't calculate an unknown file. - return 0 - class MapFile(Base, ModelHelper): __tablename__ = 'map_file' @@ -35,7 +22,7 @@ class MapFile(Base, ModelHelper): 'id': self.id, 'imported': self.imported, 'path': self.path, - 'seconds_to_import': (seconds := seconds_to_import(self.path, self.size)), + 'seconds_to_import': (seconds := lib.seconds_to_import(self.size)), 'size': self.size, 'time_to_import': dates.seconds_to_timestamp(seconds), } diff --git a/modules/map/test/test_models.py b/modules/map/test/test_models.py index 4fb79ebc..46d7a5d5 100644 --- a/modules/map/test/test_models.py +++ b/modules/map/test/test_models.py @@ -2,17 +2,18 @@ from pathlib import Path import pytest +import modules.map.lib from modules.map import models @pytest.mark.parametrize('size,expected', [ (0, 0), (-1, 0), - (17737381, 4498), - (63434267, 16087), - (87745484, 22253), - (136372996, 34586), - (116318111, 29499), + (17737381, 252), + (63434267, 904), + (87745484, 1251), + (136372996, 1944), + (116318111, 1658), ]) def test_seconds_to_import_rpi4(size, expected): - assert models.seconds_to_import(Path('file.osm.pbf'), size) == expected + assert modules.map.lib.seconds_to_import(size) == expected diff --git a/wrolpi/dates.py b/wrolpi/dates.py index d44caeb6..7daec46c 100644 --- a/wrolpi/dates.py +++ b/wrolpi/dates.py @@ -68,7 +68,15 @@ def from_timestamp(timestamp: float) -> datetime: def seconds_to_timestamp(seconds: Union[int, float]) -> str: - """Convert an integer into a timestamp string.""" + """Convert an integer into a timestamp string. + + >>> seconds_to_timestamp(5) + # '00:00:05' + >>> seconds_to_timestamp(60) + # '00:01:00' + >>> seconds_to_timestamp(86400) + # '1d 00:00:00' + """ seconds = int(seconds) weeks, seconds = divmod(seconds, Seconds.week) days, seconds = divmod(seconds, Seconds.day) diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index 2d31341f..148e0096 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -516,7 +516,6 @@ class DownloadManager: download = Download(url=url, status='new') session.add(download) session.flush() - print(f'{download.settings=}') download.manager = self return download diff --git a/wrolpi/events.py b/wrolpi/events.py index 7c222243..3c2451a5 100644 --- a/wrolpi/events.py +++ b/wrolpi/events.py @@ -65,6 +65,14 @@ class Events: def send_created(message: str = None): send_event('created', message, subject='created') + @staticmethod + def send_map_import_complete(message: str = None): + send_event('map_import_complete', message, subject='map') + + @staticmethod + def send_map_import_failed(message: str = None): + send_event('map_import_failed', message, subject='map') + def log_event(event: str, message: str = None, action: str = None, subject: str = None): log = f'{event=}' diff --git a/wrolpi/flags.py b/wrolpi/flags.py index 998aff52..b31f8374 100644 --- a/wrolpi/flags.py +++ b/wrolpi/flags.py @@ -114,6 +114,8 @@ refreshing = Flag('refreshing') outdated_zims = Flag('outdated_zims', store_db=True) # Kiwix server needs to be restarted. kiwix_restart = Flag('kiwix_restart') +# Map Importing +map_importing = Flag('map_importing') # Steps of refreshing. refresh_counting = Flag('refresh_counting') diff --git a/wrolpi/test/test_dates.py b/wrolpi/test/test_dates.py index b7408daf..d919beb0 100644 --- a/wrolpi/test/test_dates.py +++ b/wrolpi/test/test_dates.py @@ -6,7 +6,7 @@ from sqlalchemy import Column, Integer from wrolpi import dates from wrolpi.common import Base -from wrolpi.dates import TZDateTime, timedelta_to_timestamp +from wrolpi.dates import TZDateTime, timedelta_to_timestamp, seconds_to_timestamp from wrolpi.db import get_db_session, get_db_curs @@ -56,6 +56,18 @@ def test_TZDateTime(test_session): assert_raw_datetime('2021-10-05T23:20:10.346823') +@pytest.mark.parametrize('td,expected', [ + (0, '00:00:00'), + (1, '00:00:01'), + (60, '00:01:00'), + (61, '00:01:01'), + (65, '00:01:05'), + (24 * 60 * 60, '1d 00:00:00'), +]) +def test_seconds_to_timestamp(td, expected): + assert seconds_to_timestamp(td) == expected + + @pytest.mark.parametrize('td,expected', [ (timedelta(seconds=0), '00:00:00'), (timedelta(seconds=1), '00:00:01'),