Coverage for toardb / data / crud.py: 78%

1# SPDX-FileCopyrightText: 2021 Forschungszentrum Jülich GmbH 

2# SPDX-License-Identifier: MIT 

3 

4""" 

5Create, Read, Update, Delete functionality 

6 

7""" 

8 

9import sys 

10from io import StringIO 

11from contextlib import closing 

12import requests 

13import json 

14from collections import defaultdict 

15import os 

16 

17from sqlalchemy import insert, delete, select, and_, text, func 

18from sqlalchemy.orm import Session 

19from sqlalchemy.engine import Engine 

20from geoalchemy2.elements import WKBElement, WKTElement 

21from fastapi import File, UploadFile 

22from typing import List 

23from fastapi import HTTPException 

24from fastapi.responses import JSONResponse, Response 

25import datetime as dt 

26import pytz 

27import pandas as pd 

28import csv 

29 

30from . import models, schemas 

31from toardb.variables import models as variables_models 

32from toardb.variables.crud import get_variable 

33from toardb.stationmeta import models as stationmeta_models 

34from toardb.stationmeta.crud import get_stationmeta 

35from toardb.timeseries.models import TimeseriesChangelog 

36from toardb.timeseries.schemas import TimeseriesWithCitation 

37from toardb.timeseries.crud import get_timeseries_by_unique_constraints, get_timeseries, get_citation, search_all, \ 

38 get_role_id_from_string 

39from toardb.utils.utils import get_value_from_str, get_str_from_value, create_filter 

40import toardb 

41from toarqc import get_toarqc_config, run_toarqc 

42from toarqc.tests import RangeTest 

43 

44def create_filter_from_aggreated_flags(i: int, filter_string: str, agg_flag_num: int): 

45 # definition of aggregated_flags: 

46 # agg. flag | composition of unique flags 

47 # ----------|---------------------------- 

48 # 100 | 0-6 

49 # 101 | 0-2 

50 # 102 | 3-5 

51 # 103 | 0, 1, 3, 4 

52 # 104 | 2, 5, 6 

53 # 110 | 10-16 

54 # 111 | 10-12 

55 # 112 | 13-16 

56 # 120 | 20-28 

57 # 121 | 20-23 

58 # 122 | 24-28 

59 # 130 | 10-28 

60 # 131 | 10-12, 20-23 

61 # 132 | 13-16, 24-28 

62 # 140 | 7, 16, 28 

63 filter_dict = { 100: [ 0, 1, 2, 3, 4, 5, 6], 

64 101: [ 0, 1, 2], 

65 102: [ 3, 4, 5], 

66 103: [ 0, 1, 3, 4], 

67 104: [ 2, 5, 6], 

68 110: [10, 11, 12, 13, 14, 15, 16], 

69 111: [10, 11, 12], 

70 112: [13, 14, 15, 16], 

71 120: [20, 21, 22, 23, 24, 25, 26, 27, 28], 

72 121: [20, 21, 22, 23], 

73 122: [24, 25, 26, 27, 28], 

74 130: [10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 

75 20, 21, 22, 23, 24, 25, 26, 27, 28], 

76 131: [10, 11, 12, 

77 20, 21, 22, 23], 

78 132: [13, 16, 24, 25, 26, 27, 28], 

79 140: [ 7, 16, 28] } 

80 for flag_num in filter_dict[agg_flag_num]: 

81 if i == 0: 

82 filter_string = f"(data.flags = {flag_num})" 

83 else: 

84 filter_string = filter_string + f" OR (data.flags = {flag_num})" 

85 i += 1 

86 return i, filter_string 

87 

88 

89def create_filter_from_flags(flags: str): 

90 i = 0 

91 filter_string = '' 

92 for flag in flags.split(','): 

93 flag_num = get_value_from_str(toardb.toardb.DF_vocabulary,flag) 

94 # check, whether an aggregated flag was given 

95 if (flag_num >= 100): 

96 i, filter_string = create_filter_from_aggreated_flags(i,filter_string,flag_num) 

97 else: 

98 if i == 0: 

99 filter_string = f"(data.flags = {flag_num})" 

100 else: 

101 filter_string = filter_string + f" OR (data.flags = {flag_num})" 

102 i += 1 

103 return f"({filter_string})" 

104 

105 

106def get_data(db: Session, timeseries_id: int, path_params, query_params, daterange = None): 

107 

