#!/usr/bin/env python
# coding: utf8
#
# Copyright (c) 2020 Centre National d'Etudes Spatiales (CNES).
#
# This file is part of CARS
# (see https://github.com/CNES/cars).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# pylint: disable=too-many-lines
# attribute-defined-outside-init is disabled so that we can create and use
# attributes however we need, to stick to the "everything is attribute" logic
# introduced in issue#895
# pylint: disable=attribute-defined-outside-init
# pylint: disable=too-many-nested-blocks
"""
CARS default pipeline class file
"""
# Standard imports
from __future__ import print_function
import copy
import json
import os
import shutil
from collections import OrderedDict
import yaml
# CARS imports
from cars.core import cars_logging, inputs, preprocessing, projection
from cars.core.cars_logging import logger
from cars.core.progress.progress import ProgressTree
from cars.core.utils import safe_makedirs
from cars.data_structures import cars_dataset
from cars.orchestrator.cluster import log_wrapper
from cars.orchestrator.cluster.log_wrapper import cars_profile
from cars.pipelines import pipeline_constants as pipeline_cst
from cars.pipelines.filling.filling import FillingPipeline
from cars.pipelines.formatting.formatting import FormattingPipeline
from cars.pipelines.merging.merging import MergingPipeline
from cars.pipelines.parameters import advanced_parameters
from cars.pipelines.parameters import advanced_parameters_constants as adv_cst
from cars.pipelines.parameters import dsm_inputs
from cars.pipelines.parameters import dsm_inputs_constants as dsm_cst
from cars.pipelines.parameters import output_constants as out_cst
from cars.pipelines.parameters import output_parameters, sensor_inputs
from cars.pipelines.parameters import sensor_inputs_constants as sens_cst
from cars.pipelines.parameters.output_constants import AUXILIARY
from cars.pipelines.pipeline import Pipeline
from cars.pipelines.pipeline_constants import (
ADVANCED,
APPLICATIONS,
INPUT,
ORCHESTRATOR,
OUTPUT,
PIPELINE,
)
from cars.pipelines.pipeline_template import PipelineTemplate
from cars.pipelines.subsampling.subsampling import SubsamplingPipeline
from cars.pipelines.surface_modeling.surface_modeling import (
SurfaceModelingPipeline,
)
package_path = os.path.dirname(__file__)
FIRST_RES = "first_resolution"
INTERMEDIATE_RES = "intermediate_resolution"
FINAL_RES = "final_resolution"
MONOCULAR = "monocular"
PIPELINE_CONFS = {
FIRST_RES: os.path.join(
package_path,
"..",
"conf_resolution",
"conf_first_resolution.yaml",
),
INTERMEDIATE_RES: os.path.join(
package_path,
"..",
"conf_resolution",
"conf_intermediate_resolution.yaml",
),
FINAL_RES: os.path.join(
package_path,
"..",
"conf_resolution",
"conf_final_resolution.yaml",
),
}
# pylint: disable=E0401
[docs]
@Pipeline.register(
"default",
)
class DefaultPipeline(PipelineTemplate):
"""
DefaultPipeline
"""
# pylint: disable=too-many-instance-attributes
[docs]
def setup_progress_tracking(self, parent_pipeline_id=None):
"""
Setup progress tracking for the default pipeline.
The default pipeline owns the top-level progress hierarchy for the
sub-pipelines it launches.
:param parent_pipeline_id: Optional parent pipeline ID for nesting
:type parent_pipeline_id: int or None
"""
progress_tree = ProgressTree()
progress_tasks = {}
subsampling_pid = None
if self.pipeline_to_use[pipeline_cst.SUBSAMPLING]:
subsampling_pid = progress_tree.begin_pipeline(
"Subsampling", parent_id=parent_pipeline_id
)
monocular_pid = None
if self.use_monocular:
monocular_pid = progress_tree.begin_pipeline(
"Monocular", parent_id=parent_pipeline_id
)
self.monocular_pid = monocular_pid
sm_pids = {}
sm_tie_points_pids = {}
if self.pipeline_to_use[pipeline_cst.SURFACE_MODELING]:
# one pipeline per resolution directly
for epipolar_res in self.resolutions:
sm_pids[epipolar_res] = progress_tree.begin_pipeline(
f"surface_modeling_res{epipolar_res}",
parent_id=parent_pipeline_id,
)
if self.pipeline_to_use[pipeline_cst.TIE_POINTS]:
sm_tie_points_pids[epipolar_res] = (
progress_tree.begin_pipeline(
"Tie Points",
parent_id=sm_pids[epipolar_res],
)
)
if self.pipeline_to_use[pipeline_cst.MERGING]:
merging_pid = progress_tree.begin_pipeline("Merging")
self.merging_pid = merging_pid
filling_pid = None
if self.pipeline_to_use[pipeline_cst.FILLING]:
filling_pid = progress_tree.begin_pipeline("Filling")
if self.pipeline_to_use[pipeline_cst.FORMATTING]:
formatting_pid = progress_tree.begin_pipeline("Formatting")
progress_tasks["formatting"] = progress_tree.register_task(
formatting_pid,
"Formatting",
weight=100,
)
self.progress_tasks = progress_tasks
self.monocular_pid = monocular_pid
self.subsampling_pid = subsampling_pid
self.sm_pids = sm_pids
self.sm_tie_points_pids = sm_tie_points_pids
self.filling_pid = filling_pid
# Show the progress panel immediately after tree initialization.
ProgressTree().draw()
def __init__(self, conf, config_dir=None): # noqa: C901
"""
Creates pipeline
:param pipeline_name: name of the pipeline.
:type pipeline_name: str
:param cfg: configuration {'matching_cost_method': value}
:type cfg: dictionary
:param config_dir: path to dir containing json or yaml file
:type config_dir: str
"""
self.config_dir = config_dir
# Transform relative path to absolute path
if config_dir is not None:
config_dir = os.path.abspath(config_dir)
# Check global conf
self.check_global_schema(conf)
# init global metadata
self.metadata = {}
self.out_dir = conf[OUTPUT][out_cst.OUT_DIRECTORY]
# Original resolution
self.original_resolution = conf[OUTPUT].get(out_cst.RESOLUTION)
conf[PIPELINE] = self.check_pipeline(conf)
self.pipeline_to_use = conf[PIPELINE]
self.use_monocular = self.use_of_monocular(conf)
# Check input
conf[INPUT] = self.check_inputs(conf, config_dir=config_dir)
# check output
conf[OUTPUT] = self.check_output(conf)
self.intermediate_data_dir = os.path.join(
self.out_dir, "intermediate_data"
)
conf[ORCHESTRATOR] = self.check_orchestrator(
conf.get(ORCHESTRATOR, None)
)
if self.pipeline_to_use[pipeline_cst.SUBSAMPLING]:
self.subsampling_conf = self.construct_subsampling_conf(conf)
conf[pipeline_cst.SUBSAMPLING] = self.check_subsampling(
self.subsampling_conf,
)
# Get epipolar resolutions to use
self.resolutions = conf[pipeline_cst.SUBSAMPLING][ADVANCED][
adv_cst.RESOLUTIONS
]
else:
self.resolutions = [1]
if dsm_cst.DSMS in conf[INPUT]:
self.use_monocular = False
self.optimal_pandora_conf = None
if self.pipeline_to_use[pipeline_cst.SURFACE_MODELING]:
self.optimal_pandora_conf = (
self.compute_optimal_pandora_configuration(conf)
)
if self.use_monocular:
if 1 in self.resolutions:
if self.use_monocular == "auto":
self.use_monocular = True
preprocessing.apply_land_cover_auto_configuration(
conf,
self.optimal_pandora_conf,
use_monocular=bool(self.use_monocular),
)
for pipeline, activated in self.pipeline_to_use.items():
if pipeline in conf and not activated:
logger.warning(
f"You tried to override the {pipeline} pipeline but "
f"didn't specify it in the pipeline section. "
"Therefore, this pipeline will not be used"
)
if pipeline in conf and activated:
# Check pipeline conf format
self.check_pipeline_section(pipeline, conf[pipeline])
if pipeline_cst.SURFACE_MODELING not in conf:
conf[pipeline_cst.SURFACE_MODELING] = {}
used_configurations = {}
self.positions = {}
self.used_conf = {}
self.keep_low_res_dir = True
if dsm_cst.DSMS in conf[INPUT] and len(self.resolutions) != 1:
logger.debug(
"For the use of those pipelines, "
"you have to give only one resolution"
)
# overide epipolar resolutions
# TODO: delete with external dsm pipeline (refactoring)
self.resolutions = [1]
elif (
not self.pipeline_to_use[pipeline_cst.SUBSAMPLING]
and len(self.resolutions) != 1
):
logger.warning(
"As you're not using the subsampling pipeline, "
"the working resolution will be 1"
)
self.resolutions = [1]
if self.pipeline_to_use[pipeline_cst.FILLING]:
self.filling_conf = self.construct_filling_conf(conf)
conf[pipeline_cst.FILLING] = self.check_filling(self.filling_conf)
consistent_filling = False
for app in conf[pipeline_cst.FILLING][APPLICATIONS]:
if "dsm_filling" in app:
consistent_filling = True
for filling_method in conf[INPUT][sens_cst.FILLING]:
if conf[INPUT][sens_cst.FILLING][filling_method]:
consistent_filling = True
if not consistent_filling:
self.pipeline_to_use[pipeline_cst.FILLING] = False
if (
self.pipeline_to_use[pipeline_cst.SURFACE_MODELING]
and self.use_monocular
):
self.monocular_out_dir = os.path.join(
self.intermediate_data_dir, MONOCULAR
)
self.monocular_conf = self.construct_monocular_conf(conf)
conf[pipeline_cst.MONOCULAR] = self.check_monocular(
self.monocular_conf
)
subsampling_used_conf = conf.get(pipeline_cst.SUBSAMPLING, {})
filling_used_conf = conf.get(pipeline_cst.FILLING, {})
monocular_used_conf = conf.get(pipeline_cst.MONOCULAR, {})
if self.pipeline_to_use[pipeline_cst.SURFACE_MODELING]:
for epipolar_resolution_index, epipolar_res in enumerate(
self.resolutions
):
first_res = epipolar_resolution_index == 0
last_res = (
epipolar_resolution_index == len(self.resolutions) - 1
)
intermediate_res = not first_res and not last_res
# set computed bool
self.positions[epipolar_resolution_index] = {
"first_res": first_res,
"intermediate_res": intermediate_res,
"last_res": last_res,
}
current_conf = copy.deepcopy(conf)
current_conf = extract_conf_with_resolution(
current_conf,
epipolar_res,
first_res,
intermediate_res,
last_res,
self.intermediate_data_dir,
self.pipeline_to_use,
)
if not isinstance(epipolar_res, int) or epipolar_res < 0:
raise RuntimeError("The resolution has to be an int > 0")
self.used_conf[epipolar_resolution_index] = current_conf
# Initialize unit pipeline in order to retrieve the
# used configuration
# This pipeline will not be run
_ = current_conf.pop(pipeline_cst.SUBSAMPLING, None)
_ = current_conf.pop(pipeline_cst.FILLING, None)
current_unit_pipeline = SurfaceModelingPipeline(
current_conf,
config_dir=self.config_dir,
)
if last_res and self.pipeline_to_use[pipeline_cst.FILLING]:
has_classif = False
# check classif is available in input before forcing it
for sens in current_unit_pipeline.used_conf[INPUT][
sens_cst.SENSORS
].values():
if (
sens_cst.INPUT_CLASSIFICATION in sens
and sens[sens_cst.INPUT_CLASSIFICATION]
):
has_classif = True
break
if has_classif:
# Force classification saving for filling
if not current_unit_pipeline.used_conf[OUTPUT][
AUXILIARY
][sens_cst.INPUT_CLASSIFICATION]:
current_unit_pipeline.used_conf[OUTPUT][AUXILIARY][
sens_cst.INPUT_CLASSIFICATION
] = True
if not self.filling_conf[OUTPUT][AUXILIARY][
sens_cst.INPUT_CLASSIFICATION
]:
self.filling_conf[OUTPUT][AUXILIARY][
sens_cst.INPUT_CLASSIFICATION
] = True
# Get used_conf
used_configurations[epipolar_res] = (
current_unit_pipeline.used_conf
)
# Generate full used_conf
full_used_conf = merge_used_conf(
used_configurations,
self.resolutions,
os.path.abspath(self.out_dir),
)
else:
self.used_conf = copy.deepcopy(conf)
full_used_conf = self.used_conf
full_used_conf[pipeline_cst.SUBSAMPLING] = subsampling_used_conf
full_used_conf[pipeline_cst.PIPELINE] = conf[PIPELINE]
full_used_conf[pipeline_cst.FILLING] = filling_used_conf
if self.use_monocular:
full_used_conf[pipeline_cst.MONOCULAR] = monocular_used_conf
# Save used_conf
cars_dataset.save_dict(
full_used_conf,
os.path.join(self.out_dir, "global_used_conf.yaml"),
)
[docs]
def check_output(self, conf):
"""
Check the output given
:param conf: configuration of output
:type conf: dict
:return overloader output
:rtype : dict
"""
conf_output, _ = output_parameters.check_output_parameters(
conf[INPUT], conf[OUTPUT]
)
return conf_output
[docs]
def use_of_monocular(self, conf):
"""defined monocular activation mode"""
advanced = conf.get(pipeline_cst.MONOCULAR, {}).get("advanced", {})
use_monocular = advanced.get("activated", "auto")
if pipeline_cst.MONOCULAR in conf[INPUT] or use_monocular in (
True,
"auto",
):
if not monocular_available():
use_monocular = False
logger.warning(
"The monocular plugin is not installed. "
"Continuing without monocular. You can install it using : "
"pip install cars-monocular-plugin"
)
return use_monocular
[docs]
def check_pipeline(self, conf): # noqa: C901
"""
Check the pipeline section
"""
possible_pipeline = [
pipeline_cst.SUBSAMPLING,
pipeline_cst.SURFACE_MODELING,
pipeline_cst.TIE_POINTS,
pipeline_cst.FILLING,
pipeline_cst.MERGING,
pipeline_cst.FORMATTING,
]
dict_pipeline = {}
if PIPELINE not in conf:
if dsm_cst.DSMS in conf[INPUT]:
conf[PIPELINE] = [pipeline_cst.MERGING, pipeline_cst.FORMATTING]
elif sens_cst.SENSORS in conf[INPUT]:
conf[PIPELINE] = [
pipeline_cst.SUBSAMPLING,
pipeline_cst.SURFACE_MODELING,
pipeline_cst.TIE_POINTS,
pipeline_cst.FORMATTING,
]
if isinstance(conf[PIPELINE], str):
if conf[PIPELINE] not in possible_pipeline:
raise RuntimeError("This pipeline does not exist")
dict_pipeline = {conf[PIPELINE]: True}
elif isinstance(conf[PIPELINE], list):
for elem in conf[PIPELINE]:
if elem not in possible_pipeline:
raise RuntimeError(f"The pipeline {elem} does not exist")
dict_pipeline.update({elem: True})
elif isinstance(conf[PIPELINE], dict):
for key, _ in conf[PIPELINE].items():
if key not in possible_pipeline:
raise RuntimeError(f"The pipeline {key} does not exist")
dict_pipeline = copy.deepcopy(conf[PIPELINE])
for key in possible_pipeline:
if key not in dict_pipeline:
dict_pipeline.update({key: False})
if (
dsm_cst.DSMS in conf[INPUT]
and not dict_pipeline[pipeline_cst.MERGING]
):
dict_pipeline[pipeline_cst.MERGING] = True
elif (
dsm_cst.DSMS in conf[INPUT]
and dict_pipeline[pipeline_cst.SURFACE_MODELING]
):
raise RuntimeError(
"You can not use the surface modeling pipeline with dsm inputs"
)
elif (
sens_cst.SENSORS in conf[INPUT]
and dict_pipeline[pipeline_cst.MERGING]
and dsm_cst.DSMS not in conf[INPUT]
):
raise RuntimeError(
"You can not use the merging pipeline with sensors inputs only"
)
activate_filling = False
if pipeline_cst.FILLING in conf[INPUT]:
for filling_method in conf[INPUT][pipeline_cst.FILLING]:
if conf[INPUT][pipeline_cst.FILLING][filling_method]:
activate_filling = True
if pipeline_cst.FILLING in conf and conf[pipeline_cst.FILLING]:
activate_filling = True
if (
sens_cst.LOADERS in conf[INPUT]
and sens_cst.INPUT_CLASSIFICATION in conf[INPUT][sens_cst.LOADERS]
and conf[INPUT][sens_cst.LOADERS][sens_cst.INPUT_CLASSIFICATION]
== "slurp"
):
activate_filling = True
if activate_filling:
dict_pipeline[pipeline_cst.FILLING] = True
if (
pipeline_cst.SURFACE_MODELING in conf[INPUT]
and not dict_pipeline[pipeline_cst.SURFACE_MODELING]
):
dict_pipeline[pipeline_cst.SURFACE_MODELING] = True
if (
pipeline_cst.MERGING in conf[INPUT]
and not dict_pipeline[pipeline_cst.MERGING]
):
dict_pipeline[pipeline_cst.MERGING] = True
if (
pipeline_cst.SUBSAMPLING in conf[INPUT]
and not dict_pipeline[pipeline_cst.SUBSAMPLING]
):
dict_pipeline[pipeline_cst.SUBSAMPLING] = True
if (
pipeline_cst.TIE_POINTS in conf[INPUT]
and not dict_pipeline[pipeline_cst.TIE_POINTS]
):
dict_pipeline[pipeline_cst.TIE_POINTS] = True
return dict_pipeline
[docs]
def check_subsampling(self, conf):
"""
Check the subsampling section
:param conf: configuration of subsampling
:type conf: dict
"""
pipeline = SubsamplingPipeline(conf, config_dir=self.config_dir)
advanced = pipeline.check_advanced(
conf[pipeline_cst.SUBSAMPLING].get(ADVANCED, {}),
conf[INPUT],
)
applications = pipeline.check_applications(
conf[pipeline_cst.SUBSAMPLING].get(APPLICATIONS, {})
)
return {ADVANCED: advanced, APPLICATIONS: applications}
[docs]
def check_filling(self, conf):
"""
Check the filling section
:param conf: configuration of subsampling
:type conf: dict
"""
pipeline = FillingPipeline(
conf,
config_dir=self.config_dir,
pre_check=True,
)
advanced = pipeline.check_advanced(
conf[pipeline_cst.FILLING],
conf[INPUT],
)
applications = pipeline.check_applications(
conf[pipeline_cst.FILLING].get(APPLICATIONS, {})
)
return {ADVANCED: advanced, APPLICATIONS: applications}
[docs]
def check_monocular(self, conf):
"""
Check the monocular section
:param conf: configuration of monocular
:type conf: dict
"""
pipeline = Pipeline(
pipeline_cst.MONOCULAR,
conf,
self.config_dir,
)
return pipeline.used_conf["monocular"]
[docs]
def check_pipeline_section(self, pipeline_name, pipeline_conf):
"""
Check any pipeline section
:param pipeline_name: key name in conf
:type pipeline_name: str
:param pipeline_conf: pipeline configuration
:type pipeline_conf: dict
"""
for key in pipeline_conf:
if key not in [APPLICATIONS, ADVANCED]:
raise KeyError(
"Keys of a pipeline must be 'applications' or 'advanced'"
)
if pipeline_name in (
pipeline_cst.SURFACE_MODELING,
pipeline_cst.TIE_POINTS,
):
int_keys = [int(epi_res) for epi_res in self.resolutions]
string_keys = [str(key) for key in int_keys]
possible_keys = ["all"] + int_keys + string_keys
for section in pipeline_conf:
for key in pipeline_conf[section]:
if key not in possible_keys:
raise KeyError(
"When default pipeline is used, keys of {} pipeline"
"must be in {}".format(
pipeline_name,
string_keys + ["all"],
)
)
[docs]
def cleanup_low_res_dir(self):
"""
Clean low res dir
"""
if os.path.exists(self.intermediate_data_dir) and os.path.isdir(
self.intermediate_data_dir
):
try:
shutil.rmtree(self.intermediate_data_dir)
logger.debug(
f"th directory {self.intermediate_data_dir} "
f" has been cleaned."
)
except Exception as exception:
logger.error(
f"Error while deleting {self.intermediate_data_dir}: "
f"{exception}"
)
else:
logger.debug(
f"The directory {self.intermediate_data_dir} has not "
f"been deleted"
)
[docs]
def construct_merging_conf(self, conf):
"""
Construct the right conf for merging
"""
merging_conf = {}
merging_conf[INPUT] = copy.deepcopy(conf[INPUT])
merging_conf[ORCHESTRATOR] = copy.deepcopy(conf[ORCHESTRATOR])
merging_conf[OUTPUT] = {}
merging_conf[OUTPUT]["directory"] = os.path.join(
self.intermediate_data_dir, pipeline_cst.MERGING
)
merging_conf[OUTPUT][AUXILIARY] = conf[OUTPUT].get(AUXILIARY, {})
merging_conf[pipeline_cst.MERGING] = conf.get(pipeline_cst.MERGING, {})
return merging_conf
[docs]
def construct_subsampling_conf(self, conf):
"""
Construct the right conf for subsampling
"""
subsampling_conf = {}
subsampling_conf[INPUT] = copy.deepcopy(conf[INPUT])
subsampling_conf[ORCHESTRATOR] = copy.deepcopy(conf[ORCHESTRATOR])
subsampling_conf[OUTPUT] = {}
subsampling_conf[OUTPUT]["directory"] = self.intermediate_data_dir
subsampling_conf[pipeline_cst.SUBSAMPLING] = conf.get(
pipeline_cst.SUBSAMPLING, {}
)
return subsampling_conf
[docs]
def construct_filling_conf(self, conf):
"""
Construct the right conf for filling
"""
filling_conf = {}
filling_conf[INPUT] = copy.deepcopy(conf[INPUT])
_ = filling_conf[INPUT].pop(dsm_cst.DSMS, None)
filling_conf[ORCHESTRATOR] = copy.deepcopy(conf[ORCHESTRATOR])
filling_conf[OUTPUT] = copy.deepcopy(conf[OUTPUT])
filling_conf[OUTPUT]["directory"] = self.intermediate_data_dir
filling_conf[pipeline_cst.FILLING] = conf.get(pipeline_cst.FILLING, {})
return filling_conf
[docs]
def construct_monocular_conf(self, conf):
"""
Construct the configuration used to run CARS Monocular.
"""
monocular_conf = {
INPUT: copy.deepcopy(conf[INPUT]),
ORCHESTRATOR: copy.deepcopy(conf[ORCHESTRATOR]),
OUTPUT: {out_cst.OUT_DIRECTORY: self.monocular_out_dir},
pipeline_cst.MONOCULAR: copy.deepcopy(
conf.get(pipeline_cst.MONOCULAR, {})
),
}
monocular_conf[pipeline_cst.MONOCULAR].setdefault(ADVANCED, {})
# always save intermediate data for monocular,
# as it is used for the next steps
monocular_conf[pipeline_cst.MONOCULAR][ADVANCED][
"save_intermediate_data"
] = True
return monocular_conf
[docs]
def compute_intersection_poly(
self, inputs_conf, intermediate_dir, epsg, geometry_plugin
):
"""Compute polygone intersection"""
list_sensor_pairs = sensor_inputs.generate_pairs(inputs_conf)
list_intersection_poly = []
for _, (pair_key, sensor_image_left, sensor_image_right) in enumerate(
list_sensor_pairs
):
out_dir = os.path.join(intermediate_dir, "terrain_bbox", pair_key)
safe_makedirs(out_dir)
# Check that the envelopes intersect one another
logger.debug("Computing images envelopes and their intersection")
geojson1 = os.path.join(out_dir, "left_envelope.geojson")
geojson2 = os.path.join(out_dir, "right_envelope.geojson")
out_envelopes_intersection = os.path.join(
out_dir, "envelopes_intersection.geojson"
)
sensor1 = sensor_image_left[sens_cst.INPUT_IMG]
sensor2 = sensor_image_right[sens_cst.INPUT_IMG]
geomodel1 = sensor_image_left[sens_cst.INPUT_GEO_MODEL]
geomodel2 = sensor_image_right[sens_cst.INPUT_GEO_MODEL]
inter_poly, _ = projection.ground_intersection_envelopes(
sensor1["bands"]["b0"]["path"],
sensor2["bands"]["b0"]["path"],
geomodel1,
geomodel2,
geometry_plugin,
geojson1,
geojson2,
out_envelopes_intersection,
envelope_file_driver="GeoJSON",
intersect_file_driver="GeoJSON",
)
# Retrieve bounding box of the ground intersection of the envelopes
inter_poly, inter_epsg = inputs.read_vector(
out_envelopes_intersection
)
# Project polygon if epsg is different
if epsg is not None:
if epsg != inter_epsg:
inter_poly = projection.polygon_projection(
inter_poly, inter_epsg, epsg
)
list_intersection_poly.append(inter_poly)
return list_intersection_poly
[docs]
def compute_optimal_pandora_configuration(self, conf):
"""Compute the dominant world-classification Pandora preset."""
# Get the surface modeling conf
surface_modeling_conf = conf.get(pipeline_cst.SURFACE_MODELING, {})
# Get the 'all' resolution advanced parameters
conf_advanced_all_res = surface_modeling_conf.get(ADVANCED, {}).get(
"all", {}
)
# Get the last resolution advanced parameters
last_res = self.resolutions[-1]
csfa = surface_modeling_conf.get(ADVANCED, {})
conf_advanced_last_res = csfa.get(last_res, csfa.get(str(last_res), {}))
# Apply last res over all res to get the merged advanced parameters
conf_merged = {**conf_advanced_all_res, **conf_advanced_last_res}
(
_,
_,
_,
_,
geom_plugin_with_dem_and_geoid,
_,
_,
_,
) = advanced_parameters.check_advanced_parameters(
conf[INPUT],
conf_merged,
)
epsg = conf[OUTPUT][out_cst.EPSG]
list_intersection_poly = self.compute_intersection_poly(
conf[INPUT],
self.intermediate_data_dir,
epsg,
geom_plugin_with_dem_and_geoid,
)
return preprocessing.find_land_cover_class(
list_intersection_poly,
conf[INPUT][sens_cst.LAND_COVER_MAP],
conf[INPUT][sens_cst.CLASSIFICATION_TO_CONFIGURATION_MAPPING],
epsg,
)
@cars_profile(name="Run_default_pipeline", interval=0.5)
def run(self, args=None): # noqa C901
"""
Run pipeline
"""
loglevel = getattr(args, "loglevel", "INFO").upper()
self.progress_tasks = {}
self.monocular_pid = None
self.subsampling_pid = None
self.sm_pids = {}
self.sm_tie_points_pids = {}
self.filling_pid = None
self.setup_progress_tracking()
previous_out_dir = None
previous_res = None
current_surface_modeling_out_dir = None
updated_conf = {}
updated_conf["pipeline"] = self.pipeline_to_use
if self.pipeline_to_use[pipeline_cst.SUBSAMPLING]:
current_log_dir = os.path.join(self.out_dir, "logs", "subsampling")
cars_logging.setup_logging_pipeline(
loglevel,
out_dir=current_log_dir,
pipeline="subsampling",
)
subsampling_pipeline = SubsamplingPipeline(
self.subsampling_conf, self.config_dir
)
subsampling_pipeline.run(
log_dir=current_log_dir,
parent_pipeline_id=self.subsampling_pid,
)
# Update metadata
self.metadata[pipeline_cst.SUBSAMPLING] = (
subsampling_pipeline.metadata
)
# generate summary
log_wrapper.generate_summary(
current_log_dir,
subsampling_pipeline.used_conf,
pipeline_cst.SUBSAMPLING,
)
updated_conf[pipeline_cst.SUBSAMPLING] = {
ADVANCED: subsampling_pipeline.used_conf[ADVANCED],
APPLICATIONS: subsampling_pipeline.used_conf[APPLICATIONS],
}
if self.use_monocular:
current_log_dir = os.path.join(self.out_dir, "logs", MONOCULAR)
cars_logging.setup_logging_pipeline(
loglevel,
out_dir=current_log_dir,
pipeline=MONOCULAR,
)
monocular_conf = self.construct_monocular_conf(
self.used_conf[len(self.resolutions) - 1]
)
if self.pipeline_to_use[pipeline_cst.SUBSAMPLING]:
final_res = self.resolutions[-1]
if final_res != 1:
monocular_conf[INPUT] = load_subsampling_inputs(
self.intermediate_data_dir,
final_res,
)
monocular_pipeline = Pipeline(
pipeline_cst.MONOCULAR,
monocular_conf,
self.config_dir,
)
monocular_pipeline.run(parent_pipeline_id=self.monocular_pid)
self.monocular_used_conf = monocular_pipeline.used_conf
log_wrapper.generate_summary(
current_log_dir,
monocular_pipeline.used_conf,
pipeline_cst.MONOCULAR,
)
if self.pipeline_to_use[pipeline_cst.SURFACE_MODELING]:
for resolution_index, epipolar_res in enumerate(self.resolutions):
# Get tested unit pipeline
current_conf = self.used_conf[resolution_index]
current_surface_modeling_out_dir = current_conf[OUTPUT][
"directory"
]
# Put right directory for subsampling
if self.pipeline_to_use[pipeline_cst.SUBSAMPLING]:
if epipolar_res != 1:
current_conf[INPUT] = load_subsampling_inputs(
self.intermediate_data_dir, epipolar_res
)
# update directory for unit pipeline
current_conf[OUTPUT][
"directory"
] = current_surface_modeling_out_dir
# get position
first_res, _, last_res = (
self.positions[resolution_index]["first_res"],
self.positions[resolution_index]["intermediate_res"],
self.positions[resolution_index]["last_res"],
)
# setup logging
current_log_dir = os.path.join(
self.out_dir,
"logs",
"surface_modeling_res_" + str(epipolar_res),
)
cars_logging.setup_logging_pipeline(
loglevel,
out_dir=current_log_dir,
pipeline="surface_modeling",
)
logger.info(
"Starting surface modeling pipeline "
f"for resolution 1/{epipolar_res}"
)
# define wich resolution
if first_res and last_res:
which_resolution = "single"
elif first_res:
which_resolution = "first"
elif last_res:
which_resolution = "final"
else:
which_resolution = "intermediate"
# Overide with a priori
if not first_res:
dsm = os.path.join(previous_out_dir, "dsm/dsm.tif")
current_conf[INPUT][sens_cst.LOW_RES_DSM] = dsm
if last_res and self.use_monocular:
add_monocular_inputs(
current_conf[INPUT],
self.monocular_out_dir,
right_image_monocular=(
self.monocular_used_conf[pipeline_cst.MONOCULAR][
ADVANCED
]["right_image_monocular"]
),
)
elif not last_res:
for _, data in current_conf[INPUT][
sens_cst.SENSORS
].items():
data["edges"] = {
"edges_mask": None,
"depth_map": None,
"normals": None,
"tile_id": None,
}
if previous_res is not None:
res_factor = previous_res / epipolar_res
else:
res_factor = None
updated_pipeline = SurfaceModelingPipeline(
current_conf,
config_dir=self.config_dir,
)
updated_pipeline.run(
which_resolution=which_resolution,
working_res=epipolar_res,
res_factor=res_factor,
log_dir=current_log_dir,
previous_out_dir=previous_out_dir,
parent_pipeline_id=self.sm_pids.get(epipolar_res),
tie_points_pipeline_id=self.sm_tie_points_pids.get(
epipolar_res
),
optimal_pandora_configuration=self.optimal_pandora_conf,
)
# Update metadata
if pipeline_cst.SURFACE_MODELING not in self.metadata:
self.metadata[pipeline_cst.SURFACE_MODELING] = {}
self.metadata[pipeline_cst.SURFACE_MODELING][
str(epipolar_res)
] = updated_pipeline.metadata
# update previous out dir
previous_out_dir = current_surface_modeling_out_dir
previous_res = epipolar_res
# generate summary
log_wrapper.generate_summary(
current_log_dir,
updated_pipeline.used_conf,
pipeline_cst.SURFACE_MODELING,
)
updated_conf[epipolar_res] = updated_pipeline.used_conf
# Generate full used_conf
full_used_conf = merge_used_conf(
updated_conf,
self.resolutions,
os.path.abspath(self.out_dir),
)
else:
full_used_conf = self.used_conf
final_conf = None
if self.pipeline_to_use[pipeline_cst.MERGING]:
current_log_dir = os.path.join(self.out_dir, "logs", "merging")
cars_logging.setup_logging_pipeline(
loglevel,
out_dir=current_log_dir,
pipeline="merging",
)
merging_conf = self.construct_merging_conf(self.used_conf)
merging_pipeline = MergingPipeline(merging_conf, self.config_dir)
merging_pipeline.run(parent_pipeline_id=self.merging_pid)
# Update metadata
self.metadata[pipeline_cst.MERGING] = merging_pipeline.metadata
# generate summary
log_wrapper.generate_summary(
current_log_dir,
merging_pipeline.used_conf,
pipeline_cst.MERGING,
)
final_conf = merging_pipeline.used_conf
full_used_conf[pipeline_cst.MERGING] = {
ADVANCED: merging_pipeline.used_conf[ADVANCED],
APPLICATIONS: merging_pipeline.used_conf[APPLICATIONS],
}
if updated_conf and final_conf is None:
last_key = list(updated_conf.keys())[-1]
final_conf = updated_conf[last_key]
elif not updated_conf and final_conf is None:
final_conf = self.used_conf
formatting_input_dir = final_conf[OUTPUT][out_cst.OUT_DIRECTORY]
if self.pipeline_to_use[pipeline_cst.FILLING]:
current_log_dir = os.path.join(self.out_dir, "logs", "filling")
cars_logging.setup_logging_pipeline(
loglevel,
out_dir=current_log_dir,
pipeline="filling",
)
if self.filling_conf[INPUT][pipeline_cst.DSM_TO_FILL] is None:
if (
not self.pipeline_to_use[pipeline_cst.SURFACE_MODELING]
and not self.pipeline_to_use[pipeline_cst.MERGING]
):
raise RuntimeError(
"You have to fill the dsm_to_fill part of the input if "
"you want to use the filling pipeline separately"
)
self.filling_conf[INPUT][pipeline_cst.DSM_TO_FILL] = {}
aux_path = os.path.join(
final_conf[OUTPUT][out_cst.OUT_DIRECTORY], "dsm/"
)
self.filling_conf[INPUT][pipeline_cst.DSM_TO_FILL]["dsm"] = (
os.path.join(aux_path, "dsm.tif")
)
for aux_output, val in final_conf[OUTPUT][
out_cst.AUXILIARY
].items():
if val and aux_output != "filling":
self.filling_conf[INPUT][pipeline_cst.DSM_TO_FILL][
aux_output
] = os.path.join(aux_path, aux_output + ".tif")
self.filling_conf[INPUT][pipeline_cst.DSM_TO_FILL][
"invalidity_mask"
] = os.path.join(aux_path + "invalidity_mask.tif")
dtm_path = os.path.join(aux_path + "dtm.tif")
if os.path.isfile(dtm_path):
self.filling_conf[INPUT][pipeline_cst.INPUT_DTM] = (
os.path.join(aux_path + "dtm.tif")
)
initial_elevation = final_conf[INPUT][
sens_cst.INITIAL_ELEVATION
].get("dem", None)
if (
initial_elevation is not None
and "dem_median" in initial_elevation
):
self.filling_conf[INPUT][sens_cst.INITIAL_ELEVATION] = None
filling_pipeline = FillingPipeline(
self.filling_conf, self.config_dir
)
filling_pipeline.run(
log_dir=current_log_dir,
parent_pipeline_id=self.filling_pid,
)
# Update metadata
self.metadata[pipeline_cst.FILLING] = filling_pipeline.metadata
# generate summary
log_wrapper.generate_summary(
current_log_dir,
filling_pipeline.used_conf,
pipeline_cst.FILLING,
)
formatting_input_dir = os.path.join(
filling_pipeline.used_conf[OUTPUT][out_cst.OUT_DIRECTORY],
pipeline_cst.FILLING,
)
if self.pipeline_to_use[pipeline_cst.FORMATTING]:
ProgressTree().notify(
self.progress_tasks["formatting"],
"started",
total=1,
)
current_log_dir = os.path.join(self.out_dir, "logs", "formatting")
cars_logging.setup_logging_pipeline(
loglevel,
out_dir=current_log_dir,
pipeline="formatting",
)
formatting_conf = self.construct_formatting_conf(
formatting_input_dir
)
formatting_pipeline = FormattingPipeline(
formatting_conf, self.config_dir
)
formatting_pipeline.run(
surface_modeling_dir=current_surface_modeling_out_dir,
log_dir=current_log_dir,
)
# Update metadata
self.metadata[pipeline_cst.FORMATTING] = (
formatting_pipeline.metadata
)
# generate summary
log_wrapper.generate_summary(
current_log_dir,
formatting_pipeline.used_conf,
pipeline_cst.FORMATTING,
)
ProgressTree().notify(
self.progress_tasks["formatting"], "completed"
)
if self.pipeline_to_use[pipeline_cst.FILLING]:
full_used_conf[pipeline_cst.FILLING] = {
ADVANCED: filling_pipeline.used_conf[ADVANCED],
APPLICATIONS: filling_pipeline.used_conf[APPLICATIONS],
}
if self.use_monocular:
full_used_conf[pipeline_cst.MONOCULAR] = self.monocular_used_conf[
pipeline_cst.MONOCULAR
]
else:
full_used_conf[pipeline_cst.MONOCULAR] = {
"advanced": {"activated": False}
}
if isinstance(self.original_resolution, dict):
full_used_conf[OUTPUT][
out_cst.RESOLUTION
] = self.original_resolution
# Save used_conf
cars_dataset.save_dict(
full_used_conf,
os.path.join(self.out_dir, "global_used_conf.yaml"),
)
# Save metadata file
cars_dataset.save_dict(
self.metadata,
os.path.join(self.out_dir, "metadata.yaml"),
)
# Merge profiling in pdf
log_wrapper.generate_pdf_profiling(os.path.join(self.out_dir, "logs"))
# clean outdir
if not self.keep_low_res_dir:
self.cleanup_low_res_dir()
# pylint: disable=too-many-positional-arguments
[docs]
def generate_filling_applications_for_surface_modeling(inputs_conf):
"""
Generate filling applications configuration according to inputs
:param inputs_conf: inputs configuration
:type inputs_conf: dict
"""
filling_applications = {}
# Generate applications configuration
for _, values in inputs_conf[sens_cst.FILLING].items():
# No filling
if values is None:
continue
classif_values = []
for elem in values:
if isinstance(elem, int):
classif_values.append(str(elem))
if not classif_values:
continue
# Update application configuration
new_filling_conf = {
"dense_match_filling": {
"method": "zero_padding",
"fill_classification": classif_values,
}
}
# Update application configuration
filling_applications = overide_pipeline_conf(
filling_applications, new_filling_conf, append_classification=True
)
return filling_applications
[docs]
def monocular_available():
"""
Return True if CARS Monocular is installed within the environment,
False otherwise.
"""
return pipeline_cst.MONOCULAR in Pipeline.available_pipeline
[docs]
def get_monocular_sensor_keys(inputs_conf, right_image_monocular=False):
"""
Return the sensor keys for which monocular outputs are available.
"""
sensor_keys = [left for left, _ in inputs_conf["pairing"]]
if right_image_monocular:
sensor_keys.extend([right for _, right in inputs_conf["pairing"]])
return set(sensor_keys)
[docs]
def overide_pipeline_conf(conf, overiding_conf, append_classification=False):
"""
Merge two dictionaries recursively without removing keys from the base conf.
:param conf: base configuration dictionary
:type conf: dict
:param overiding_conf: overriding configuration dictionary
:type overiding_conf: dict
:return: merged configuration
:rtype: dict
"""
result = copy.deepcopy(conf)
def merge_recursive(base_dict, override_dict):
"""
Main recursive function
"""
for key, value in override_dict.items():
if (
key in base_dict
and isinstance(base_dict[key], dict)
and isinstance(value, dict)
):
merge_recursive(base_dict[key], value)
elif (
append_classification
and key in base_dict
and isinstance(base_dict[key], list)
and isinstance(value, list)
and key == "fill_classification"
):
# extend list, avoiding duplicates
base_dict[key] = list(
OrderedDict.fromkeys(base_dict[key] + value)
)
else:
base_dict[key] = value
merge_recursive(result, overiding_conf)
return result
[docs]
def update_conf(conf, overiding_conf):
"""
update_configuration
:param conf: base configuration dictionary
:type conf: dict
:param overiding_conf: overriding configuration dictionary
:type overiding_conf: dict
:return: merged configuration
:rtype: dict
"""
result = copy.deepcopy(conf)
def merge_recursive(base_dict, override_dict):
"""Main recursive function"""
for key, value in override_dict.items():
if (
key in base_dict
and isinstance(base_dict[key], dict)
and isinstance(value, dict)
):
merge_recursive(base_dict[key], value)
elif key not in base_dict:
base_dict[key] = value
merge_recursive(result, overiding_conf)
return result
[docs]
def merge_used_conf(used_configurations, resolutions, out_dir):
"""
Merge all used configuration
"""
used_configurations = copy.deepcopy(used_configurations)
merged_conf = {
INPUT: used_configurations[resolutions[-1]][INPUT],
OUTPUT: used_configurations[resolutions[-1]][OUTPUT],
ORCHESTRATOR: used_configurations[resolutions[0]][ORCHESTRATOR],
}
merged_conf[INPUT][sens_cst.INITIAL_ELEVATION][sens_cst.DEM_PATH] = None
if pipeline_cst.SUBSAMPLING in used_configurations:
merged_conf[pipeline_cst.SUBSAMPLING] = used_configurations[
pipeline_cst.SUBSAMPLING
]
if "pipeline" in used_configurations:
merged_conf["pipeline"] = used_configurations["pipeline"]
merged_conf[OUTPUT]["directory"] = out_dir
merged_conf[pipeline_cst.TIE_POINTS] = {APPLICATIONS: {}, ADVANCED: {}}
merged_conf[pipeline_cst.SURFACE_MODELING] = {
APPLICATIONS: {},
ADVANCED: {},
}
for resolution in resolutions:
used_conf = used_configurations[resolution]
if pipeline_cst.TIE_POINTS in used_conf:
merged_conf[pipeline_cst.TIE_POINTS][APPLICATIONS][
str(resolution)
] = used_conf[pipeline_cst.TIE_POINTS][APPLICATIONS]
merged_conf[pipeline_cst.TIE_POINTS][ADVANCED][str(resolution)] = (
used_conf[pipeline_cst.TIE_POINTS][ADVANCED]
)
merged_conf[pipeline_cst.SURFACE_MODELING][APPLICATIONS][
str(resolution)
] = used_conf[pipeline_cst.SURFACE_MODELING][APPLICATIONS]
merged_conf[pipeline_cst.SURFACE_MODELING][ADVANCED][
str(resolution)
] = used_conf[pipeline_cst.SURFACE_MODELING][ADVANCED]
return merged_conf