Coverage for sarvey/config.py: 77%

1#!/usr/bin/env python

3# SARvey - A multitemporal InSAR time series tool for the derivation of displacements.

7# This software was developed together with FERN.Lab (fernlab@gfz-potsdam.de) in the context

8# of the SAR4Infra project with funds of the German Federal Ministry for Digital and

9# Transport and contributions from Landesamt fuer Vermessung und Geoinformation

10# Schleswig-Holstein and Landesbetrieb Strassenbau und Verkehr Schleswig-Holstein.

11#

12# This program is free software: you can redistribute it and/or modify it under

13# the terms of the GNU General Public License as published by the Free Software

14# Foundation, either version 3 of the License, or (at your option) any later

15# version.

16#

17# Important: This package uses PyMaxFlow. The core of PyMaxflows library is the C++

18# implementation by Vladimir Kolmogorov. It is also licensed under the GPL, but it REQUIRES that you

19# cite [BOYKOV04] (see LICENSE) in any resulting publication if you use this code for research purposes.

20# This requirement extends to SARvey.

21#

22# This program is distributed in the hope that it will be useful, but WITHOUT

23# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

24# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more

25# details.

26#

27# You should have received a copy of the GNU Lesser General Public License along

28# with this program. If not, see <https://www.gnu.org/licenses/>.

30"""Configuration module for SARvey."""

31import os

32import json5

33from datetime import date

34from json import JSONDecodeError

35from typing import Optional

36from pydantic import BaseModel, Field, validator, Extra

39class General(BaseModel, extra=Extra.forbid):

40 """Template for settings in config file."""

42 input_path: str = Field(

43 title="The path to the input data directory.",

44 description="Set the path of the input data directory.",

45 default="inputs/"

46 )

48 output_path: str = Field(

49 title="The path to the processing output data directory.",

50 description="Set the path of the processing output data directory.",

51 default="outputs/"

52 )

54 num_cores: int = Field(

55 title="Number of cores",

56 description="Set the number of cores for parallel processing.",

57 default=50

58 )

60 num_patches: int = Field(

61 title="Number of patches",

62 description="Set the number of patches for processing large areas patch-wise.",

63 default=1

64 )

66 apply_temporal_unwrapping: bool = Field(

67 title="Apply temporal unwrapping",

68 description="Apply temporal unwrapping additionally to spatial unwrapping.",

69 default=True

70 )

72 spatial_unwrapping_method: str = Field(

73 title="Spatial unwrapping method",

74 description="Select spatial unwrapping method from 'ilp' and 'puma'.",

75 default='puma'

76 )

78 logging_level: str = Field(

79 title="Logging level.",

80 description="Set loggig level.",

81 default="INFO"

82 )

84 logfile_path: str = Field(

85 title="Logfile Path.",

86 description="Path to directory where the logfiles should be saved.",

87 default="logfiles/"

88 )

90 @validator('input_path')

91 def checkPathInputs(cls, v):

92 """Check if the input path exists."""

93 if v == "":

94 raise ValueError("Empty string is not allowed.")

95 if not os.path.exists(os.path.abspath(v)):

96 raise ValueError(f"input_path is invalid: {os.path.abspath(v)}")

97 if not os.path.exists(os.path.join(os.path.abspath(v), "slcStack.h5")):

98 raise ValueError(f"'slcStack.h5' does not exist: {v}")

99 if not os.path.exists(os.path.join(os.path.abspath(v), "geometryRadar.h5")):

100 raise ValueError(f"'geometryRadar.h5' does not exist: {v}")

101 return v

102

103 @validator('num_cores')

104 def checkNumCores(cls, v):

105 """Check if the number of cores is valid."""

106 if v <= 0:

107 raise ValueError("Number of cores must be greater than zero.")

108 return v

109

110 @validator('num_patches')

111 def checkNumPatches(cls, v):

112 """Check if the number of patches is valid."""

113 if v <= 0:

114 raise ValueError("Number of patches must be greater than zero.")