108 # BUT for some unknown reasons, get_timeseries does not return variable and programme! 

109 def get_timeseries_meta(timeseries_id, lmerge=False): 

110 record = get_timeseries(db, timeseries_id=timeseries_id) 

111 # for some unknown reasons, variable, changelog, and programme have to be 'loaded' into the structure?! 

112 if record: 

113 dummy = record.variable 

114 dummy = record.programme 

115 dummy = record.changelog 

116 attribution, citation, license_txt = get_citation(db, timeseries_id=timeseries_id).values() 

117 record_dict = record.__dict__ 

118 record_dict['license'] = license_txt 

119 record_dict['citation'] = citation 

120 if attribution != None: 

121 record_dict['attribution'] = attribution 

122 if lmerge: 

123 return record_dict 

124 else: 

125 return TimeseriesWithCitation(**record_dict) 

126 else: 

127 return None 

128 

129 try: 

130 # translation of flags should be done here! 

131 lmerge = False 

132 flags = ",".join([item.strip() for v in query_params.getlist("flags") for item in v.split(',')]) 

133 if not flags: 

134 flags = None 

135 station_code = query_params.get("station_code", None) 

136 if station_code: 

137 limit, offset, fields, format, filters = create_filter(query_params, "timeseries_merged") 

138 lmerge = True 

139 else: 

140 limit, offset, fields, format, filters = create_filter(query_params, "data") 

141 d_filter = filters["d_filter"] 

142 if lmerge: 

143 try: 

144 start_date = dt.datetime.fromisoformat(daterange[0]) 

145 stop_date = dt.datetime.fromisoformat(daterange[1]) 

146 except: 

147 start_date = dt.datetime.fromisoformat(f"{daterange[0]}-01-01") 

148 stop_date = dt.datetime.fromisoformat(f"{daterange[1]}-12-31") 

149 d_filter += f"datetime BETWEEN '{start_date}' AND '{stop_date}'" 

150 fields_list = [] 

151 if fields: 

152 fields_list = fields.split(',') 

153 columns = ( [getattr(models.Data, field) for field in fields_list] 

154 if fields_list 

155 else list(models.Data.__table__.columns) ) 

156 except KeyError as e: 

157 status_code=400 

158 return JSONResponse(status_code=status_code, content=str(e)) 

159 if flags: 

160 filter_string = create_filter_from_flags(flags) 

161 data = db.query(*columns).filter(models.Data.timeseries_id == timeseries_id). \ 

162 filter(text(filter_string)). \ 

163 filter(text(d_filter)). \ 

164 order_by(models.Data.datetime).all() 

165 else: 

166 data = db.query(*columns).filter(models.Data.timeseries_id == timeseries_id). \ 

167 filter(text(d_filter)). \ 

168 order_by(models.Data.datetime).all() 

169 # get advantages from pydantic, but without having another call of the REST API 

170 # (especially needed for testing with pytest!) 

171 metadata = get_timeseries_meta(timeseries_id, lmerge) 

172# TOAR day 2022-08-09: We want the timezone information for the requested data 

173# metadata['timezone'] = ... 

174 if not metadata: 

175 return None 

176 if format == 'json': 

177 if lmerge: 

178 return columns, metadata, data 

179 else: 

180 composite = schemas.Composite(metadata=metadata, data=data) 

181 return composite 

182 elif format == 'csv': 

183 if not lmerge: 

184 # start with metadata 

185 content = '#' + metadata.json(indent=4, ensure_ascii=False).replace('\n', '\n#') + '\n' 

186 # add header 

187 content += ','.join(column.name for column in columns) + '\n' 

188 # now the data 

189 content += '\n'.join(','.join(f"{getattr(curr, column.name)}" for column in columns) for curr in data) 

190 return Response(content=content, media_type="text/csv") 

191 # take response for csv in timeseries_merge 

192 else: 

193 return columns, metadata, data 

194 else: 

195 status_code=400 

196 message='Invalid format!' 

197 return JSONResponse(status_code=status_code, content=message) 

198 

199 

200def get_next_version(db: Session, timeseries_id: int, path_params, query_params): 

201 data_version = db.query(func.max(models.Data.version)). \ 

202 filter(models.Data.timeseries_id == timeseries_id). \ 

203 distinct().first() 

204 if is_preliminary(data_version[0]): 

205 status_code = 407 

206 message = 'Preliminary data has no next version!' 

207 return JSONResponse(status_code=status_code, content=message) 

