Mercurial > gemma
diff pkg/imports/fd.go @ 1841:491f5b68da9e
Implemented fairway dimension import.
author | Raimund Renkert <raimund.renkert@intevation.de> |
---|---|
date | Thu, 17 Jan 2019 09:22:17 +0100 |
parents | |
children | b1b0db195cc5 |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/pkg/imports/fd.go Thu Jan 17 09:22:17 2019 +0100 @@ -0,0 +1,287 @@ +// This is Free Software under GNU Affero General Public License v >= 3.0 +// without warranty, see README.md and license for details. +// +// SPDX-License-Identifier: AGPL-3.0-or-later +// License-Filename: LICENSES/AGPL-3.0.txt +// +// Copyright (C) 2018 by via donau +// – Österreichische Wasserstraßen-Gesellschaft mbH +// Software engineering by Intevation GmbH +// +// Author(s): +// * Raimund Renkert <raimund.renkert@intevation.de> + +package imports + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "io" + "time" + + "gemma.intevation.de/gemma/pkg/common" + "gemma.intevation.de/gemma/pkg/wfs" +) + +// FairwayDimension is an import job to import +// the fairway dimensions in form of polygon geometries +// and attribute data from a WFS service. +type FairwayDimension struct { + // URL the GetCapabilities URL of the WFS service. + URL string `json:"url"` + // FeatureType selects the feature type of the WFS service. + FeatureType string `json:"feature-type"` + SortBy string `json:"sort-by"` + LOS int `json:"los"` + MinWidth int `json:"min-width"` + MaxWidth int `json:"max-width"` + Depth int `json:"depth"` + SourceOrganization string `json:"source-organization"` +} + +type fdTime struct{ time.Time } + +func (fdt *fdTime) UnmarshalJSON(data []byte) error { + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + t, err := time.Parse("20060102", s) + if err == nil { + *fdt = fdTime{t} + } + return err +} + +// FDJobKind is the import queue type identifier. +const FDJobKind JobKind = "fd" + +type fdJobCreator struct{} + +func init() { + RegisterJobCreator(FDJobKind, fdJobCreator{}) +} + +func (fdJobCreator) Description() string { return "fairway dimension" } + +func (fdJobCreator) AutoAccept() bool { return true } + +func (fdJobCreator) Create(_ JobKind, data string) (Job, error) { + fd := new(FairwayDimension) + if err := common.FromJSONString(data, fd); err != nil { + return nil, err + } + return fd, nil +} + +func (fdJobCreator) Depends() []string { + return []string{ + "fairway_dimensions", + } +} + +// StageDone is a NOP for fairway dimensions imports. +func (fdJobCreator) StageDone(context.Context, *sql.Tx, int64) error { + return nil +} + +// CleanUp for fairway dimension imports is a NOP. +func (*FairwayDimension) CleanUp() error { return nil } + +type fairwayDimensionProperties struct { + HydroSorDat fdTime `json:"hydro_sordat"` +} + +const ( + deleteFairwayDimensionSQL = ` +WITH resp AS ( + SELECT best_utm(area::geometry) AS t, + ST_Transform(area::geometry, best_utm(area::geometry)) AS a + FROM users.responsibility_areas + WHERE country = users.current_user_country() +) +DELETE FROM waterway.fairway_dimensions +WHERE ST_Covers( + (SELECT a FROM resp), + ST_Transform(area::geometry, (SELECT t FROM resp))) +` + + // The ST_MakeValid (line125) and ST_Buffer (line124) are a workarround to + // avoid errors due to reprojection. + insertFairwayDimensionSQL = ` +WITH resp AS ( + SELECT best_utm(area::geometry) AS t, + ST_Transform(area::geometry, best_utm(area::geometry)) AS a + FROM users.responsibility_areas + WHERE country = users.current_user_country() +) +INSERT INTO waterway.fairway_dimensions (area, level_of_service, min_width, max_width, min_depth, date_info, source_organization) +SELECT ST_Transform(clipped.geom, 4326)::geography, $3, $4, $5, $6, $7, $8 FROM ( + SELECT (ST_Dump( + ST_Intersection( + (SELECT ST_Buffer(a, -0.0001) FROM resp), + ST_MakeValid(ST_Transform( + ST_GeomFromWKB($1, $2::integer), + (SELECT t FROM resp) + )) + ) + )).geom AS geom + ) AS clipped + WHERE clipped.geom IS NOT NULL +` +) + +// Do executes the actual fairway dimension import. +func (fd *FairwayDimension) Do( + ctx context.Context, + importID int64, + conn *sql.Conn, + feedback Feedback, +) (interface{}, error) { + + start := time.Now() + + feedback.Info("Import fairway dimension") + + feedback.Info("Loading capabilities from %s", fd.URL) + caps, err := wfs.GetCapabilities(fd.URL) + if err != nil { + feedback.Error("Loading capabilities failed: %v", err) + return nil, err + } + + ft := caps.FindFeatureType(fd.FeatureType) + if ft == nil { + return nil, fmt.Errorf("Unknown feature type '%s'", fd.FeatureType) + } + + feedback.Info("Found feature type '%s'", fd.FeatureType) + + epsg, err := wfs.CRSToEPSG(ft.DefaultCRS) + if err != nil { + feedback.Error("Unsupported CRS name '%s'", ft.DefaultCRS) + return nil, err + } + + urls, err := wfs.GetFeaturesGET( + caps, fd.FeatureType, "application/json", fd.SortBy) + if err != nil { + feedback.Error("Cannot create GetFeature URLs. %v", err) + return nil, err + } + + tx, err := conn.BeginTx(ctx, nil) + if err != nil { + return nil, err + } + defer tx.Rollback() + + insertStmt, err := tx.PrepareContext(ctx, insertFairwayDimensionSQL) + if err != nil { + return nil, err + } + defer insertStmt.Close() + + // Delete the old features. + if _, err := tx.ExecContext(ctx, deleteFairwayDimensionSQL); err != nil { + return nil, err + } + + var ( + unsupported = stringCounter{} + missingProperties int + badProperties int + features int + ) + + if err := wfs.DownloadURLs(urls, func(r io.Reader) error { + rfc, err := wfs.ParseRawFeatureCollection(r) + if err != nil { + return fmt.Errorf("parsing GetFeature document failed: %v", err) + } + if rfc.CRS != nil { + crsName := rfc.CRS.Properties.Name + if epsg, err = wfs.CRSToEPSG(crsName); err != nil { + feedback.Error("Unsupported CRS: %d", crsName) + return err + } + } + + // No features -> ignore. + if rfc.Features == nil { + return nil + } + + feedback.Info("Using EPSG: %d", epsg) + + for _, feature := range rfc.Features { + if feature.Geometry.Coordinates == nil { + missingProperties++ + continue + } + + var props fairwayDimensionProperties + + if err := json.Unmarshal(*feature.Properties, &props); err != nil { + badProperties++ + continue + } + switch feature.Geometry.Type { + case "Polygon": + var p polygonSlice + if err := json.Unmarshal(*feature.Geometry.Coordinates, &p); err != nil { + return err + } + if _, err := insertStmt.ExecContext( + ctx, + p.asWKB(), + epsg, + fd.LOS, + fd.MinWidth, + fd.MaxWidth, + fd.Depth, + props.HydroSorDat.Time, + fd.SourceOrganization, + ); err != nil { + feedback.Error("error: %s", err) + return err + } + features++ + default: + unsupported[feature.Geometry.Type]++ + } + } + return nil + }); err != nil { + feedback.Error("Downloading features failed: %v", err) + return nil, err + } + + if badProperties > 0 { + feedback.Warn("Bad properties: %d", badProperties) + } + + if missingProperties > 0 { + feedback.Warn("Missing properties: %d", missingProperties) + } + + if len(unsupported) != 0 { + feedback.Warn("Unsupported types found: %s", unsupported) + } + + if features == 0 { + err := errors.New("No features found") + feedback.Error("%v", err) + return nil, err + } + + if err = tx.Commit(); err == nil { + feedback.Info("Storing %d features took %s", + features, time.Since(start)) + } + + return nil, err +}