Mercurial > gemma

comparison pkg/imports/st.go @ 1906:32c56e6c089a

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Stretch import: Added forgotten source file.
author Sascha L. Teichmann <sascha.teichmann@intevation.de>
date Fri, 18 Jan 2019 17:13:12 +0100
parents
children 6ffc372cde1e
comparison
equal deleted inserted replaced
1904:931b15be6d7f 1906:32c56e6c089a
1 // This is Free Software under GNU Affero General Public License v >= 3.0
2 // without warranty, see README.md and license for details.
3 //
4 // SPDX-License-Identifier: AGPL-3.0-or-later
5 // License-Filename: LICENSES/AGPL-3.0.txt
6 //
7 // Copyright (C) 2018 by via donau
8 // – Österreichische Wasserstraßen-Gesellschaft mbH
9 // Software engineering by Intevation GmbH
10 //
11 // Author(s):
12 // * Sascha L. Teichmann <sascha.teichmann@intevation.de>
13
14 package imports
15
16 import (
17 "context"
18 "database/sql"
19 "errors"
20 "time"
21
22 "gemma.intevation.de/gemma/pkg/common"
23 "gemma.intevation.de/gemma/pkg/models"
24 )
25
26 type Stretch struct {
27 Name string `json:"name"`
28 From models.Isrs `json:"from"`
29 To models.Isrs `json:"to"`
30 ObjNam string `json:"objnam"`
31 NObjNam *string `json:"nobjnam"`
32 Source string `json:"source-organization"`
33 Date models.Date `json:"date-info"`
34 Countries models.UniqueCountries `json:"countries"`
35 }
36
37 const STJobKind JobKind = "st"
38
39 type stJobCreator struct{}
40
41 func init() {
42 RegisterJobCreator(STJobKind, stJobCreator{})
43 }
44
45 func (stJobCreator) Description() string { return "stretch" }
46
47 func (stJobCreator) AutoAccept() bool { return false }
48
49 func (stJobCreator) Create(_ JobKind, data string) (Job, error) {
50 st := new(Stretch)
51 if err := common.FromJSONString(data, st); err != nil {
52 return nil, err
53 }
54 return st, nil
55 }
56
57 func (stJobCreator) Depends() []string {
58 return []string{
59 "stretches",
60 }
61 }
62
63 const (
64 stDeleteSQL = `
65 DELETE FROM waterway.stretches WHERE
66 staging_done AND name = (
67 SELECT name
68 FROM waterway.stretches WHERE
69 id = (
70 SELECT key from waterway.track_imports
71 WHERE import_id = $1 AND
72 relation = 'waterway.stretches'::regclass)
73 AND NOT staging_done
74 )`
75
76 stStageDoneSQL = `
77 UPDATE waterway.stretches SET staging_done = true
78 WHERE id IN (
79 SELECT key from waterway.track_imports
80 WHERE import_id = $1 AND
81 relation = 'waterway.stretches'::regclass)`
82
83 stInsertSQL = `
84 WITH r AS (
85 SELECT isrsrange(
86 isrs(
87 $1::char(2),
88 $2::char(3),
89 $3::char(5),
90 $4::char(5),
91 $5::int),
92 isrs(
93 $6::char(2),
94 $7::char(3),
95 $8::char(5),
96 $9::char(5),
97 $10::int)) AS r
98 )
99 INSERT INTO waterway.stretches (
100 name,
101 stretch,
102 geom,
103 objnam,
104 nobjnam,
105 date_info
106 ) VALUES (
107 $11,
108 (SELECT r FROM r),
109 ISRSrange_area(
110 (SELECT r FROM r),
111 (SELECT ST_Union(CAST(area AS geometry))
112 FROM waterway.waterway_area)),
113 $12,
114 $13,
115 $14)
116 RETURNING id`
117
118 stInsertCountrySQL = `
119 INSERT INTO waterway.stretch_countries (
120 stretches_id,
121 country_code
122 ) VALUES (
123 $1,
124 $2
125 )`
126 )
127
128 // StageDone moves the imported stretch out of the staging area.
129 func (stJobCreator) StageDone(
130 ctx context.Context,
131 tx *sql.Tx,
132 id int64,
133 ) error {
134 if _, err := tx.ExecContext(ctx, stDeleteSQL, id); err != nil {
135 return err
136 }
137 _, err := tx.ExecContext(ctx, stStageDoneSQL, id)
138 return err
139 }
140
141 // CleanUp of a stretch import is a NOP.
142 func (*Stretch) CleanUp() error { return nil }
143
144 // Do executes the actual bottleneck import.
145 func (st *Stretch) Do(
146 ctx context.Context,
147 importID int64,
148 conn *sql.Conn,
149 feedback Feedback,
150 ) (interface{}, error) {
151
152 start := time.Now()
153
154 feedback.Info("Storing stretch '%s'", st.Name)
155
156 if len(st.Countries) == 0 {
157 return nil, errors.New("List of countries is empty")
158 }
159
160 tx, err := conn.BeginTx(ctx, nil)
161 if err != nil {
162 return nil, err
163 }
164 defer tx.Rollback()
165
166 insertCountryStmt, err := tx.PrepareContext(ctx, stInsertCountrySQL)
167 if err != nil {
168 return nil, err
169 }
170
171 var nobjnm sql.NullString
172 if st.NObjNam != nil {
173 nobjnm = sql.NullString{String: *st.NObjNam, Valid: true}
174 }
175
176 var id int64
177 if err := tx.QueryRowContext(
178 ctx,
179 stInsertSQL,
180 st.From.CountryCode,
181 st.From.LoCode,
182 st.From.FairwaySection,
183 st.From.Orc,
184 st.From.Hectometre,
185 st.To.CountryCode,
186 st.To.LoCode,
187 st.To.FairwaySection,
188 st.To.Orc,
189 st.To.Hectometre,
190 st.Name,
191 st.ObjNam,
192 nobjnm,
193 st.Date.Time,
194 ).Scan(&err); err != nil {
195 return nil, err
196 }
197
198 // store the associated countries.
199 for _, c := range st.Countries {
200 if _, err := insertCountryStmt.ExecContext(ctx, id, c); err != nil {
201 return nil, err
202 }
203 }
204
205 feedback.Info("Storing stretch '%s' took %s", st.Name, time.Since(start))
206 if err := tx.Commit(); err != nil {
207 return nil, err
208 }
209 feedback.Info("Import of stretch was successful")
210
211 summary := struct {
212 Stretch string `json:"stretch"`
213 }{
214 Stretch: st.Name,
215 }
216
217 return &summary, nil
218 }