208 else: 

209 splitted = data_version[0].split('.') 

210 lmajor = query_params.get("major", "False").lower() == "true" 

211 major = int(splitted[0]) 

212 if lmajor: 

213 major += 1 

214 minor = 0 

215 else: 

216 minor = int(splitted[1]) + 1 

217 return f"{major:06d}.{minor:06d}.00000000000000" 

218 

219 

220def get_map_data(db:Session, variable_id, daterange): 

221 daterange = daterange.split(',') 

222 query_string = f"timeseries_id, value FROM timeseries t, data d WHERE d.timeseries_id=t.id AND t.variable_id={variable_id} AND d.datetime BETWEEN '{daterange[0]}' AND '{daterange[1]}'" 

223 data = db.execute(select(text(query_string))).all() 

224 return data 

225 

226 

227def get_data_with_staging(db: Session, timeseries_id: int, flags: str, format: str): 

228 

229 # BUT for some unknown reasons, get_timeseries does not return variable and programme! 

230 def get_timeseries_meta(timeseries_id): 

231 record = get_timeseries(db, timeseries_id=timeseries_id) 

232 # for some unknown reasons, variable, changelog, and programme have to be 'loaded' into the structure?! 

233 if record: 

234 dummy = record.variable 

235 dummy = record.programme 

236 dummy = record.changelog 

237 attribution, citation, license_txt = get_citation(db, timeseries_id=timeseries_id).values() 

238 record_dict = record.__dict__ 

239 record_dict['citation'] = citation 

240 record_dict['license'] =license_txt 

241 if attribution != None: 

242 record_dict['attribution'] = attribution 

243 return TimeseriesWithCitation(**record_dict) 

244 else: 

245 return None 

246 if flags: 

247 filter_string = create_filter_from_flags(flags) 

248 data = db.query(models.Data).filter(models.Data.timeseries_id == timeseries_id).filter(text(filter_string)).order_by(models.Data.datetime).all() 

249 else: 

250 query_txt = f"SELECT distinct ON (datetime) * \ 

251 FROM ( SELECT *,1 AS db FROM staging.data WHERE timeseries_id={timeseries_id} \ 

252 UNION \ 

253 SELECT *,2 AS db FROM public.data WHERE timeseries_id={timeseries_id}) AS mix \ 

254 ORDER BY datetime, db" 

255 data = db.query(models.Data).from_statement(text(query_txt)).all() 

256 

257 # get advantages from pydantic, but without having another call of the REST API 

258 # (especially needed for testing with pytest!) 

259 metadata = get_timeseries_meta(timeseries_id) 

260# TOAR day 2022-08-09: We want the timezone information for the requested data 

261# metadata['timezone'] = ... 

262 if not metadata: 

263 return None 

264 if format == 'json': 

265 composite = schemas.Composite(metadata=metadata, data=data) 

266 return composite 

267 elif format == 'csv': 

268 # start with metadata 

269 content = '#' + metadata.json(indent=4, ensure_ascii=False).replace('\n', '\n#') + '\n' 

270 # add header 

271 content += ','.join(column.name for column in models.Data.__mapper__.columns) + '\n' 

272 # now the data 

273 content += '\n'.join(','.join(f"{getattr(curr, column.name)}" for column in models.Data.__mapper__.columns) for curr in data) 

274 return Response(content=content, media_type="text/csv") 

275 else: 

276 status_code=400 

277 message='Invalid format!' 

278 return JSONResponse(status_code=status_code, content=message) 

279 

280 

281def get_all_merged_timeseries_ids(data): 

282 ts_ids = [] 

283 for _, _, ts_id in data: 

284 if not ts_id in ts_ids: 

285 ts_ids.append(ts_id) 

286 return ts_ids 

287 

288def replace_v8_v9_elements(elements): 

289 # Initialize variables to store the elements with 'v8' and 'v9' 

290 v8_elements = [] 

291 v9_elements = [] 

292 

293 # Iterate through the list of elements 

294 for i, element in enumerate(elements): 

295 # Check if the fourth element is 2012 

296 if element[3] == 2012: 

297 # Check if the seventh element is 'v8' or 'v9' -- there may be more than one element of v8 and v9 

298 if element[6] == 'v8': 

299 v8_elements.append(element) 

300 elif element[6] == 'v9': 

301 v9_elements.append(element) 

302 

303 # Check if both 'v8' and 'v9' elements were found 