115 return v

116

117 @validator('spatial_unwrapping_method')

118 def checkUnwMethod(cls, v):

119 """Check if unwrapping_method is valid."""

120 if (v != "ilp") & (v != "puma"):

121 raise ValueError("Unwrapping method must be either 'ilp' or 'puma'.")

122 return v

123

124 @validator('logging_level')

125 def checkLoggingLevel(cls, v):

126 """Check if the logging level is valid."""

127 if v == "":

128 raise ValueError("Empty string is not allowed.")

129 v = v.upper()

130 if v not in ["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "NOTSET"]:

131 raise ValueError("Logging level must be one of ('CRITICAL', 'ERROR', "

132 "'WARNING', 'INFO', 'DEBUG', 'NOTSET').")

133 return v

134

135

136class PhaseLinking(BaseModel, extra=Extra.forbid):

137 """Template for settings in config file."""

138

139 use_phase_linking_results: bool = Field(

140 title="Use phase linking results.",

141 description="Use pixels selected in phase linking.",

142 default=False

143 )

144

145 inverted_path: str = Field(

146 title="The path to the phase linking inverted data directory.",

147 description="Set the path of the inverted data directory.",

148 default="inverted/"

149 )

150

151 num_siblings: int = Field(

152 title="Sibling threshold.",

153 description="Threshold on the number of siblings applied during phase linking to distinguish PS from DS"

154 "candidates.",

155 default=20

156 )

157

158 mask_phase_linking_file: Optional[str] = Field(

159 title="Path to spatial mask file for phase linking results.",

160 description="Path to the mask file, e.g. created by sarvey_mask.",

161 default=""

162 )

163

164 use_ps: bool = Field(

165 title="Use point-like scatterers.",

166 description="Use point-like scatterers (pixels with a low number of siblings) selected in phase linking."

167 "Is applied, only if 'use_phase_linking_results' is true.",

168 default=False

169 )

170

171 mask_ps_file: str = Field(

172 title="The path to the mask file for ps pixels from phase linking.",

173 description="Set the path of the 'maskPS.h5' file (optional).",

174 default="maskPS.h5"

175 )

176

177 @validator('inverted_path')

178 def checkPathInverted(cls, v, values):

179 """Check if the inverted path exists."""

180 if values["use_phase_linking_results"]:

181 if v == "":

182 raise ValueError("Empty string is not allowed.")

183 if not os.path.exists(os.path.abspath(v)):

184 raise ValueError(f"inverted_path is invalid: {os.path.abspath(v)}")

185 if not os.path.exists(os.path.join(os.path.abspath(v), "phase_series.h5")):

186 raise ValueError(f"'phase_series.h5' does not exist: {v}")

187 return v

188

189 @validator('num_siblings')

190 def checkNumSiblings(cls, v, values):

191 """Check is no_siblings is valid."""

192 if not values["use_phase_linking_results"]:

193 if v < 1:

194 raise ValueError("'num_siblings' has to be greater than 0.")

195 return v

196

197 @validator('mask_phase_linking_file')

198 def checkSpatialMaskPath(cls, v, values):

199 """Check if the path is correct."""

200 if values["use_phase_linking_results"]:

201 if v == "" or v is None:

202 return None

203 else:

204 if not os.path.exists(os.path.abspath(v)):

205 raise ValueError(f"mask_phase_linking_file path is invalid: {v}")

206 return v

207

208 @validator('use_ps')

209 def checkUsePS(cls, v, values):

210 """Check if use_ps will be applied."""

211 if (not values["use_phase_linking_results"]) and v:

212 raise ValueError("'use_ps' will not be applied, because 'phase_linking' is set to False.")

213 return v

214

215 @validator('mask_ps_file')

216 def checkPathMaskFilePS(cls, v, values):

217 """Check if the mask file exists."""

218 if values["use_phase_linking_results"] and values["use_ps"]:

219 if v == "":

220 raise ValueError("Empty string is not allowed.")

221 if not os.path.exists(os.path.abspath(v)):

222 raise ValueError(f"mask_ps_file is invalid: {os.path.abspath(v)}")

223 return v

224

225

226class Preparation(BaseModel, extra=Extra.forbid):

227 """Template for settings in config file."""

228

229 start_date: Optional[str] = Field(

230 title="Start date",

231 description="Format: YYYY-MM-DD.",

232 default=None

233 )

234

235 end_date: Optional[str] = Field(

236 title="End date",

237 description="Format: YYYY-MM-DD.",

238 default=None

239 )

240

241 ifg_network_type: str = Field(

242 title="Interferogram network type.",

243 description="Set the intererogram network type: 'sb' (small baseline), 'stb' (small temporal baseline), "

244 "'stb_year' (small temporal baseline and yearly ifgs), 'delaunay' (delaunay network), "

245 "or 'star' (single-reference).",

246 default="sb"

247 )

248

249 num_ifgs: Optional[int] = Field(

250 title="Number of interferograms",

251 description="Set the number of interferograms per image. Might be violated .",

252 default=3

253 )

254

255 max_tbase: Optional[int] = Field(

256 title="Maximum temporal baseline [days]",

257 description="Set the maximum temporal baseline for the ifg network. (required for: 'sb')",

258 default=100

259 )

260

261 filter_window_size: int = Field(

262 title="Size of filtering window [pixel]",

263 description="Set the size of window for lowpass filtering.",

264 default=9

265 )

266

267 @validator('start_date', 'end_date')

268 def checkDates(cls, v):

269 """Check if date format is valid."""

270 if v == "":

271 v = None

272

273 if v is not None:

274 try:

275 date.fromisoformat(v)

276 except Exception as e:

277 raise ValueError(f"Date needs to be in format: YYYY-MM-DD. {e}")

278 return v

279

280 @validator('ifg_network_type')

281 def checkNetworkType(cls, v):

282 """Check if the ifg network type is valid."""

283 if (v != "sb") and (v != "star") and (v != "delaunay") and (v != "stb") and (v != "stb_year"):

284 raise ValueError("Interferogram network type has to be 'sb', 'stb', Ästb_year', 'delaunay' or 'star'.")

285 return v

286

287 @validator('num_ifgs')

288 def checkNumIfgs(cls, v):

289 """Check if the number of ifgs is valid."""

290 if v is not None:

291 if v <= 0:

292 raise ValueError("Number of ifgs must be greater than zero.")

293 return v

294

295 @validator('max_tbase')

296 def checkMaxTBase(cls, v):

297 """Check if the value for maximum time baseline is valid."""

298 if v is not None:

299 if v <= 0:

300 raise ValueError("Maximum baseline must be greater than zero.")

301 return v

302

303 @validator('filter_window_size')

304 def checkFilterWdwSize(cls, v):

305 """Check if the filter window size is valid."""

306 if v <= 0:

307 raise ValueError("Filter window size must be greater than zero.")

308 return v

309

310

311class ConsistencyCheck(BaseModel, extra=Extra.forbid):

312 """Template for settings in config file."""

313

314 coherence_p1: float = Field(

315 title="Temporal coherence threshold for first-order points",

316 description="Set the temporal coherence threshold of first-order points for the consistency check.",

317 default=0.9

318 )

319

320 grid_size: int = Field(

321 title="Grid size [m]",

322 description="Set the grid size in [m] for the consistency check. No grid is applied if 'grid_size' is Zero.",

323 default=200

324 )

325

326 mask_p1_file: Optional[str] = Field(

327 title="Path to mask file for first-order points",

328 description="Set the path to the mask file in .h5 format.",

329 default=""

330 )

331

332 num_nearest_neighbours: int = Field(

333 title="Number of nearest neighbours",

334 description="Set number of nearest neighbours for creating arcs.",

335 default=30

336 )

337

338 max_arc_length: Optional[int] = Field(

339 title="Maximum length of arcs [m]",

340 description="Set the maximum length of arcs.",

341 default=None

342 )

343

344 velocity_bound: float = Field(

345 title="Bounds on mean velocity for temporal unwrapping [m/year]",

346 description="Set the bound (symmetric) for the mean velocity estimation in temporal unwrapping.",

347 default=0.1

348 )

349

350 dem_error_bound: float = Field(

351 title="Bounds on DEM error for temporal unwrapping [m]",

352 description="Set the bound (symmetric) for the DEM error estimation in temporal unwrapping.",

353 default=100.0

354 )

355

356 num_optimization_samples: int = Field(

357 title="Number of samples in the search space for temporal unwrapping",

358 description="Set the number of samples evaluated along the search space for temporal unwrapping.",

359 default=100

360 )

361

362 arc_unwrapping_coherence: float = Field(

363 title="Arc unwrapping coherence threshold",

364 description="Set the arc unwrapping coherence threshold for the consistency check.",

365 default=0.6

366 )

367

368 min_num_arc: int = Field(

369 title="Minimum number of arcs per point",

370 description="Set the minimum number of arcs per point.",

371 default=3

372 )

373

374 @validator('coherence_p1')

375 def checkCoherenceP1(cls, v):

376 """Check if the temporal coherence threshold is valid."""

377 if v < 0:

378 raise ValueError("Temporal coherence threshold cannot be negative.")

379 if v > 1:

380 raise ValueError("Temporal coherence threshold cannot be greater than 1.")

381 return v

382

383 @validator('grid_size')

384 def checkGridSize(cls, v):

385 """Check if the grid size is valid."""

386 if v < 0:

387 raise ValueError('Grid size cannot be negative.')

388 if v == 0:

389 v = None

390 return v

391

392 @validator('mask_p1_file')

393 def checkSpatialMaskPath(cls, v):

394 """Check if the path is correct."""

395 if v == "" or v is None:

396 return None

397 else:

398 if not os.path.exists(os.path.abspath(v)):

399 raise ValueError(f"mask_p1_file path is invalid: {v}")

400 return v

401

402 @validator('num_nearest_neighbours')

403 def checkKNN(cls, v):

404 """Check if the k-nearest neighbours is valid."""

405 if v <= 0:

406 raise ValueError('Number of nearest neighbours cannot be negative or zero.')

407 return v

408

409 @validator('max_arc_length')

410 def checkMaxArcLength(cls, v):

411 """Check if the maximum length of arcs is valid."""

412 if v is None:

413 return 999999

414 if v <= 0:

415 raise ValueError('Maximum arc length must be positive.')

416 return v

417

418 @validator('velocity_bound')

419 def checkVelocityBound(cls, v):

420 """Check if the velocity bound is valid."""

421 if v <= 0:

422 raise ValueError('Velocity bound cannot be negative or zero.')

423 return v

424

425 @validator('dem_error_bound')

426 def checkDEMErrorBound(cls, v):

427 """Check if the DEM error bound is valid."""

428 if v <= 0:

429 raise ValueError('DEM error bound cannot be negative or zero.')

430 return v

431

432 @validator('num_optimization_samples')

433 def checkNumSamples(cls, v):

434 """Check if the number of samples for the search space is valid."""

435 if v <= 0:

436 raise ValueError('Number of optimization samples cannot be negative or zero.')

437 return v

438

439 @validator('arc_unwrapping_coherence')

440 def checkArcCoherence(cls, v):

441 """Check if the arc coherence threshold is valid."""

442 if v < 0:

443 raise ValueError('Arc unwrapping coherence threshold cannot be negativ.')

444 if v > 1:

445 raise ValueError('Arc unwrapping coherence threshold cannot be greater than 1.')

446 return v

447

448 @validator('min_num_arc')

449 def checkMinNumArc(cls, v):

450 """Check if the minimum number of arcs is valid."""

451 if v < 0:

452 raise ValueError('Velocity bound cannot be negative.')

453 return v

454

455

456class Unwrapping(BaseModel, extra=Extra.forbid):

457 """Template for settings in config file."""

458

459 use_arcs_from_temporal_unwrapping: bool = Field(

460 title="Use arcs from temporal unwrapping",

461 description="If true, use same arcs from temporal unwrapping, where bad arcs were already removed."

462 "If false, apply new delaunay triangulation.",

463 default=True

464 )

465

466

467class Filtering(BaseModel, extra=Extra.forbid):

468 """Template for filtering settings in config file."""

469

470 coherence_p2: float = Field(

471 title="Temporal coherence threshold",

472 description="Set the temporal coherence threshold for the filtering step.",

473 default=0.8

474 )

475

476 apply_aps_filtering: bool = Field(

477 title="Apply atmosphere filtering",

478 description="Set whether to filter atmosphere or to skip it.",

479 default=True

480 )

481

482 interpolation_method: str = Field(

483 title="Spatial interpolation method.",

484 description="Method for interpolating atmosphere in space ('linear', 'cubic' or 'kriging').",

485 default="kriging"

486 )

487

488 grid_size: int = Field(

489 title="Grid size [m].",

490 description="Set the grid size for spatial filtering.",

491 default=1000

492 )

493

494 mask_p2_file: Optional[str] = Field(

495 title="Path to spatial mask file for second-order points.",

496 description="Path to the mask file, e.g. created by sarvey_mask.",

497 default=""

498 )

499

500 use_moving_points: bool = Field(

501 title="Use moving points",

502 description="Set whether to use moving points in the filtering step.",

503 default=True

504 )

505

506 max_temporal_autocorrelation: float = Field(

507 title="Max temporal auto correlation.",

508 description="Set temporal autocorrelation threshold for the selection of stable/linearly moving points.",

509 default=0.3

510 )

511

512 @validator('coherence_p2')

513 def checkTempCohThrsh2(cls, v):

514 """Check if the temporal coherence threshold is valid."""

515 if v < 0:

516 raise ValueError("Temporal coherence threshold cannot be negative.")

517 if v > 1:

518 raise ValueError("Temporal coherence threshold cannot be greater than 1.")

519 return v

520

521 @validator('interpolation_method')

522 def checkInterpolationMethod(cls, v):

523 """Check if the interpolation method is valid."""

524 if (v.lower() != "linear") and (v.lower() != "cubic") and (v.lower() != "kriging"):

525 raise ValueError("Method for interpolating atmosphere in space needs to be either 'linear', 'cubic' "

526 "or 'kriging'.")

527 return v

528

529 @validator('grid_size')

530 def checkGridSize(cls, v):

531 """Check if the grid size is valid."""

532 if v < 0:

533 raise ValueError("Grid size cannot be negative.")

534 else:

535 return v

536

537 @validator('mask_p2_file')

538 def checkSpatialMaskPath(cls, v):

539 """Check if the path is correct."""

540 if v == "" or v is None:

541 return None

542 else:

543 if not os.path.exists(os.path.abspath(v)):

544 raise ValueError(f"mask_p2_file path is invalid: {v}")

545 return v

546

547 @validator('max_temporal_autocorrelation')

548 def checkMaxAutoCorr(cls, v):

549 """Check if the value is correct."""

550 if v < 0 or v > 1:

551 raise ValueError(f"max_temporal_autocorrelation has to be between 0 and 1: {v}")

552 return v

553

554

555class Densification(BaseModel, extra=Extra.forbid):

556 """Template for densification settings in config file."""

557

558 num_connections_to_p1: int = Field(

559 title="Number of connections in temporal unwrapping.",

560 description="Set number of connections between second-order point and closest first-order points for temporal "

561 "unwrapping.",

562 default=5

563 )

564

565 max_distance_to_p1: int = Field(

566 title="Maximum distance to nearest first-order point [m]",

567 description="Set threshold on the distance between first-order points and to be temporally unwrapped"

568 "second-order point.",

569 default=2000

570 )

571

572 velocity_bound: float = Field(

573 title="Bounds on mean velocity for temporal unwrapping [m/year]",

574 description="Set the bound (symmetric) for the mean velocity in temporal unwrapping.",

575 default=0.15

576 )

577

578 dem_error_bound: float = Field(

579 title="Bounds on DEM error for temporal unwrapping [m]",

580 description="Set the bound (symmetric) for the DEM error estimation in temporal unwrapping.",

581 default=100.0

582 )

583

584 num_optimization_samples: int = Field(

585 title="Number of samples in the search space for temporal unwrapping",

586 description="Set the number of samples evaluated along the search space for temporal unwrapping.",

587 default=100

588 )

589

590 arc_unwrapping_coherence: float = Field(

591 title="Arc unwrapping coherence threshold for densification",

592 description="Set arc unwrapping coherence threshold for densification.",

593 default=0.5

594 )

595

596 @validator('num_connections_to_p1')

597 def checkNumConn1(cls, v):

598 """Check if num_connections_p1 are valid."""

599 if v <= 0:

600 raise ValueError(f"num_connections_p1 must be greater than 0: {v}")

601 return v

602

603 @validator('max_distance_to_p1')

604 def checkMaxDistanceP1(cls, v):

605 """Check if the maximum distance to nearest first-order points is valid."""

606 if v < 0:

607 raise ValueError('Maximum distance to first-order points cannot be negative.')

608 return v

609

610 @validator('velocity_bound')

611 def checkVelocityBound(cls, v):

612 """Check if the velocity bound is valid."""

613 if v <= 0:

614 raise ValueError('Velocity bound cannot be negative or zero.')

615 return v

616

617 @validator('dem_error_bound')

618 def checkDEMErrorBound(cls, v):

619 """Check if the DEM error bound is valid."""

620 if v <= 0:

621 raise ValueError('DEM error bound cannot be negative or zero.')

622 return v

623

624 @validator('num_optimization_samples')

625 def checkNumSamples(cls, v):

626 """Check if the number of samples for the search space is valid."""

627 if v <= 0:

628 raise ValueError('Number of optimization samples cannot be negative or zero.')

629 return v

630

631 @validator('arc_unwrapping_coherence')

632 def checkCoherenceThresh(cls, v):

633 """Check if arc_unwrapping_coherence is valid."""

634 if v < 0 or v > 1:

635 raise ValueError(f"coherence_threshold is not between 0 and 1: {v}")

636 return v

637

638

639class Config(BaseModel):

640 """Configuration for sarvey."""

641

642 # title has to be the name of the class. Needed for creating default file

643 general: General = Field(

644 title="General", description=""

645 )

646

647 phase_linking: PhaseLinking = Field(

648 title="PhaseLinking", description=""

649 )

650

651 preparation: Preparation = Field(

652 title="Preparation", description=""

653 )

654

655 consistency_check: ConsistencyCheck = Field(

656 title="ConsistencyCheck", description=""

657 )

658

659 unwrapping: Unwrapping = Field(

660 title="Unwrapping", description=""

661 )

662

663 filtering: Filtering = Field(

664 title="Filtering", description=""

665 )

666

667 densification: Densification = Field(

668 title="Densification", description=""

669 )

670

671

672def loadConfiguration(*, path: str) -> dict:

673 """Load configuration json file.

674

675 Parameters

676 ----------

677 path : str

678 Path to the configuration json file.

679

680 Returns

681 -------

682 : Config

683 An object of class Config

684

685 Raises

686 ------

687 JSONDecodeError

688 If failed to parse the json file to the dictionary.

689 FileNotFoundError

690 Config file not found.

691 IOError

692 Invalid JSON file.

693 ValueError

694 Invalid value for configuration object.

695 """

696 try:

697 with open(path) as config_fp:

698 config = json5.load(config_fp)

699 config = Config(**config)

700 except JSONDecodeError as e:

701 raise IOError(f'Failed to load the configuration json file => {e}')

702 return config