rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 19:01:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Using flags.map_importing to handle map importing control. Updating RPi 4 map import estimate using create import tests, rather than slower append import tests.

Browse source
This commit is contained in:
Roland 2023-08-17 08:27:49 -06:00
parent 6405188a98
commit 6376e6539b
No known key found for this signature in database
GPG key ID: 9D6DBD3ACF11AB93
9 changed files with 130 additions and 94 deletions
Download patch file Download diff file Expand all files Collapse all files

5
modules/map/api.py
View file

@ -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)

154
modules/map/lib.py
View file

@ -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)

17
modules/map/models.py
View file

@ -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),
}

13
modules/map/test/test_models.py
View file

@ -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

10
wrolpi/dates.py
View file

@ -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)

1
wrolpi/downloader.py
View file

@ -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

8
wrolpi/events.py
View file

@ -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=}'

2
wrolpi/flags.py
View file

@ -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')

14
wrolpi/test/test_dates.py
View file

@ -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'),