304 if v8_elements != [] and v9_elements != []: 

305 # Create a new element with the third item substituted with 'v89' 

306 # use elements with highest coverages 

307 

308 v8_idx = max(range(len(v8_elements)), key=lambda i: v8_elements[i][4]) 

309 v8_element = v8_elements[v8_idx] 

310 v9_idx = max(range(len(v9_elements)), key=lambda i: v9_elements[i][4]) 

311 v9_element = v9_elements[v9_idx] 

312 new_element = (v8_element[0], v8_element[1], 'v89', v8_element[3], v8_element[4], v8_element[5], v8_element[6], v8_element[7], v8_element[8]) 

313 

314 # To edit: if there are other elements of the sequential years, edit there timeseies id to 'v89' 

315 # Add the new element to the list 

316 index_v8_element = elements.index(v8_element) 

317 elements[index_v8_element] = new_element 

318 # Remove the 'v8' and 'v9' elements from the list 

319 for i in range(len(elements) - 1, -1, -1): 

320 if elements[i][3] == 2012 and elements[i][6] in ['v8', 'v9'] and elements[i][2] != 'v89': 

321 del elements[i] 

322 

323 return elements, v8_element, v9_element 

324 else: 

325 return elements, None, None 

326 

327 

328def select_provider(data): 

329 # Organize data by year. Extract the timeseries entry from the provider with the highest rank. 

330 # Input data example: [(46, 5, 18763, 1991, 0.3442922374429223, 1, 'N/A', dt.datetime(1993, 1, 1, 2, 0, tzinfo=dt.timezone.utc), dt.datetime(2025, 3, 5, 23, 0, tzinfo=dt.timezone.utc)), (46, 5, 18763, 1992, 0.345, 1, 'N/A', dt.datetime(1993, 1, 1, 2, 0, tzinfo=dt.timezone.utc), dt.datetime(2025, 3, 5, 23, 0, tzinfo=dt.timezone.utc)), (46, 5, 18763, 1993, 0.902054794520548, 1, 'N/A', dt.datetime(1993, 1, 1, 2, 0, tzinfo=dt.timezone.utc), dt.datetime(2025, 3, 5, 23, 0, tzinfo=dt.timezone.utc)), (46, 5, 18764, 1994, 0.56, 1, 'N/A', dt.datetime(2008, 1, 1, 1, 0, tzinfo=dt.timezone.utc) 

331 year_data = defaultdict(list) 

332 for _, _, ts_id, year, coverage, order, version, start_date, end_date in data: 

333 year_data[year].append((ts_id, coverage, order, version, start_date, end_date)) 

334 

335 selected = [] 

336 # Process each year 

337 for year, entries in year_data.items(): 

338 # Sort by provider order (ascending, since lower is more relevant) 

339 entries.sort(key=lambda x: x[2]) 

340 

341 best_ts_id, best_coverage = entries[0][0], entries[0][1] 

342 

343 if best_coverage >= 0.75: 

344 selected.append((year, best_ts_id)) 

345 continue 

346 

347 for ts_id, coverage, *_ in entries[1:]: 

348 # do not confuse user in taking data of lowest quality, even if there is no data 

349 if (coverage < 0.2 and best_coverage == 0.): 

350 continue 

351 if coverage >= best_coverage * 1.2: 

352 best_ts_id, best_coverage = ts_id, coverage 

353 

354 if best_coverage >= 0.75: 

355 break 

356 selected.append((year, best_ts_id)) 

357 return selected 

358 

359def ensure_v8_before_v9_and_strip_tags(entries): 

360 #Handling the case when v9's TS ID is smaller than v8's TS ID. v9 has to be always located after v8, whatever TS ID they contain. 

361 v8_index = None 

362 v9_index = None 

363 

364 # Find the indices of v8 and v9 

365 for i, entry in enumerate(entries): 

366 if len(entry) == 3: 

367 if entry[2] == 'v8': 

368 v8_index = i 

369 elif entry[2] == 'v9': 

370 v9_index = i 

371 

372 # Swap if v9 comes before v8 

373 if v9_index is not None and v8_index is not None and v9_index < v8_index: 

374 entries[v8_index], entries[v9_index] = entries[v9_index], entries[v8_index] 

375 

376 # Strip the third element if present 

377 cleaned_entries = [(entry[0], entry[1]) if len(entry) == 3 else entry for entry in entries] 

