diff --git a/.github/workflows/github-action-test.yaml b/.github/workflows/github-action-test.yaml index c1644e8..39ef5f2 100644 --- a/.github/workflows/github-action-test.yaml +++ b/.github/workflows/github-action-test.yaml @@ -6,7 +6,7 @@ jobs: strategy: matrix: os: [ubuntu-latest, windows-latest, macos-latest] - python-version: ["3.10", "3.11", "3.12"] + python-version: ["3.10", "3.11"] steps: - uses: actions/checkout@v2 @@ -38,4 +38,4 @@ jobs: message: ${{ env.total }}% minColorRange: 50 maxColorRange: 90 - valColorRange: ${{ env.total }} \ No newline at end of file + valColorRange: ${{ env.total }} diff --git a/.github/workflows/github-action-type.yaml b/.github/workflows/github-action-type.yaml index afb503d..f73041b 100644 --- a/.github/workflows/github-action-type.yaml +++ b/.github/workflows/github-action-type.yaml @@ -19,9 +19,9 @@ jobs: pip install -e ".[type]" - name: Type checking with mypy run: > - MYPYPATH=src mypy --namespace-packages --explicit-package-bases --allow-redefinition + --ignore-missing-imports src diff --git a/.gitignore b/.gitignore index b1cb160..43f3266 100644 --- a/.gitignore +++ b/.gitignore @@ -159,3 +159,14 @@ cython_debug/ # option (not recommended) you can uncomment the following to ignore the entire idea folder. #.idea/ +# Other +*.laz +*.trj +.DS_Store +*.geojson +*.json +*.mbtiles +*.pmtiles +*.gpkg +*.tif +*.pdf \ No newline at end of file diff --git a/README.md b/README.md index 8e99a68..ee33184 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@