378 

379 return cleaned_entries 

380 

381def merge_sequences(selected): 

382 # Merge the timeseries (same timeseries id) from sequential years together. 

383 merged = [] 

384 if not selected: 

385 return merged 

386 

387 selected.sort() 

388 eeasorted_selected = ensure_v8_before_v9_and_strip_tags(selected) 

389 

390 start_year, ts_id = eeasorted_selected[0] 

391 end_year = start_year 

392 

393 for year, current_ts_id in eeasorted_selected[1:]: 

394 if current_ts_id != ts_id: 

395 merged.append([start_year, end_year, ts_id]) 

396 start_year = year 

397 end_year = start_year 

398 ts_id = current_ts_id 

399 else: 

400 end_year = year 

401 merged.append([start_year, end_year, ts_id]) 

402 return merged 

403 

404def format_timeseries(data, v8_element, v9_element): 

405 def format_start(start_year): 

406 # Formatting the data outputted from the previous function. 

407 formatted_date = f"{start_year}-01-01 00:00" 

408 return formatted_date 

409 def format_end(end_year): 

410 formatted_date = f"{end_year}-12-31 23:00" 

411 return formatted_date 

412 

413 v8_end_time = None 

414 

415 if v8_element and v9_element: 

416 # Treating the case when v8_element and v9_element exist. 

417 idelete = 0 

418 for i, (start, end, ts_id) in enumerate(data): 

419 if ts_id == v8_element[2]: 

420 new_end = v8_element[8] # Replace second element 

421 if new_end.year > end: 

422 new_end = dt.datetime(end,12,31,23,0, tzinfo=dt.timezone.utc) 

423 v8_end_time = new_end # Save for comparison 

424 data[i] = [format_start(start), new_end.strftime("%Y-%m-%d %H:%M"), ts_id] 

425 elif ts_id == v9_element[2]: 

426 new_start = v9_element[7] 

427 if new_start.year < start: 

428 new_start = dt.datetime(start,1,1,0,0, tzinfo=dt.timezone.utc) 

429 if new_start <= v8_end_time: 

430 new_start = v8_end_time + dt.timedelta(hours=1) 

431 if v8_end_time and v8_end_time == dt.datetime(2012, 12, 31, 23, tzinfo=pytz.utc): 

432 new_start = dt.datetime(2013,1,1,0,0, tzinfo=dt.timezone.utc) 

433 if new_start > v9_element[8]: 

434 idelete = i 

435 data[i] = [new_start.strftime("%Y-%m-%d %H:%M"), format_end(end), ts_id] # Replace first element 

436 else: 

437 # Format other dates 

438 data[i] = [format_start(start), format_end(end), ts_id] 

439 if idelete: 

440 del data[idelete] 

441 else: 

442 for i, (start, end, ts_id) in enumerate(data): 

443 data[i] = [format_start(start), format_end(end), ts_id] 

444 return data 

445 

446def substitute_v89_entry(entries, v8_element, v9_element): 

447 new_elements = [(2012, v8_element[2], "v8"), (2012, v9_element[2], "v9")] 

448 modified_entries = [(year, tsid) for year, tsid in entries if tsid != 'v89'] + new_elements 

449 return modified_entries 

450 

451 

452def get_merging_list(db, station_code: str, variable_id: str, daterange: str, role: str = None): 

453 from_tables = "yearly_coverage y, timeseries t, stationmeta_core s" 

454 join_clauses = "t.id=y.timeseries_id AND s.id = y.station_id" 

455 filter_clauses = f"'{station_code}'=ANY(s.codes) AND y.variable_id={variable_id}" 

456 ordering = " ORDER BY year, t.order, y.coverage DESC, y.timeseries_id" 

457 if role: 

458 from_tables = from_tables + ", timeseries_timeseries_roles tr" 

459 join_clauses = join_clauses + " AND t.id = tr.timeseries_id" 

460 if role.startswith('~'): 

461 role_ids = [ x for r in role[1:].split(',') for x in get_role_id_from_string(db, r)] 

462 if role_ids != []: 

463 filter_clauses = filter_clauses + f" AND NOT EXISTS (SELECT 1 FROM timeseries_timeseries_roles tr2" \ 

464 + " WHERE tr2.timeseries_id = t.id" \ 

465 + f" AND tr2.role_id = ANY(ARRAY{role_ids}))" 

466 else: 

467 role_ids = [ x for r in role.split(',') for x in get_role_id_from_string(db, r)] 

468 if role_ids != []: # this results in an empty query result 

469 filter_clauses = filter_clauses + f" AND tr.role_id=ANY(ARRAY{role_ids})" 

470 else: # this results in an empty query result 

471 filter_clauses = filter_clauses + f" AND tr.role_id=ANY(ARRAY[-999])" 

472 query_string = f"y.*, t.order, t.provider_version, t.data_start_date, t.data_end_date FROM {from_tables} WHERE {join_clauses} AND {filter_clauses}{ordering};" 

473 records = db.execute(select(text(query_string))).all() 

474 

475 new_records, v8_element, v9_element = replace_v8_v9_elements(records) # Create fake timeseries IDs for v8 and v9 timeseries to make them distinguishable in the next steps 

476 result = select_provider(new_records) # Coverage check 

477 

478 eea_ts_id = next((ts_id for year, ts_id in result if year == 2012), None) # Extract ts_id for any time series of 2012 

479 modified_entries = None 

480 if eea_ts_id == 'v89': # if a timeseries with "v89" has passed through the coverage check, the next steps will execute the processing of the 2012 EEA particular data along with the general data. 

481 modified_entries = substitute_v89_entry(result, v8_element, v9_element) 

482 

483 if modified_entries: 

484 merged_results = merge_sequences(modified_entries) 

485 ts_list = format_timeseries(merged_results, v8_element, v9_element) 

486 else: 

487 merged_results = merge_sequences(result) # no special processing, since no splitted 2012 EEA data were found. 

488 ts_list = format_timeseries(merged_results, None, None) 

489 

490 if daterange: 

491 start_date, end_date = [dt.datetime.fromisoformat(date) for date in daterange.split(',')] 

492 ts_list_daterange = [] 

493 for ts in ts_list: 

494 try: 

495 ts_start = dt.datetime.fromisoformat(ts[0]) 

496 ts_end = dt.datetime.fromisoformat(ts[1]) 

497 except: 

498 ts_start = dt.datetime.fromisoformat(f"{ts[0]}-01-01") 

499 ts_end = dt.datetime.fromisoformat(f"{ts[1]}-12-31") 

500 if (start_date <= ts_start <= end_date) or (start_date <= ts_end <= end_date) or (ts_start <= start_date and ts_end >= end_date): 

501 intersection_start = max(start_date, ts_start) 

502 intersection_end = min(end_date, ts_end) 

503 ts_list_daterange.append([intersection_start.isoformat(), intersection_end.isoformat(), ts[2]]) 

504 ts_list = ts_list_daterange 

505 timeseries_ids = [] 

506 if ts_list == []: 

507 timeseries_ids = get_all_merged_timeseries_ids(merged_results) 

508 return ts_list, timeseries_ids 

509 

510 

511def get_merged_data(db: Session, variable_id: int, station_code: str, role: str, path_params, query_params): 

512 

513 ## still to be done: the daterange given by the user should be taken into account to only load the data that is needed 

514 # do this in the first place (do not cut the data short later) 

515 # --> take the daterange as a separate argument to be able know it, when being ignored by utils.create_filters 

516 

517 daterange = query_params.get("daterange", None) 

518 

519 ## check, whether station_code exists 

520 db_stationmeta = get_stationmeta(db, station_code=station_code, fields="id") 

521 if db_stationmeta is None: 

522 raise HTTPException(status_code=404, detail=f"Metadata for station '{station_code}' not found.") 

523 

524 ts_list, timeseries_ids = get_merging_list(db, station_code, variable_id, daterange, role) 

525 data_merged = {'metadata': [], 

526 'data': []} 

527 used_timeseries_ids = [] 

528 if ts_list == []: 

529 # some time series do not have an entry in the yearly coverage table! 

530 if timeseries_ids == []: 

531 data_merged['metadata'] = search_all(db, path_params, query_params, endpoint='timeseries_merged') 

532 data_merged['data'] = [] 

533 return data_merged 

534 for ts_id in timeseries_ids: 

535 ts_list.append(['1960-01-01', '1961-01-01', ts_id]) 

536 for ts_part in ts_list: 

537 columns, metadata, data = get_data(db, ts_part[2], path_params, query_params, daterange=[ts_part[0], ts_part[1]]) 

538 data_merged['data'] += data 

539 # do not show metadata twice 

540 if not ts_part[2] in used_timeseries_ids: 