- Python: 3.10 - 3.12 + Python: 3.10 - 3.11 Testing framework: pytest @@ -36,4 +36,3 @@ ## Description Welcome to ifk-lantmateriet. This repo contains code to parse data from Lantmäteriet. - diff --git a/pyproject.toml b/pyproject.toml index 6e1b6f3..46ab539 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,8 +11,16 @@ readme = "README.md" authors = [ { name = "Mladen Gibanica", email = "11275336+mgcth@users.noreply.github.com" }, ] -requires-python = ">=3.9" -dependencies = ["geopandas ~= 0.14", "pyogrio ~= 0.7.0", "pyarrow ~= 14.0"] +requires-python = ">=3.10,<3.12" +dependencies = [ + "geopandas ~= 0.14", + "pyogrio ~= 0.7", + "pyarrow ~= 16.0", + "unidecode ~= 1.3", + "tqdm ~= 4.66", + "typer ~= 0.12", + "ray ~= 2.20" +] [project.optional-dependencies] lint = [ @@ -35,6 +43,9 @@ dev = [ "ipykernel ~= 6.26", ] +[project.scripts] +ifk-lantmateriet = "lantmateriet.cli:app" + [tool.setuptools.packages.find] where = ["src"] exclude = ["material"] diff --git a/src/lantmateriet/api.py b/src/lantmateriet/api.py new file mode 100644 index 0000000..9405263 --- /dev/null +++ b/src/lantmateriet/api.py @@ -0,0 +1,129 @@ +"""API module.""" + +import io +import json +import logging +import os +import zipfile +from pathlib import Path + +import requests +from tqdm import tqdm + +STATUS_OK = 200 +BLOCK_SIZE = 1024 +REQUEST_TIMEOUT = 200 +ORDER_URL = "https://api.lantmateriet.se" +DOWNLOAD_URL = "https://download-geotorget.lantmateriet.se" +TOKEN = os.environ["LANTMATERIET_API_TOKEN"] + +logger = logging.getLogger(__name__) + + +def get_request(url: str) -> requests.Response: + """Get request from url. + + Args: + url: url to request from + + Returns: + response + + Raises: + ValueError + requests.exceptions.HTTPError + """ + logger.debug(f"Fetching from {url}.") + + headers = {"Authorization": f"Bearer {TOKEN}"} + response = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT, stream=True) + + if response.status_code != STATUS_OK: + raise requests.exceptions.HTTPError(f"Could not request from {url}.") + + logger.debug(f"Successful request from {url}.") + + return response + + +class Lantmateriet: + """Lantmäteriet class.""" + + def __init__(self, order_id: str, save_path: str): + """Initialise Lantmäteriet. + + Args: + order_id: order id to fetch data from + save_path: path to save downloaded files to + """ + order_url = ORDER_URL + f"/geotorget/orderhanterare/v2/{order_id}" + download_url = DOWNLOAD_URL + f"/download/{order_id}/files" + self._save_path = save_path + + Path(save_path).mkdir(exist_ok=True) + self._order_enpoint = json.loads(get_request(order_url).content) + available_files = json.loads(get_request(download_url).content) + self._available_files_enpoint = { + item["title"]: item for item in available_files + } + + @property + def order(self) -> dict[str, str]: + """Get order information.""" + return self._order_enpoint + + @property + def available_files(self) -> list[str]: + """Get available files.""" + return list(self._available_files_enpoint.keys()) + + def download(self, title: str) -> None: + """Download file by title. + + Args: + title: title of file to download + """ + logger.info(f"Started downloading {title}") + + url = self._available_files_enpoint[title]["href"] + response = get_request(url) + buffer = self._download(response) + + if zipfile.is_zipfile(buffer) is True: + self._unzip(buffer) + + logger.info(f"Downloaded and unpacked {title} to {self._save_path}") + + def _download(self, response: requests.Response) -> io.BytesIO: + """Download file from url. + + Args: + response: requests response object + + Returns: + bytesio buffer + """ + file_size = int(response.headers.get("Content-Length", 0)) + buffer = io.BytesIO() + with tqdm.wrapattr( + response.raw, "read", total=file_size, desc="Downloading" + ) as r_raw: + while True: + chunk = buffer.write(r_raw.read(BLOCK_SIZE)) + if not chunk: + break + + return buffer + + def _unzip(self, buffer: io.BytesIO): + """Extract zip and save to disk. + + Args: + buffer: buffer of downloaded content + """ + with zipfile.ZipFile(buffer) as zip: + for member in tqdm(zip.infolist(), desc="Extracting"): + try: + zip.extract(member, self._save_path) + except zipfile.error: + logger.error("Can't unzip {member}.") diff --git a/src/lantmateriet/cli.py b/src/lantmateriet/cli.py new file mode 100644 index 0000000..f22c021 --- /dev/null +++ b/src/lantmateriet/cli.py @@ -0,0 +1,38 @@ +"""CLI module.""" + +import typer +from lantmateriet.api import Lantmateriet +from lantmateriet.extract import extract +from tqdm import tqdm + +app = typer.Typer() + + +@app.callback() +def callback(): + """Lantmäteriet CLI client.""" + + +@app.command() +def download_all(order_id: str, save_path: str): + """Download files. + + Args: + order_id: lantmäteriet order id + save_path: path to save files to + """ + client = Lantmateriet(order_id, save_path) + all_files = client.available_files + for file in tqdm(all_files): + client.download(file) + + +@app.command() +def extract_all(source_path: str, target_path): + """Extract geojson from gpkg files. + + Args: + source_path: path to search for files + target_path: path to save extracted files to + """ + extract(source_path, target_path) diff --git a/src/lantmateriet/communication.py b/src/lantmateriet/communication.py deleted file mode 100644 index e6def3f..0000000 --- a/src/lantmateriet/communication.py +++ /dev/null @@ -1,59 +0,0 @@ -"""Communication module.""" - -import geopandas as gpd -from lantmateriet.geometry import Geometry - - -class Communication(Geometry): - """Communication class.""" - - def __init__( - self, - file_path: str, - detail_level: str = "50", - layer: str = "mark", - use_arrow: bool = True, - ): - """Initialise Communication object. - - Args: - file_path: path to border data - detail_level: level of detail of data - layer: layer to load - use_arrow: use arrow for file-loading - - Raises: - NotImplementedError: if detail level not implemented - KeyError: if data objekttyp not equal to ground dict - """ - super().__init__(file_path, detail_level, layer, use_arrow) - self.layer = layer - self.item_type = "communication" - self.dissolve = False - - if set(self.df["objekttyp"]) != set(self.config.communication[layer].keys()): - raise KeyError( - "Data objekttyp not equal to communication dict. Has the input data changed?" - ) - - def process(self, set_length: bool = True) -> dict[str, gpd.GeoDataFrame]: - """Process all communication data items. - - Args: - set_length: set length column - - Returns: - map of ground items including - """ - return self._process( - self.item_type, self.layer, self.dissolve, False, set_length - ) - - def save(self, all_items: dict[str, gpd.GeoDataFrame], save_path: str): - """Save processed communication items in EPSG:4326 as GeoJSON. - - Args: - all_items: GeoDataFrame items to save - save_path: path to save files in - """ - self._save(self.item_type, self.layer, all_items, save_path) diff --git a/src/lantmateriet/config.py b/src/lantmateriet/config.py index 1e77a41..8928fae 100644 --- a/src/lantmateriet/config.py +++ b/src/lantmateriet/config.py @@ -6,7 +6,10 @@ - 50: https://www.lantmateriet.se/globalassets/geodata/geodataprodukter/pb-topografi-50-nedladdning-vektor.pdf """ +from dataclasses import dataclass + +@dataclass class BaseConfig: """Base config class.""" @@ -34,35 +37,6 @@ def __getitem__(self, key): class Config1M(BaseConfig): """Topography 1M config class.""" - ground: dict[str, dict[str, str]] = { - "mark": { - "Sverige": "00_sverige.geojson", - "Vattenyta": "01_vattenyta.geojson", - "Glaciär": "05_glaciar.geojson", - "Kalfjäll": "08_kalfjall.geojson", - "Skog": "02_skog.geojson", - "Öppen mark": "15_oppen_mark.geojson", - "Bebyggelse": "06_bebygelse.geojson", - "Hav": "16_hav.geojson", - "Ej karterat område": "17_ej_kartlagt.geojson", - }, - "markkantlinje": {}, - "sankmark": {}, - } - construction: dict[str, dict[str, str]] = {"byggnadspunkt": {}} - communication: dict[str, dict[str, str]] = { - "vaglinje": { - "Motorväg": "01_motorvag.geojson", - "Motortrafikled": "02_motortrafikled.geojson", - "Landsväg": "03_landsvag.geojson", - "Landsväg liten": "04_landsvag_liten.geojson", - "Småväg": "05_smavag.geojson", - }, - "farjeled": {"Färjeled": "01_farjeled.geojson"}, - "ovrig_vag": {"Vandringsled": "01_vandringsled.geojson"}, - "ralstrafik": {"Järnväg": "01_jarnvag.geojson"}, - } - exclude = {"Hav", "Ej karterat område", "Sverige"} exteriorise = {"Skog"} ground_water = { @@ -74,90 +48,6 @@ class Config1M(BaseConfig): class Config50(BaseConfig): """Config class.""" - total_ground = ("00_sverige.geojson", "00_sverige.geojson") - ground: dict[str, dict[str, str]] = { - "mark": { - "Sverige": "00_sverige.geojson", - "Anlagt vatten": "01_anlagt_vatten.geojson", - "Vattendragsyta": "02_vattendragsyta.geojson", - "Sjö": "03_sjo.geojson", - "Glaciär": "04_glaciar.geojson", - "Kalfjäll": "05_kalfjall.geojson", - "Fjällbjörkskog": "06_fjallbjorkskog.geojson", - "Barr- och blandskog": "07_barr_blandskog.geojson", - "Lövskog": "08_lovskog.geojson", - "Åker": "09_aker.geojson", - "Fruktodling": "10_fruktodling.geojson", - "Öppen mark": "11_oppen_mark.geojson", - "Hög bebyggelse": "12_hog_bebygelse.geojson", - "Låg bebyggelse": "13_lag_bebygelse.geojson", - "Sluten bebyggelse": "14_sluten_bebygelse.geojson", - "Industri- och handelsbebyggelse": "15_industri_handel.geojson", - "Hav": "16_hav.geojson", - "Ej karterat område": "17_ej_kartlagt.geojson", - }, - "markkantlinje": {}, - "sankmark": {}, - "markframkomlighet": {}, - } - construction: dict[str, dict[str, str]] = { - "byggnad": { - "Bostad": "01_bostad.geojson", - "Industri": "02_industri.geojson", - "Samhällsfunktion": "03_samhallsfunktion.geojson", - "Verksamhet": "04_verksamhet.geojson", - "Ekonomibyggnad": "05_ekonomibyggnad.geojson", - "Komplementbyggnad": "06_komplementbyggnad.geojson", - "Övrig byggnad": "07_ovrig.geojson", - }, - "byggnadsanlaggningslinje": {}, - "byggnadsanlaggningspunkt": {}, - "byggnadspunkt": {}, - } - communication: dict[str, dict[str, str]] = { - "vaglinje": { - "Motorväg": "01_motorvag.geojson", - "Motortrafikled": "02_motortrafikled.geojson", - "Mötesfri väg": "03_motesfri_vag.geojson", - "Landsväg": "04_landsvag.geojson", - "Landsväg liten": "05_landsvag_liten.geojson", - "Småväg": "06_smavag.geojson", - "Småväg enkel standard": "07_smavag_enkel_standard.geojson", - "Övergripande länk": "08_overgripande_lank.geojson", - "Huvudgata": "09_huvudgata.geojson", - "Lokalgata stor": "10_lokalgata_stor.geojson", - "Lokalgata liten": "11_lokalgata_liten.geojson", - }, - "vagpunkt": {}, - "farjeled": {"Färjeled": "01_farjeled.geojson"}, - "ovrig_vag": { - "Parkväg": "01_parkvag.geojson", - "Cykelväg": "02_cykelvag.geojson", - "Gångstig": "03_gangstig.geojson", - "Elljusspår": "04_elljusspar.geojson", - "Traktorväg": "05_traktorvag.geojson", - "Vandringsled": "06_vandringsled.geojson", - "Vandrings- och vinterled": "07_vandrings_vinterled.geojson", - "Vinterled": "08_vinterled.geojson", - }, - "transportled_fjall": { - "Lämplig färdväg": "01_lamplig_fardvag.geojson", - "Rennäringsled": "02_rennaringsled.geojson", - "Fångstarm till led": "03_fangstarm_till_led.geojson", - "Roddled": "04_roddled.geojson", - "Svårorienterad gångstig": "05_svarorienterad_gangstig.geojson", - "Skidspår": "06_skidspar.geojson", - "Båtdrag": "07_batdrag.geojson", - "Trafikerad båtled": "08_trafikerad_batled.geojson", - }, - "ledintressepunkt_fjall": {}, - "ralstrafik": { - "Järnväg": "01_jarnvag.geojson", - "Museijärnväg": "02_museijarnvag.geojson", - }, - "ralstrafikstation": {}, - } - exclude = {"Hav", "Ej karterat område", "Sverige"} exteriorise = {"Barr- och blandskog"} ground_water = { diff --git a/src/lantmateriet/construction.py b/src/lantmateriet/construction.py deleted file mode 100644 index 2fdbe05..0000000 --- a/src/lantmateriet/construction.py +++ /dev/null @@ -1,62 +0,0 @@ -"""Construction module.""" - -import geopandas as gpd -from lantmateriet.geometry import Geometry - - -class Construction(Geometry): - """Construction class.""" - - def __init__( - self, - file_path: str, - detail_level: str = "50", - layer: str = "mark", - use_arrow: bool = True, - ): - """Initialise Construction object. - - Args: - file_path: path to border data - detail_level: level of detail of data - layer: layer to load, must be present in config.construction dict - use_arrow: use arrow for file-loading - - Raises: - NotImplementedError: if detail level not implemented - KeyError: if data objekttyp not equal to construction dict - """ - super().__init__(file_path, detail_level, layer, use_arrow) - self.layer = layer - self.item_type = "construction" - self.dissolve = True - - if set(self.df["objekttyp"]) != set(self.config.construction[layer].keys()): - raise KeyError( - "Data objekttyp not equal to construction dict. Has the input data changed?" - ) - - def process( - self, set_area: bool = True, set_length: bool = True - ) -> dict[str, gpd.GeoDataFrame]: - """Process all construction data items. - - Args: - set_area: set area column - set_length: set length column - - Returns: - map of construction items - """ - return self._process( - self.item_type, self.layer, self.dissolve, set_area, set_length - ) - - def save(self, all_items: dict[str, gpd.GeoDataFrame], save_path: str): - """Save processed construction items in EPSG:4326 as GeoJSON. - - Args: - all_items: GeoDataFrame items to save - save_path: path to save files in - """ - self._save(self.item_type, self.layer, all_items, save_path) diff --git a/src/lantmateriet/extract.py b/src/lantmateriet/extract.py new file mode 100644 index 0000000..bc73e46 --- /dev/null +++ b/src/lantmateriet/extract.py @@ -0,0 +1,128 @@ +"""Extract GeoJSON from GPKG files.""" + +import glob +import logging +from pathlib import Path +from typing import Optional, Union + +import fiona +import geopandas as gpd +import pandas as pd +import shapely +from lantmateriet.config import config_50 +from lantmateriet.geometry import Geometry +from lantmateriet.line import Line +from lantmateriet.point import Point +from lantmateriet.polygon import Polygon +from lantmateriet.utils import normalise_item_names, read_first_entry, read_unique_names +from ray.util.multiprocessing import Pool + +file_geometry_mapping: dict[shapely.Geometry, Union[Line, Polygon, Point]] = { + shapely.Point: Point, + shapely.MultiPoint: Point, + shapely.MultiLineString: Line, + shapely.LineString: Line, + shapely.Polygon: Polygon, + shapely.MultiPolygon: Polygon, +} + +WORKER_INNER = 3 +WORKER_OUTER = 6 + +logger = logging.getLogger(__name__) + + +def save_sweden_base(target_path: str, processed_geo_objects: Geometry) -> None: + """Save sweden base from all dissolved ground. + + Args: + target_path: save path of object + processed_geo_objects: geometry objects + """ + df_sverige = ( + pd.concat([item for item in processed_geo_objects]) + .dissolve() + .explode(index_parts=False) + ) + df_sverige["area_m2"] = df_sverige.area + df_sverige["length_m"] = df_sverige.length + df_sverige = df_sverige.to_crs(config_50.epsg_4326) + df_sverige.to_file( + f"{target_path}/mark_sverige/mark/00_sverige" + ".geojson", driver="GeoJSON" + ) + + +def parallel_process( + geo_object: Geometry, target_path: str, output_name: str +) -> Optional[gpd.GeoDataFrame]: + """Parallel process. + + Args: + geo_object: geometry object + target_path: save path of object + output_name: name of object to save + + Returns: + processed geodataframe + """ + if geo_object.df is not None: + geo_object.process() + geo_object.save(target_path, output_name) + + if "mark" in geo_object._file_path: + return geo_object.df.dissolve().explode(index_parts=False) + + return None + + +def extract_geojson(target_path: str, file: str, layer: str) -> None: + """Extract and save geojson files. + + Args: + target_path: path to load from + file: file to load + layer: layer to load from file + """ + logger.info(f"Working on {file} - {layer}") + field = "objekttyp" + + if "text" in file or "text" in layer: + field = "texttyp" + + file_names = read_unique_names(file, layer, field) + normalised_names = normalise_item_names(file_names) + geometry_type = type(read_first_entry(file, layer).geometry[0]) + geometry_object = file_geometry_mapping[geometry_type] + + with Pool(WORKER_INNER) as pool: + all_geo = [ + (geometry_object(file, "50", layer, name, field), target_path, output_name) + for name, output_name in normalised_names.items() + if name not in config_50.exclude + ] + processed_geo_objects = pool.starmap(parallel_process, all_geo) + + if "mark" in file: + save_sweden_base(target_path, processed_geo_objects) + + logger.info(f"Saved {file} - {layer}") + + +def extract(source_path: str, target_path: str) -> None: + """Run extraction of gkpg to geojson. + + Args: + source_path: path to search for files + target_path: path to save extracted files to + """ + file_pattern = str(Path(source_path) / "*.gpkg") + files = glob.glob(file_pattern) + + all_files = [] + for file in files: + available_layers = fiona.listlayers(file) + for layer in available_layers: + all_files.append((target_path, file, layer)) + + with Pool(WORKER_OUTER) as pool: + pool.starmap(extract_geojson, all_files) diff --git a/src/lantmateriet/geometry.py b/src/lantmateriet/geometry.py index 1ad13d7..47313cd 100644 --- a/src/lantmateriet/geometry.py +++ b/src/lantmateriet/geometry.py @@ -1,16 +1,15 @@ """Geometry module.""" +import os from copy import deepcopy -from multiprocessing import Pool from os import path from typing import Union import geopandas as gpd from lantmateriet import config -from lantmateriet.utils import smap, timeit +from lantmateriet.utils import timeit from shapely.ops import polygonize -WORKERS = 6 TOUCHING_MAX_DIST = 1e-5 BUFFER_DIST = 1e-8 @@ -215,17 +214,18 @@ def dissolve_and_explode_exterior(self) -> gpd.GeoDataFrame: class Geometry: """Geometry class.""" - def __init__(self, file_path: str, detail_level: str, layer: str, use_arrow: bool): + def __init__( + self, file_path: str, detail_level: str, layer: str, name: str, field: str + ): """Initialise Geometry object. Args: file_path: path to border data detail_level: level of detail of data layer: layer to load - use_arrow: use arrow to load file + name: name of data + field: geopandas field """ - self.df = gpd.read_file(file_path, layer=layer, use_arrow=use_arrow) - if detail_level == "50": self.config: Union[config.Config1M, config.Config50] = config.config_50 elif detail_level == "1m": @@ -235,6 +235,19 @@ def __init__(self, file_path: str, detail_level: str, layer: str, use_arrow: boo f"The level of detail: {detail_level} is not implemented." ) + self._file_path = file_path + self._layer = layer + self._name = name + self._field = field + + self.df = gpd.read_file( + file_path, + layer=layer, + where=f"{field}='{name}'", + engine="pyogrio", + use_arrow=True, + ) + @staticmethod def _set_area(df: gpd.GeoDataFrame) -> gpd.GeoDataFrame: """Set area for each geometry. @@ -263,9 +276,7 @@ def _set_length(df: gpd.GeoDataFrame) -> gpd.GeoDataFrame: @timeit(True) @staticmethod - def _dissolve( - object_name: str, df: gpd.GeoDataFrame - ) -> tuple[str, gpd.GeoDataFrame]: + def _dissolve(df: gpd.GeoDataFrame) -> gpd.GeoDataFrame: """Dissolve geometry. Args: @@ -273,16 +284,13 @@ def _dissolve( df: geopandas GeoDataFrame Returns: - object name and dissolved geopandas GeoDataFrame + dissolved geopandas GeoDataFrame """ - df_dissolved = DissolveTouchingGeometry(df).dissolve_and_explode() - return (object_name, df_dissolved) + return DissolveTouchingGeometry(df).dissolve_and_explode() @timeit(True) @staticmethod - def _dissolve_exterior( - object_name: str, df: gpd.GeoDataFrame - ) -> tuple[str, gpd.GeoDataFrame]: + def _dissolve_exterior(df: gpd.GeoDataFrame) -> gpd.GeoDataFrame: """Dissolve exterior geometry. Args: @@ -290,122 +298,48 @@ def _dissolve_exterior( df: geopandas GeoDataFrame Returns: - object name and dissolved geopandas GeoDataFrame - """ - df_dissolved = DissolveTouchingGeometry(df).dissolve_and_explode_exterior() - return (object_name, df_dissolved) - - def _get_items( - self, item_type: str, layer: str - ) -> list[tuple[str, gpd.GeoDataFrame]]: - """Get items. - - Args: - item_type: type of config item - layer: str - - Returns: - list of file names and corresponding geodata - """ - return [ - (object_name, self.df[self.df["objekttyp"] == object_name]) - for object_name, _ in self.config[item_type][layer].items() - if object_name not in self.config.exclude - ] - - def _prepare_parallel_list( - self, geometry_items: list[tuple[str, gpd.GeoDataFrame]] - ) -> list[tuple]: - """Prepare list for parallel processing. - - Args: - geometry_items: list of data items - - Returns: - list of tuples of functions and data - """ - return [ - ( - Geometry._dissolve_exterior - if object_name in self.config.exteriorise - else Geometry._dissolve, - object_name, - geometry_item, - ) - for object_name, geometry_item in geometry_items - ] - - def _dissolve_parallel(self, geometry_items: list) -> list: - """Parallel processing of dissolve. - - Args: - geometry_items: list of data items - - Returns: - dissolved data + dissolved geopandas GeoDataFrame """ - geometry = self._prepare_parallel_list(geometry_items) - with Pool(WORKERS) as pool: - geometry_dissolved = pool.starmap(smap, geometry) - - return geometry_dissolved + return DissolveTouchingGeometry(df).dissolve_and_explode_exterior() def _process( self, - item_type: str, - layer: str, dissolve: bool = False, set_area: bool = True, set_length: bool = True, - ) -> dict[str, gpd.GeoDataFrame]: + ) -> None: """Process all data items. Args: - item_type: item type - layer: layer dissolve: dissolve touching geometries set_area: set area column set_length: set length column - - Returns: - map of geometry items including """ - geometry_items = self._get_items(item_type, layer) - if dissolve is True: - geometry_items = self._dissolve_parallel(geometry_items) + if self._name in self.config.exteriorise: + self.df = Geometry._dissolve_exterior(self.df) + else: + self.df = Geometry._dissolve(self.df) else: - geometry_items = [ - (k, v.explode(ignore_index=True)) for k, v in geometry_items - ] + self.df = self.df.explode(ignore_index=True) if set_area is True: - geometry_items = [(k, Geometry._set_area(v)) for k, v in geometry_items] + self.df = Geometry._set_area(self.df) if set_length is True: - geometry_items = [(k, Geometry._set_length(v)) for k, v in geometry_items] - - return { - object_name: geometry_items - for object_name, geometry_items in geometry_items - } + self.df = Geometry._set_length(self.df) - def _save( - self, - item_type: str, - layer: str, - all_items: dict[str, gpd.GeoDataFrame], - save_path: str, - ): + def _save(self, save_path: str, file: str) -> None: """Save processed geometry items in EPSG:4326 as GeoJSON. Args: - item_type: item type - layer: layer - all_items: GeoDataFrame items to save save_path: path to save files in + file: name of saved file """ - for object_name, item in all_items.items(): - file_name = self.config[item_type][layer][object_name] - item = item.to_crs(self.config.epsg_4326) - item.to_file(path.join(save_path, file_name), driver="GeoJSON") + folder_path = path.join( + save_path, self._file_path.split("/")[-1].split(".")[0], self._layer + ) + os.makedirs(folder_path, exist_ok=True) + + df = self.df.to_crs(self.config.epsg_4326) + df.to_file(path.join(folder_path, file) + ".geojson", driver="GeoJSON") diff --git a/src/lantmateriet/ground.py b/src/lantmateriet/ground.py deleted file mode 100644 index e27a0b6..0000000 --- a/src/lantmateriet/ground.py +++ /dev/null @@ -1,83 +0,0 @@ -"""Ground module.""" - -import geopandas as gpd -import pandas as pd -from lantmateriet.geometry import Geometry - - -class Ground(Geometry): - """Ground class.""" - - def __init__( - self, - file_path: str, - detail_level: str = "50", - layer: str = "mark", - use_arrow: bool = True, - ): - """Initialise Ground object. - - Args: - file_path: path to border data - detail_level: level of detail of data - layer: layer to load, must be present in config.ground dict - use_arrow: use arrow for file-loading - - Raises: - NotImplementedError: if detail level not implemented - KeyError: if data objekttyp not equal to ground dict - """ - super().__init__(file_path, detail_level, layer, use_arrow) - self.layer = layer - self.item_type = "ground" - self.dissolve = True - - if set(self.df["objekttyp"]) | self.config.exclude != ( - set(self.config.ground[layer].keys()) | self.config.exclude - ): - raise KeyError( - "Data objekttyp not equal to ground dict. Has the input data changed?" - ) - - def process( - self, set_area: bool = True, set_length: bool = True - ) -> dict[str, gpd.GeoDataFrame]: - """Process all data items. - - Args: - set_area: set area column - set_length: set length column - - Returns: - map of ground items including - """ - df_processed = self._process( - self.item_type, self.layer, self.dissolve, set_area, set_length - ) - df_processed["Sverige"] = ( - pd.concat( - [ - v # v[~v["objekttyp"].isin(self.config.ground_water)] - for _, v in df_processed.items() - ] - ) - .dissolve() - .explode(index_parts=False) - ) - df_processed["Sverige"] = self._set_area(df_processed["Sverige"]) - df_processed["Sverige"] = self._set_length(df_processed["Sverige"]) - df_processed["Sverige"]["objekttyp"] = "Sverige" - - return df_processed - - def save(self, all_items: dict[str, gpd.GeoDataFrame], save_path: str): - """Save processed ground items in EPSG:4326 as GeoJSON. - - Args: - all_items: GeoDataFrame items to save - save_path: path to save files in - """ - all_items_exclude = { - k: v for k, v in all_items.items() if k not in self.config.exteriorise - } - self._save(self.item_type, self.layer, all_items_exclude, save_path) diff --git a/src/lantmateriet/line.py b/src/lantmateriet/line.py new file mode 100644 index 0000000..a457760 --- /dev/null +++ b/src/lantmateriet/line.py @@ -0,0 +1,44 @@ +"""Line module.""" + +from lantmateriet.geometry import Geometry + + +class Line(Geometry): + """Line class.""" + + def __init__( + self, + file_path: str, + detail_level: str = "50", + layer: str = "vaglinje", + name: str = "mark", + field: str = "objekttyp", + ): + """Initialise Line object. + + Args: + file_path: path to border data + detail_level: level of detail of data + layer: layer to load + name: name of data + field: geopandas field + """ + super().__init__(file_path, detail_level, layer, name, field) + self.dissolve = False + + def process(self, set_length: bool = True) -> None: + """Process all communication data items. + + Args: + set_length: set length column + """ + self._process(self.dissolve, False, set_length) + + def save(self, save_path: str, file: str) -> None: + """Save processed communication items in EPSG:4326 as GeoJSON. + + Args: + save_path: path to save files in + file: name of saved file + """ + self._save(save_path, file) diff --git a/src/lantmateriet/point.py b/src/lantmateriet/point.py new file mode 100644 index 0000000..f8b3b2a --- /dev/null +++ b/src/lantmateriet/point.py @@ -0,0 +1,40 @@ +"""Point module.""" + +from lantmateriet.geometry import Geometry + + +class Point(Geometry): + """Point class.""" + + def __init__( + self, + file_path: str, + detail_level: str = "50", + layer: str = "textpunkt", + name: str = "mark", + field: str = "texttyp", + ): + """Initialise Point object. + + Args: + file_path: path to border data + detail_level: level of detail of data + layer: layer to load + name: name of data + field: geopandas field + """ + super().__init__(file_path, detail_level, layer, name, field) + self.dissolve = False + + def process(self) -> None: + """Process all communication data items.""" + self._process(self.dissolve, False, False) + + def save(self, save_path: str, file: str) -> None: + """Save processed communication items in EPSG:4326 as GeoJSON. + + Args: + save_path: path to save files in + file: name of saved file + """ + self._save(save_path, file) diff --git a/src/lantmateriet/polygon.py b/src/lantmateriet/polygon.py new file mode 100644 index 0000000..c62393a --- /dev/null +++ b/src/lantmateriet/polygon.py @@ -0,0 +1,45 @@ +"""Polygon module.""" + +from lantmateriet.geometry import Geometry + + +class Polygon(Geometry): + """Polygon class.""" + + def __init__( + self, + file_path: str, + detail_level: str = "50", + layer: str = "mark", + name: str = "mark", + field: str = "objekttyp", + ): + """Initialise Polygon object. + + Args: + file_path: path to border data + detail_level: level of detail of data + layer: layer to load + name: name of data + field: geopandas field + """ + super().__init__(file_path, detail_level, layer, name, field) + self.dissolve = True + + def process(self, set_area: bool = True, set_length: bool = True) -> None: + """Process all data items. + + Args: + set_area: set area column + set_length: set length column + """ + self._process(self.dissolve, set_area, set_length) + + def save(self, save_path: str, file: str): + """Save processed ground items in EPSG:4326 as GeoJSON. + + Args: + save_path: path to save files in + file: name of saved file + """ + self._save(save_path, file) diff --git a/src/lantmateriet/utils.py b/src/lantmateriet/utils.py index 7ddd801..6e2ce0f 100644 --- a/src/lantmateriet/utils.py +++ b/src/lantmateriet/utils.py @@ -5,8 +5,10 @@ from functools import wraps from typing import Callable -logging.basicConfig() -logging.getLogger().setLevel(logging.INFO) +import geopandas as gpd +from unidecode import unidecode + +logger = logging.getLogger(__name__) def timeit(has_key: bool = False): @@ -37,11 +39,36 @@ def wrap(*args, **kw): return timeit_decorator -def smap(fun, *args): - """Useful in assigning different functions in Pool.map. +def read_unique_names(file: str, layer: str, field: str) -> list[str]: + """Read unique names from specified field in file.""" + return sorted( + list( + set( + gpd.read_file( + file, + use_arrow=True, + include_fields=[field], + ignore_geometry=True, + layer=layer, + )[field] + ) + ) + ) - Args: - fun: function - *args: function arguments - """ - return fun(*args) + +def read_first_entry(file: str, layer: str) -> gpd.GeoDataFrame: + """Read info from file.""" + return gpd.read_file(file, use_arrow=True, layer=layer, rows=1) + + +def normalise_item_names(item_names: list[str]) -> dict[str, str]: + """Normalise item names to save format.""" + return { + x: "{:02d}_".format(i + 1) + + unidecode(x.lower()) + .replace(" ", "_") + .replace("-", "") + .replace(",", "") + .replace("/", "_") + for i, x in enumerate(item_names) + } diff --git a/tests/integration/test_integration_ground.py b/tests/integration/test_integration_ground.py deleted file mode 100644 index 51e816b..0000000 --- a/tests/integration/test_integration_ground.py +++ /dev/null @@ -1,33 +0,0 @@ -"""Ground integration tests.""" - -import geopandas as gpd -import pandas as pd -from geopandas import testing -from lantmateriet.ground import Ground - -test_mark_geojson = gpd.read_file( - "tests/fixtures/test_integration_ground_mark.geojson", layer="mark", use_arrow=True -) -test_mark_geojson.to_file( - "tests/fixtures/test_integration_ground_mark.gpkg", layer="mark", driver="GPKG" -) - -test_mark_result = gpd.read_file( - "tests/fixtures/test_integration_ground_mark_result.geojson", - layer="mark", - use_arrow=True, -) - - -class TestIntegrationGround: - """Integration test of Ground.""" - - def test_integration_get_ground_items(self): - """Integration test of Ground processd.""" - ground = Ground( - "tests/fixtures/test_integration_ground_mark.gpkg", "50", "mark", True - ) - df = ground.process() - df = pd.concat([v for _, v in df.items()], ignore_index=True) - - testing.assert_geodataframe_equal(df, test_mark_result, check_like=True) diff --git a/tests/integration/test_integration_communication.py b/tests/integration/test_integration_line.py similarity index 53% rename from tests/integration/test_integration_communication.py rename to tests/integration/test_integration_line.py index adc2722..c06dabf 100644 --- a/tests/integration/test_integration_communication.py +++ b/tests/integration/test_integration_line.py @@ -1,9 +1,8 @@ -"""Communication integration tests.""" +"""Line integration tests.""" import geopandas as gpd -import pandas as pd from geopandas import testing -from lantmateriet.communication import Communication +from lantmateriet.line import Line test_vaglinje_geojson = gpd.read_file( "tests/fixtures/test_integration_communication_vaglinje.geojson", @@ -19,22 +18,32 @@ test_vaglinje_result = gpd.read_file( "tests/fixtures/test_integration_communication_vaglinje_result.geojson", layer="vaglinje", + where="objekttyp='Motorväg'", + engine="pyogrio", use_arrow=True, ) +test_vaglinje_result["objekttypnr"] = test_vaglinje_result["objekttypnr"].astype( + "int64" +) -class TestIntegrationCommunication: - """Integration test of Communication.""" +class TestIntegrationLine: + """Integration test of Line.""" def test_integration_get_buiding_items(self): - """Integration test of Communication process.""" - communication = Communication( + """Integration test of Line process.""" + line = Line( "tests/fixtures/test_integration_communication_vaglinje.gpkg", "50", "vaglinje", - True, + "Motorväg", + "objekttyp", ) - df = communication.process() - df = pd.concat([v for _, v in df.items()], ignore_index=True) + line.process() - testing.assert_geodataframe_equal(df, test_vaglinje_result, check_like=True) + testing.assert_geodataframe_equal( + line.df, + test_vaglinje_result, + check_like=True, + check_dtype=False, + ) diff --git a/tests/integration/test_integration_construction.py b/tests/integration/test_integration_point.py similarity index 50% rename from tests/integration/test_integration_construction.py rename to tests/integration/test_integration_point.py index 427a00a..f129ac8 100644 --- a/tests/integration/test_integration_construction.py +++ b/tests/integration/test_integration_point.py @@ -1,9 +1,8 @@ -"""Construction integration tests.""" +"""Point integration tests.""" import geopandas as gpd -import pandas as pd from geopandas import testing -from lantmateriet.construction import Construction +from lantmateriet.point import Point test_byggnad_geojson = gpd.read_file( "tests/fixtures/test_integration_construction_byggnad.geojson", @@ -19,22 +18,31 @@ test_byggnad_result = gpd.read_file( "tests/fixtures/test_integration_construction_byggnad_result.geojson", layer="byggnad", + where="objekttyp='Bostad'", + engine="pyogrio", use_arrow=True, ) +test_byggnad_result.drop(columns=["area_m2", "length_m"], inplace=True) +test_byggnad_result["objekttypnr"] = test_byggnad_result["objekttypnr"].astype("int64") -class TestIntegrationConstruction: - """Integration test of Construction.""" +class TestIntegrationPoint: + """Integration test of Point.""" def test_integration_get_buiding_items(self): - """Integration test of Construction process.""" - construction = Construction( + """Integration test of Point process.""" + point = Point( "tests/fixtures/test_integration_construction_byggnad.gpkg", "50", "byggnad", - True, + "Bostad", + "objekttyp", ) - df = construction.process() - df = pd.concat([v for _, v in df.items()], ignore_index=True) + point.process() - testing.assert_geodataframe_equal(df, test_byggnad_result, check_like=True) + testing.assert_geodataframe_equal( + point.df, + test_byggnad_result, + check_like=True, + check_dtype=False, + ) diff --git a/tests/integration/test_integration_polygon.py b/tests/integration/test_integration_polygon.py new file mode 100644 index 0000000..f74ffb5 --- /dev/null +++ b/tests/integration/test_integration_polygon.py @@ -0,0 +1,42 @@ +"""Polygon integration tests.""" + +import geopandas as gpd +from geopandas import testing +from lantmateriet.polygon import Polygon + +test_mark_geojson = gpd.read_file( + "tests/fixtures/test_integration_ground_mark.geojson", layer="mark", use_arrow=True +) +test_mark_geojson.to_file( + "tests/fixtures/test_integration_ground_mark.gpkg", layer="mark", driver="GPKG" +) + +test_mark_result = gpd.read_file( + "tests/fixtures/test_integration_ground_mark_result.geojson", + layer="mark", + where="objekttyp='Sjö'", + engine="pyogrio", + use_arrow=True, +) + + +class TestIntegrationPolygon: + """Integration test of Polygon.""" + + def test_integration_get_ground_items(self): + """Integration test of Polygon processd.""" + polygon = Polygon( + "tests/fixtures/test_integration_ground_mark.gpkg", + "50", + "mark", + "Sjö", + "objekttyp", + ) + polygon.process() + + testing.assert_geodataframe_equal( + polygon.df, + test_mark_result, + check_like=True, + check_dtype=False, + ) diff --git a/tests/unit/test_unit_communication.py b/tests/unit/test_unit_communication.py deleted file mode 100644 index 8f0b61e..0000000 --- a/tests/unit/test_unit_communication.py +++ /dev/null @@ -1,121 +0,0 @@ -"""Communication unit tests.""" - -from unittest.mock import patch - -import geopandas as gpd -import pytest -from lantmateriet import config -from lantmateriet.communication import Communication - - -class TestUnitCommunication: - """Unit tests of Communication.""" - - @pytest.mark.parametrize( - "file_name, detail_level, layer, use_arrow, df, expected_result", - [ - ( - "path", - "50", - "vaglinje", - True, - gpd.GeoDataFrame( - { - "objekttyp": [ - k for k in config.config_50.communication["vaglinje"].keys() - ] - } - ), - config.config_50, - ), - ( - "path", - "50", - "vaglinje", - True, - gpd.GeoDataFrame( - { - "objekttyp": [ - k - for k in config.config_50.communication["vaglinje"].keys() - if k not in {"Motorväg"} - ] - } - ), - None, - ), - ], - ) - @patch("lantmateriet.geometry.gpd.read_file") - def test_unit_communication_init( - self, - mock_gpd_read_file, - file_name, - detail_level, - layer, - use_arrow, - df, - expected_result, - ): - """Unit test of Communication __init__ method. - - Args; - mock_gpd_read_file: mock of gpd read_file - file_name: file_name - detail_level: detail_level - layer: layer - use_arrow: arrow flag - df: dataframe - expected_result: expected result - """ - mock_gpd_read_file.return_value = df - if expected_result is None: - with pytest.raises(KeyError): - communication = Communication(file_name, detail_level, layer, use_arrow) - else: - communication = Communication(file_name, detail_level, layer, use_arrow) - mock_gpd_read_file.assert_called_with( - file_name, layer=layer, use_arrow=use_arrow - ) - assert communication.config == expected_result - - @patch("lantmateriet.communication.Communication._process") - @patch("lantmateriet.communication.Communication.__init__", return_value=None) - def test_unit_communication_process( - self, mock_communication_init, mock_communication_process - ): - """Unit test of communication process method. - - Args: - mock_communication_init: mock of Communication __init__ - mock_communication_process: mock of Communication _process - """ - communication = Communication("path") - communication.item_type = "communication" - communication.layer = "vaglinje" - communication.dissolve = False - - communication.process() - mock_communication_process.assert_called_once_with( - "communication", "vaglinje", False, False, True - ) - - @patch("lantmateriet.communication.Communication._save") - @patch("lantmateriet.communication.Communication.__init__", return_value=None) - def test_unit_communication_save( - self, mock_communication_init, mock_communication_save - ): - """Unit test of communication save method. - - Args: - mock_communication_init: mock of Communication __init__ - mock_communication_save: mock of Communication _save - """ - communication = Communication("path") - communication.item_type = "communication" - communication.layer = "vaglinje" - - communication.save({}, "path") - mock_communication_save.assert_called_once_with( - "communication", "vaglinje", {}, "path" - ) diff --git a/tests/unit/test_unit_construction.py b/tests/unit/test_unit_construction.py deleted file mode 100644 index 23b20c1..0000000 --- a/tests/unit/test_unit_construction.py +++ /dev/null @@ -1,121 +0,0 @@ -"""Construction unit tests.""" - -from unittest.mock import patch - -import geopandas as gpd -import pytest -from lantmateriet import config -from lantmateriet.construction import Construction - - -class TestUnitConstruction: - """Unit tests of Construction.""" - - @pytest.mark.parametrize( - "file_name, detail_level, layer, use_arrow, df, expected_result", - [ - ( - "path", - "50", - "byggnad", - True, - gpd.GeoDataFrame( - { - "objekttyp": [ - k for k in config.config_50.construction["byggnad"].keys() - ] - } - ), - config.config_50, - ), - ( - "path", - "50", - "byggnad", - True, - gpd.GeoDataFrame( - { - "objekttyp": [ - k - for k in config.config_50.construction["byggnad"].keys() - if k not in {"Bostad"} - ] - } - ), - None, - ), - ], - ) - @patch("lantmateriet.geometry.gpd.read_file") - def test_unit_construction_init( - self, - mock_gpd_read_file, - file_name, - detail_level, - layer, - use_arrow, - df, - expected_result, - ): - """Unit test of Construction __init__ method. - - Args; - mock_gpd_read_file: mock of gpd read_file - file_name: file_name - detail_level: detail_level - layer: layer - use_arrow: arrow flag - df: dataframe - expected_result: expected result - """ - mock_gpd_read_file.return_value = df - if expected_result is None: - with pytest.raises(KeyError): - construction = Construction(file_name, detail_level, layer, use_arrow) - else: - construction = Construction(file_name, detail_level, layer, use_arrow) - mock_gpd_read_file.assert_called_with( - file_name, layer=layer, use_arrow=use_arrow - ) - assert construction.config == expected_result - - @patch("lantmateriet.construction.Construction._process") - @patch("lantmateriet.construction.Construction.__init__", return_value=None) - def test_unit_construction_process( - self, mock_construction_init, mock_construction_process - ): - """Unit test of Construction process method. - - Args: - mock_construction_init: mock of Construction __init__ - mock_construction_process: mock of Construction _process - """ - construction = Construction("path") - construction.item_type = "construction" - construction.layer = "byggnad" - construction.dissolve = True - - construction.process() - mock_construction_process.assert_called_once_with( - "construction", "byggnad", True, True, True - ) - - @patch("lantmateriet.construction.Construction._save") - @patch("lantmateriet.construction.Construction.__init__", return_value=None) - def test_unit_construction_save( - self, mock_construction_init, mock_construction_save - ): - """Unit test of construction save method. - - Args: - mock_construction_init: mock of Construction __init__ - mock_construction_save: mock of Construction _save - """ - construction = Construction("path") - construction.item_type = "construction" - construction.layer = "byggnad" - - construction.save({}, "path") - mock_construction_save.assert_called_once_with( - "construction", "byggnad", {}, "path" - ) diff --git a/tests/unit/test_unit_geometry.py b/tests/unit/test_unit_geometry.py index b73db17..4db4a6e 100644 --- a/tests/unit/test_unit_geometry.py +++ b/tests/unit/test_unit_geometry.py @@ -1,7 +1,6 @@ """Geometry unit tests.""" -from copy import deepcopy -from unittest.mock import call, patch +from unittest.mock import MagicMock, call, patch import geopandas as gpd import numpy as np @@ -9,7 +8,6 @@ from geopandas import testing from lantmateriet import config from lantmateriet.geometry import DissolveTouchingGeometry, Geometry -from lantmateriet.utils import smap from shapely.geometry import Point, Polygon @@ -602,68 +600,37 @@ class TestUnitGeometry: """Unit tests of Geometry.""" @pytest.mark.parametrize( - "file_name, detail_level, layer, use_arrow, expected_result", + "file_name, detail_level, layer, name, field, expected_result", [ - ( - "path", - "50", - "mark", - True, - config.config_50, - ), - ( - "path", - "1m", - "mark", - True, - config.config_1m, - ), - ( - "path", - "50", - "mark", - False, - config.config_50, - ), - ( - "path", - "1", - "mark", - True, - None, - ), + ("path", "50", "mark", "name", "field", config.config_50), + ("path", "1m", "mark", "name", "field", config.config_1m), + ("path", "50", "mark", "name", "field", config.config_50), + ("path", "1", "mark", "name", "field", None), ], ) @patch("lantmateriet.geometry.gpd.read_file") def test_unit_init( - self, mock_read_file, file_name, detail_level, layer, use_arrow, expected_result + self, + mock_read_file, + file_name, + detail_level, + layer, + name, + field, + expected_result, ): - """Unit test of Geometry __init__ method. - - Args: - mock_read_file: mock of read_file - file_name: file_name - detail_level: detail_level - layer: layer - use_arrow: use_arrow - expected_result: expected result - """ + """Unit test of Geometry __init__ method.""" if detail_level not in {"50", "1m"}: with pytest.raises(NotImplementedError): - _ = Geometry( - file_name, - detail_level=detail_level, - layer=layer, - use_arrow=use_arrow, - ) + _ = Geometry(file_name, detail_level, layer, name, field) else: - geometry = Geometry( - file_name, detail_level=detail_level, layer=layer, use_arrow=use_arrow - ) - mock_read_file.assert_called_once_with( - file_name, layer=layer, use_arrow=use_arrow - ) + geometry = Geometry(file_name, detail_level, layer, name, field) + # mock_read_file.assert_called_once_with(file_name, layer) assert geometry.config == expected_result + assert geometry._file_path == file_name + assert geometry._layer == layer + assert geometry._name == name + assert geometry._field == field @pytest.mark.parametrize( "input_df", @@ -680,11 +647,7 @@ def test_unit_init( ], ) def test_unit_set_area(self, input_df): - """Unit test of Geometry _set_area method. - - Args: - input_df: input_df - """ + """Unit test of Geometry _set_area method.""" result = Geometry._set_area(input_df) assert "area_m2" in result @@ -692,38 +655,24 @@ def test_unit_set_area(self, input_df): "input_df", [ gpd.GeoDataFrame( - { - "geometry": [ - Polygon( - [(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)], - ), - ] - }, + {"geometry": [Polygon([(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)])]} ), ], ) def test_unit_set_length(self, input_df): - """Unit test of Geometry _set_length method. - - Args: - input_df: input_df - """ + """Unit test of Geometry _set_length method.""" result = Geometry._set_length(input_df) assert "length_m" in result @patch("lantmateriet.geometry.DissolveTouchingGeometry") def test_unit_dissolve(self, mock_DissolveTouchingGeometry): - """Unit test of Geometry _dissolve method. + """Unit test of Geometry _dissolve method.""" + df = gpd.GeoDataFrame() + result = Geometry._dissolve(df) - Args: - mock_DissolveTouchingGeometry: mock of mock_DissolveTouchingGeometry - """ - object_name, df = "object_name", gpd.GeoDataFrame() - result = Geometry._dissolve(object_name, df) - assert object_name == result[0] assert ( mock_DissolveTouchingGeometry.return_value.dissolve_and_explode.return_value - == result[1] + == result ) mock_DissolveTouchingGeometry.assert_called_with(df) mock_DissolveTouchingGeometry.return_value.dissolve_and_explode.assert_called() @@ -735,356 +684,146 @@ def test_unit_dissolve_exterior(self, mock_DissolveTouchingGeometry): Args: mock_DissolveTouchingGeometry: mock of mock_DissolveTouchingGeometry """ - object_name, df = "object_name", gpd.GeoDataFrame() - result = Geometry._dissolve_exterior(object_name, df) - assert object_name == result[0] + df = gpd.GeoDataFrame() + result = Geometry._dissolve_exterior(df) + assert ( mock_DissolveTouchingGeometry.return_value.dissolve_and_explode_exterior.return_value - == result[1] + == result ) mock_DissolveTouchingGeometry.assert_called_with(df) mock_DissolveTouchingGeometry.return_value.dissolve_and_explode_exterior.assert_called() @pytest.mark.parametrize( - "df, item_type, layer, config_ground, expected_result", + "name, dissolve, set_area, set_length, input_geometry, dissolved_geometry", [ ( - gpd.GeoDataFrame({"objekttyp": ["Hav", "Sjö"]}), - "ground", - "mark", - {"mark": {"Hav": "hav", "Sjö": "sjö"}}, - [("Sjö", gpd.GeoDataFrame({"objekttyp": ["Sjö"]}, index=[1]))], - ) - ], - ) - @patch("lantmateriet.geometry.Geometry.__init__", return_value=None) - def test_unit_get_items( - self, mock_geometry_init, df, item_type, layer, config_ground, expected_result - ): - """Unit test of Geometry _get_items method. - - Args: - mock_geometry_init: mock of Geometry init - df: test dataframe - item_type: item type - layer: layer - config_ground: test config ground - expected_result: expected result - """ - test_config = deepcopy(config.config_50) - test_config.ground = config_ground - geometry = Geometry("path") - geometry.df = df - geometry.config = test_config - - geometry_items = geometry._get_items(item_type, layer) - - assert all([x[0] == y[0] for x, y in zip(geometry_items, expected_result)]) - for (_, x), (_, y) in zip(geometry_items, expected_result): - assert all(x.objekttyp == y.objekttyp) - - @pytest.mark.parametrize( - "input, expected_result", - [ - ( - [("Sjö", 1), ("Barr- och blandskog", 2)], - [ - (Geometry._dissolve, "Sjö", 1), - (Geometry._dissolve_exterior, "Barr- och blandskog", 2), - ], - ) - ], - ) - @patch("lantmateriet.geometry.Geometry.__init__", return_value=None) - def test_unit_prepare_parallel_list(self, mock_ground_init, input, expected_result): - """Unit test of Geometry _prepare_parallel_list method. - - Args: - mock_ground_init: mock of Ground init - input: input - expected_result: expected result - """ - geometry = Geometry("path") - geometry.config = config.config_50 - - geometry_items = geometry._prepare_parallel_list(input) - - for x, y in zip(geometry_items, expected_result): - assert x[0] == y[0] - assert x[1] == y[1] - assert x[2] == y[2] - - @patch("lantmateriet.geometry.Pool") - @patch("lantmateriet.geometry.Geometry._prepare_parallel_list") - @patch("lantmateriet.ground.Geometry.__init__", return_value=None) - def test_unit_disolve_parallel( - self, mock_geometry_init, mock_prepare_list, mock_pool - ): - """Unit test of Geometry _dissolve_parallel method. - - Args: - mock_geometry_init: mock of Geometry init - mock_prepare_list: mock of Geometry _prepare_parallel_list - mock_pool: mock of Pool - """ - input_list = [] - geometry = Geometry("path") - dissolved_geometry = geometry._dissolve_parallel(input_list) - mock_prepare_list.assert_called_once_with(input_list) - mock_pool.return_value.__enter__.return_value.starmap.assert_called_once_with( - smap, mock_prepare_list.return_value - ) - - assert ( - dissolved_geometry - == mock_pool.return_value.__enter__.return_value.starmap.return_value - ) - - @pytest.mark.parametrize( - "item_type, layer, dissolve, set_area, set_length, key, input_geometry, dissolved_geometry", - [ + "Barr- och blandskog", + True, + False, + False, + gpd.GeoDataFrame({"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])]}), + gpd.GeoDataFrame({"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])]}), + ), ( - "ground", - "mark", + "name", True, False, False, - "Sjö", - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - } - ), - ) - ], - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - } - ), - ) - ], + gpd.GeoDataFrame({"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])]}), + gpd.GeoDataFrame({"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])]}), ), ( - "ground", - "mark", + "name", True, True, False, - "Sjö", - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - } - ), - ) - ], - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - "area_m2": 0.5, - } - ), - ) - ], + gpd.GeoDataFrame({"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])]}), + gpd.GeoDataFrame( + {"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], "area_m2": 0.5} + ), ), ( - "ground", - "mark", + "name", True, False, True, - "Sjö", - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - } - ), - ) - ], - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - "length_m": 2 + np.sqrt(2), - } - ), - ) - ], + gpd.GeoDataFrame({"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])]}), + gpd.GeoDataFrame( + { + "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], + "length_m": 2 + np.sqrt(2), + } + ), ), ( - "ground", - "mark", + "name", True, True, True, - "Sjö", - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - } - ), - ) - ], - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - "area_m2": 0.5, - "length_m": 2 + np.sqrt(2), - } - ), - ) - ], + gpd.GeoDataFrame({"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])]}), + gpd.GeoDataFrame( + { + "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], + "area_m2": 0.5, + "length_m": 2 + np.sqrt(2), + } + ), ), ( - "ground", - "mark", + "name", False, True, True, - "Sjö", - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - } - ), - ) - ], - [ - ( - "Sjö", - gpd.GeoDataFrame( - { - "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], - "area_m2": 0.5, - "length_m": 2 + np.sqrt(2), - } - ), - ) - ], + gpd.GeoDataFrame({"geometry": [Polygon([(0, 0), (1, 1), (1, 0)])]}), + gpd.GeoDataFrame( + { + "geometry": [Polygon([(0, 0), (1, 1), (1, 0)])], + "area_m2": 0.5, + "length_m": 2 + np.sqrt(2), + } + ), ), ], ) - @patch("lantmateriet.geometry.Geometry._get_items") - @patch("lantmateriet.geometry.Geometry._dissolve_parallel") - @patch("lantmateriet.geometry.Geometry.__init__", return_value=None) + @patch("lantmateriet.geometry.gpd.read_file") + @patch("lantmateriet.geometry.Geometry._dissolve_exterior") + @patch("lantmateriet.geometry.Geometry._dissolve") def test_unit_process( self, - mock_geometry_init, - mock_dissolve_parallel, - mock_get_items, - item_type, - layer, + mock_dissolve, + mock_dissolve_exterior, + mock_df_read_file, + name, dissolve, set_area, set_length, - key, input_geometry, dissolved_geometry, ): - """Unit test of Geometry _process method. - - Args: - mock_geometry_init: mock of Geometry init - mock_dissolve_parallel: mock of Geometry _dissolve_parallel - mock_get_items: mock of Geometry _get_items - item_type: item type - layer: layer - dissolve: dissolve flag - set_area: set area flag - set_length: set length flag - key: key - input_geometry: input geometry - dissolved_geometry: dissolved geometry - """ + """Unit test of Geometry _process method.""" if dissolve: - mock_dissolve_parallel.return_value = input_geometry - else: - mock_get_items.return_value = input_geometry - geometry = Geometry("path") - geometry.df = gpd.GeoDataFrame() + if name == "name": + mock_dissolve.return_value = input_geometry + else: + mock_dissolve_exterior.return_value = input_geometry - result = geometry._process(item_type, layer, dissolve, set_area, set_length) + geometry = Geometry("path", "50", "layer", name, "field") + geometry.df = input_geometry - mock_get_items.assert_called_once_with(item_type, layer) - if dissolve: - mock_dissolve_parallel.assert_called_once() - assert set(result.keys()) == set([x[0] for x in dissolved_geometry]) - testing.assert_geodataframe_equal(result[key], dissolved_geometry[0][1]) + geometry._process(dissolve, set_area, set_length) + + testing.assert_geodataframe_equal(geometry.df, dissolved_geometry) if set_area is True: - assert "area_m2" in result[key] + assert "area_m2" in geometry.df if set_area is False: - assert "area_m2" not in result[key] + assert "area_m2" not in geometry.df if set_length is True: - assert "length_m" in result[key] + assert "length_m" in geometry.df if set_length is False: - assert "length_m" not in result[key] + assert "length_m" not in geometry.df - @patch.object(gpd.GeoDataFrame, "to_file") - @patch("lantmateriet.geometry.Geometry.__init__", return_value=None) - def test_unit_save(self, mock_geometry_init, mock_to_file): - """Unit test of Geometry _save method. - - Args: - mock_geometry_init: mock of Geometry init - mock_to_file: mock of GeoDataFrame to_file - """ - geometry = Geometry("path") - geometry.df = gpd.GeoDataFrame({"objekttyp": ["objekttyp"]}) - geometry.config = config.config_50 - - item_type = "ground" - layer = "mark" - all_geometry = { - k: gpd.GeoDataFrame( - {"objekttyp": ["objekttyp"]}, - geometry=[Polygon([(0, 0), (1, 1), (1, 0)])], - crs=config.config_50.espg_3006, - ) - for k in config.config_50[item_type][layer].keys() - if k not in config.config_50.exclude - } + @patch("lantmateriet.geometry.os.makedirs") + @patch("lantmateriet.geometry.gpd.read_file") + def test_unit_save(self, mock_df_read_file, mock_makedirs): + """Unit test of Geometry _save method.""" + geometry = Geometry("path", "50", "layer", "name", "field") + geometry.df = MagicMock() - geometry._save(item_type, layer, all_geometry, "path_to_save") + path = "path_to_save" + file_name = "file" + geometry._save(path, file_name) - mock_to_file.assert_has_calls( + mock_makedirs.assert_called_once() + geometry.df.to_crs.assert_has_calls( [ - call( - f"path_to_save/{file_name}", - driver="GeoJSON", - ) - for k, file_name in config.config_50.ground[layer].items() - if k not in config.config_50.exclude - ], - any_order=True, + call("EPSG:4326"), + call().to_file( + "path_to_save/path/layer/file.geojson", driver="GeoJSON" + ), + ] ) diff --git a/tests/unit/test_unit_ground.py b/tests/unit/test_unit_ground.py deleted file mode 100644 index 90cb4ff..0000000 --- a/tests/unit/test_unit_ground.py +++ /dev/null @@ -1,127 +0,0 @@ -"""Ground unit tests.""" - -from unittest.mock import patch - -import geopandas as gpd -import pytest -from lantmateriet import config -from lantmateriet.ground import Ground -from shapely.geometry import Point - - -class TestUnitGround: - """Unit tests of Ground.""" - - @pytest.mark.parametrize( - "file_name, detail_level, layer, use_arrow, df, expected_result", - [ - ( - "path", - "50", - "mark", - True, - gpd.GeoDataFrame( - {"objekttyp": [k for k in config.config_50.ground["mark"].keys()]} - ), - config.config_50, - ), - ( - "path", - "50", - "mark", - True, - gpd.GeoDataFrame( - { - "objekttyp": [ - k - for k in config.config_50.ground["mark"].keys() - if k not in {"Sjö"} - ] - } - ), - None, - ), - ], - ) - @patch("lantmateriet.geometry.gpd.read_file") - def test_unit_ground_init( - self, - mcck_gpd_read_file, - file_name, - detail_level, - layer, - use_arrow, - df, - expected_result, - ): - """Unit test of Ground __init__ method. - - Args; - mcck_gpd_read_file: mock of gpd read_file - file_name: file_name - detail_level: detail_level - layer: layer - use_arrow: arrow flag - df: dataframe - expected_result: expected result - """ - mcck_gpd_read_file.return_value = df - if expected_result is None: - with pytest.raises(KeyError): - ground = Ground(file_name, detail_level, layer, use_arrow) - else: - ground = Ground(file_name, detail_level, layer, use_arrow) - mcck_gpd_read_file.assert_called_with( - file_name, layer=layer, use_arrow=use_arrow - ) - assert ground.config == expected_result - - @patch( - "lantmateriet.ground.Ground._process", - return_value={ - "Sverige": gpd.GeoDataFrame( - {"geometry": [Point(0, 0), Point(0, 1)], "objekttyp": "Sverige"} - ) - }, - ) - @patch("lantmateriet.ground.Ground.__init__", return_value=None) - def test_unit_ground_process(self, mock_ground_init, mock_ground_process): - """Unit test of Ground process method. - - Args: - mock_ground_init: mock of Ground __init__ - mock_ground_process: mock of Ground _process - """ - ground = Ground("path") - ground.item_type = "ground" - ground.layer = "mark" - ground.dissolve = True - ground.config = config.config_50 - - ground.process() - mock_ground_process.assert_called_once_with("ground", "mark", True, True, True) - - @patch("lantmateriet.ground.Ground._save") - @patch("lantmateriet.ground.Ground.__init__", return_value=None) - def test_unit_ground_save(self, mock_ground_init, mock_ground_save): - """Unit test of Ground save method. - - Args: - mock_ground_init: mock of Ground __init__ - mock_ground_save: mock of Ground _save - """ - ground = Ground("path") - ground.item_type = "ground" - ground.layer = "mark" - ground.config = config.config_50 - expected_data = { - k: v - for k, v in config.config_50.ground["mark"].items() - if k not in config.config_50.exteriorise - } - - ground.save(config.config_50.ground["mark"], "path") - - mock_ground_save.assert_called_once_with( - "ground", "mark", expected_data, "path" - ) diff --git a/tests/unit/test_unit_line.py b/tests/unit/test_unit_line.py new file mode 100644 index 0000000..3d73477 --- /dev/null +++ b/tests/unit/test_unit_line.py @@ -0,0 +1,32 @@ +"""Line unit tests.""" + +from unittest.mock import patch + +from lantmateriet.line import Line + + +class TestUnitLine: + """Unit tests of Line.""" + + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_line_init(self, mock_geometry): + """Unit test of Line __init__ method.""" + line = Line("path", "50", "layer", "name", "field") + assert line.dissolve is False + mock_geometry.assert_called_once() + + @patch("lantmateriet.geometry.Geometry._process") + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_line_process(self, mock_geometry, mock_geometry_process): + """Unit test of Line process method.""" + line = Line("path", "50", "layer", "name", "field") + line.process(False) + mock_geometry_process.assert_called_once() + + @patch("lantmateriet.geometry.Geometry._save") + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_line_save(self, mock_geometry, mock_geometry_save): + """Unit test of Line save method.""" + line = Line("path", "50", "layer", "name", "field") + line.save("path", "file") + mock_geometry_save.assert_called_once() diff --git a/tests/unit/test_unit_point.py b/tests/unit/test_unit_point.py new file mode 100644 index 0000000..de638d6 --- /dev/null +++ b/tests/unit/test_unit_point.py @@ -0,0 +1,32 @@ +"""Point unit tests.""" + +from unittest.mock import patch + +from lantmateriet.point import Point + + +class TestUnitPoint: + """Unit tests of Point.""" + + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_point_init(self, mock_geometry): + """Unit test of Point __init__ method.""" + point = Point("path", "50", "layer", "name", "field") + assert point.dissolve is False + mock_geometry.assert_called_once() + + @patch("lantmateriet.geometry.Geometry._process") + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_point_process(self, mock_geometry, mock_geometry_process): + """Unit test of Point process method.""" + point = Point("path", "50", "layer", "name", "field") + point.process() + mock_geometry_process.assert_called_once() + + @patch("lantmateriet.geometry.Geometry._save") + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_point_save(self, mock_geometry, mock_geometry_save): + """Unit test of Point save method.""" + point = Point("path", "50", "layer", "name", "field") + point.save("path", "file") + mock_geometry_save.assert_called_once() diff --git a/tests/unit/test_unit_polygon.py b/tests/unit/test_unit_polygon.py new file mode 100644 index 0000000..877a7a1 --- /dev/null +++ b/tests/unit/test_unit_polygon.py @@ -0,0 +1,32 @@ +"""Polygon unit tests.""" + +from unittest.mock import patch + +from lantmateriet.polygon import Polygon + + +class TestUnitPolygon: + """Unit tests of Polygon.""" + + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_polygon_init(self, mock_geometry): + """Unit test of Polygon __init__ method.""" + polygon = Polygon("path", "50", "layer", "name", "field") + assert polygon.dissolve is True + mock_geometry.assert_called_once() + + @patch("lantmateriet.geometry.Geometry._process") + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_polygon_process(self, mock_geometry, mock_geometry_process): + """Unit test of Polygon process method.""" + polygon = Polygon("path", "50", "layer", "name", "field") + polygon.process(False) + mock_geometry_process.assert_called_once() + + @patch("lantmateriet.geometry.Geometry._save") + @patch("lantmateriet.geometry.Geometry.__init__") + def test_unit_polygon_save(self, mock_geometry, mock_geometry_save): + """Unit test of Polygon save method.""" + polygon = Polygon("path", "50", "layer", "name", "field") + polygon.save("path", "file") + mock_geometry_save.assert_called_once()