541 data_merged['metadata'] += [ TimeseriesWithCitation(**metadata) ] 

542 used_timeseries_ids.append(ts_part[2]) 

543 format = query_params.get("format", "json") 

544 if format == 'csv': 

545 content = '' 

546 # start with metadata 

547 for part_meta in data_merged['metadata']: 

548 content += '#' + part_meta.json(indent=4, ensure_ascii=False).replace('\n', '\n#') + '\n' 

549 # add header 

550 content += ','.join(column.name for column in columns) + '\n' 

551 # now the data 

552 content += '\n'.join(','.join(f"{getattr(curr, column.name)}" for column in columns) for curr in data_merged['data']) 

553 return Response(content=content, media_type="text/csv") 

554 else: 

555 return data_merged 

556 

557def get_data_by_datetime_and_timeseriesid(db: Session, datetime: dt.datetime, timeseries_id: int): 

558 return db.query(models.Data).filter([models.Data.datetime== datetime, models.Data.timeseries_id == timeseries_id]).first() 

559 

560def get_all_data(db: Session, limit: int, offset: int = 0): 

561 return db.query(models.Data).order_by(models.Data.datetime).limit(limit).all() 

562 

563def is_preliminary(version: str): 

564 return (version.split('.')[0] == '000000') 

565 

566def create_data_record(db: Session, engine: Engine, 

567 series_id: int, datetime: dt.datetime, 

568 value: float, flag: str, version: str, 

569 author_id: int): 

570 toarqc_config_type: str = 'standard' 

571 timeseries = get_timeseries(db=db,timeseries_id=series_id) 

572 variable = get_variable(db=db, variable_id=timeseries.variable_id) 

573 parameter = variable.name 

574 data_dict = {"datetime": datetime, 

575 "value": value, 

576 "flags": flag, 

577 "version": version, 

578 "timeseries_id": series_id} 

579 df = pd.DataFrame([data_dict]).set_index("datetime") 

580 try: 

581 test_config = get_toarqc_config('static/../toardb/data/toarqc_config',parameter, toarqc_config_type) 

582 except FileNotFoundError: 

583 message = f'no toarqc configuration found for {parameter}, {toarqc_config_type}' 

584 status_code = 400 

585 return JSONResponse(status_code=status_code, content=message) 

586 result = run_toarqc(test_config, df, ok_limit=0.85, questionable_limit=0.6) 

587 combined_flag_matrix = {( 'OK', 'OK') : 'OKPreliminaryNotChecked', 

588 ( 'Questionable', 'OK') : 'QuestionablePreliminaryNotChecked', 

589 ( 'Erroneous', 'OK') : 'ErroneousPreliminaryNotChecked', 

590 ( 'OK', 'Erroneous') : 'ErroneousPreliminaryFlagged1', 

591 ( 'Questionable', 'Erroneous') : 'ErroneousPreliminaryFlagged2', 

592 ( 'Erroneous', 'Erroneous') : 'Erroneous_Preliminary_Confirmed'} 

593 combined_flag = combined_flag_matrix[(flag,result['flags'].iloc[0])] 

594 data_dict["flags"] = get_value_from_str(toardb.toardb.DF_vocabulary,combined_flag.strip()) 

595 data = models.Data(**data_dict) 

596 db.rollback() 

597 db.add(data) 

598 # adjust data_start_date, data_end_date 

599 datetime = datetime.replace(tzinfo=timeseries.data_end_date.tzinfo) 

600 if datetime < timeseries.data_start_date: 

601 timeseries.data_start_date = datetime 

602 if datetime > timeseries.data_end_date: 

603 timeseries.data_end_date = datetime 

604 db.add(timeseries) 

605 result = db.commit() 

606 db.refresh(data) 

607 # if not preliminary data: create changelog entry 

608 if not is_preliminary(version): 

609 type_of_change = get_value_from_str(toardb.toardb.CL_vocabulary,"Created") 

610 description="data record created" 

611 db_changelog = TimeseriesChangelog(description=description, timeseries_id=series_id, author_id=author_id, type_of_change=type_of_change, 

612 old_value='', new_value='', period_start=datetime, period_end=datetime, version=version) 

613 db.add(db_changelog) 

614 db.commit() 

615 status_code=200 

616 message='Data successfully inserted.' 

617 return JSONResponse(status_code=status_code, content=message) 

618 

619 

620def insert_dataframe (db: Session, engine: Engine, df: pd.DataFrame, toarqc_config_type: str = 'standard', dry_run: bool = False, parameter: str = 'o3', preliminary: bool = False, force: bool = False): 

621 # df: pandas.DataFrame 

622 # index: datetime 

623 # 1st column: value 

624 # 2nd column: flags (0: 'OK', 1: 'Questionable', 2: 'Erroneous') 

625 # 3rd column: timeseries_id 

626 # 4th column: version 

627 df['flags'] = df['flags'].replace([0],'OK') 

628 df['flags'] = df['flags'].replace([1],'Questionable') 

629 df['flags'] = df['flags'].replace([2],'Erroneous') 

630 try: 

631 test_config = get_toarqc_config('static/../toardb/data/toarqc_config',parameter, toarqc_config_type) 

632 except FileNotFoundError: 

633 message = f'no toarqc configuration found for {parameter}, {toarqc_config_type}' 

634 status_code = 400 

635 return JSONResponse(status_code=status_code, content=message) 

636 # get flags and qc-statistics from toarqc  

637 # available keys in result (at the moment): 

638 # 'time_series' 

639 # 'test_config' 

640 # 'ok_limit' 

641 # 'questionable_limit' 

642 # 'metadata' 

643 # 'plot' 

644 # 'outfile' 

645 # 'plot_type' 

646 # 'results' 

647 # 'qc-score' 

648 # 'flags' 

649 # 'number_OK' 

650 # 'number_Questionable' 

651 # 'number_Erroneous' 

652 # 'percentage_OK' 

653 # 'percentage_Questionable' 

654 # 'percentage_Erroneous' 

655 # 'qc-score_g0' 

656 # 'flags_g0' 

657 # 'qc-score_g1' 

658 # 'flags_g1' 

659 # 'qc-score_g2' 

660 # 'flags_g2' 

661 # 'qc-score_g3' 

662 # 'flags_g3' 

663 # 'qc-score_g0_range_test' 

664 # 'flags_g0_range_test' 

665 # 'qc-score_g1_sigma_test' 

666 # 'flags_g1_sigma_test' 

667 # 'qc-score_g2_constant_value_test' 

668 # 'flags_g2_constant_value_test' 

669 # 'qc-score_g2_positive_spike_test' 

670 # 'flags_g2_positive_spike_test' 

671 # 'qc-score_g2_negative_spike_test' 

672 # 'flags_g2_negative_spike_test' 

673 # 'qc-score_g3_before_nan_test' 

674 # 'flags_g3_before_nan_test' 

675 # 'qc-score_g3_after_nan_test' 

676 # 'flags_g3_after_nan_test' 

677 # toarqc expects pandas.Series 

678 # --> if getting pandas.DataFrame, it will take index and first column 

679 # (other columns will be ignored) 

680 if toarqc_config_type != 'realtime': 

681 try: 

682 result = run_toarqc(test_config, df, ok_limit=0.85, questionable_limit=0.6) 

683 except ValueError as ve: 

684 message = {"detail":{"message":f'Aborted by automatic quality control: {ve}'}} 

685 status_code = 445 

686 return JSONResponse(status_code=status_code, content=message) 

687 percentage_OK = result['percentage_OK'] 

688 if percentage_OK >= 0.9 or force: 

689 target_schema='public' 

690 message = {"detail":{"message":"Data successfully inserted."}} 

691 status_code = 200 

692 else: 

693 target_schema='staging' 

694 message = f'Aborted by automatic quality control: percentage_OK = {percentage_OK}' 

695 status_code = 446 

696 toarqc_flags = result['flags'] 

697 # combine toarqc flags with flags given by provider (see matrix: TOAR_UG_Vol03_Database_2021-05.docx, chapter 5.2) 

698 # preliminary provider toarqc combined 

699 combined_flag_matrix = {( False, 'OK', 'OK') : 'OKValidatedQCPassed', 

700 ( False, 'OK', 'Questionable') : 'QuestionableValidatedFlagged', 

701 ( False, 'OK', 'Erroneous') : 'ErroneousValidatedFlagged1', 

702 ( False, 'Questionable', 'OK') : 'QuestionableValidatedUnconfirmed', 

703 ( False, 'Questionable', 'Questionable') : 'QuestionableValidatedConfirmed', 

704 ( False, 'Questionable', 'Erroneous') : 'ErroneousValidatedFlagged2',