16371583884

Committed 18 Jul 2025 01:19PM UTC coverage: 70.054% (-0.9%) from 70.952%

Build # 16371583884

Build Type

push

github

Committed by

web-flow

Commit Message

fix:fix for airspace access and shapely use (#2839)

* fix for airspace access and shapely use

* pep8 fixes

Run Details

17 of 18 new or added lines in 1 file covered. (94.44%)

1738 existing lines in 41 files now uncovered.

14427 of 20594 relevant lines covered (70.05%)

0.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.41

/mslib/utils/airdata.py

# -*- coding: utf-8 -*-
"""

    mslib.utils.airdata
    ~~~~~~~~~~~~~~~~

    Functions for getting and downloading airspaces and airports.

    This file is part of MSS.

    :copyright: Copyright 2021 May Bär
    :copyright: Copyright 2021-2025 by the MSS team, see AUTHORS.
    :license: APACHE-2.0, see LICENSE for details.

    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.
"""


import csv
import humanfriendly
import os
import requests
from pathlib import Path
from PyQt5 import QtWidgets
import logging
import time

import defusedxml.ElementTree as etree
from mslib.msui.constants import MSUI_CONFIG_PATH

AIPDIR = Path(MSUI_CONFIG_PATH) / "downloads" / "aip"
AIPDIR.mkdir(parents=True, exist_ok=True)


class Airspace:
    """
    This class provides airspaces/airports URLs and variables
    """
    airports = []

    airports_mtime = 0

    data = []

    data_mtime = {}

    data_url = "https://storage.googleapis.com/storage/v1/b/29f98e10-a489-4c82-ae5e-489dbcd4912f/o"

    data_download_url = "https://storage.googleapis.com/storage/v1/b/29f98e10-a489-4c82-ae5e-489dbcd4912f/o/" \
                        "{}_asp.xml?alt=media"

    data_cache = [('ad_asp.xml', '377'), ('ae_asp.xml', '110238'), ('af_asp.xml', '377'), ('ag_asp.xml', '377'),
                  ('ai_asp.xml', '377'), ('al_asp.xml', '11360'), ('am_asp.xml', '377'), ('ao_asp.xml', '377'),
                  ('aq_asp.xml', '377'), ('ar_asp.xml', '975402'), ('as_asp.xml', '377'),
                  ('at_asp.xml', '1059077'),
                  ('au_asp.xml', '9667669'), ('aw_asp.xml', '377'), ('ax_asp.xml', '377'), ('az_asp.xml', '377'),
                  ('ba_asp.xml', '76717'), ('bb_asp.xml', '377'), ('bd_asp.xml', '377'), ('be_asp.xml', '462667'),
                  ('bf_asp.xml', '377'), ('bg_asp.xml', '440690'), ('bh_asp.xml', '68519'), ('bi_asp.xml', '377'),
                  ('bj_asp.xml', '377'), ('bl_asp.xml', '377')
                ]


def download_progress(file_path, url, progress_callback=lambda f: logging.info("%sKB Downloaded", int(f))):
    """
    Downloads the file at the given url to file_path and keeps track of the progress
    """
    try:
        with open(file_path, "wb+") as file:
            logging.info("Downloading to %s. This might take a while.", file_path)
            response = requests.get(url, stream=True, timeout=5)
            length = response.headers.get("content-length")
            if length is None:  # no content length header
                file.write(response.content)
            else:
                dl = 0
                for data in response.iter_content(chunk_size=1024 * 1024):
                    dl += len(data)
                    file.write(data)
                    progress_callback(dl / 1024)
    except requests.exceptions.RequestException:
        os.remove(file_path)
        QtWidgets.QMessageBox.information(None, "Download failed", f"{url} was unreachable, please try again later.")


def get_airports(force_download=False, url=None):
    """
    Gets or downloads the airports.csv in ~/.config/msui/downloads/aip and returns all airports within
    """
    _airports = Airspace().airports
    _airports_mtime = Airspace().airports_mtime

    if url is None:
        url = "https://davidmegginson.github.io/ourairports-data/airports.csv"

    file_exists = (Path(AIPDIR) / "airports.csv").exists()

    if _airports and file_exists and \
            os.path.getmtime(str(AIPDIR / "airports.csv")) == _airports_mtime:
        return _airports

    time_outdated = 60 * 60 * 24 * 30  # 30 days
    is_outdated = (file_exists and
                   (time.time() - os.path.getmtime(str(Path(AIPDIR / "airports.csv")))) > time_outdated)

    if (force_download or is_outdated or not file_exists) \
            and QtWidgets.QMessageBox.question(None, "Allow download", f"You selected airports to be "
                                               f"{'drawn' if not force_download else 'downloaded (~10 MB)'}." +
                                               ("\nThe airports file first needs to be downloaded or updated (~10 MB)."
                                                if not force_download else "") + "\nIs now a good time?",
                                               QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
                                               QtWidgets.QMessageBox.Yes) \
            == QtWidgets.QMessageBox.Yes:
        download_progress(os.path.join(AIPDIR / "airports.csv"), url)

    if Path(AIPDIR / "airports.csv").exists():
        with (AIPDIR / "airports.csv").open("r", encoding="utf8") as file:
            _airports_mtime = (AIPDIR / "airports.csv").stat().st_mtime
            return list(csv.DictReader(file, delimiter=","))

    else:
        return []


def get_available_airspaces():
    """
    Gets and returns all available airspaces and their sizes from openaip
    """
    try:
        next_page_token = None
        all_airspaces = []
        while True:
            params = {}
            if next_page_token:
                params["pageToken"] = next_page_token
            response = requests.get(Airspace.data_url, params=params, timeout=5)
            if response.status_code != 200:
                return Airspace.data_cache
            data = response.json()
            for item in data.get("items", []):
                if "_asp.xml" in item["name"] and item["size"] != "0":
                    all_airspaces.append((item["name"], item["size"]))
            next_page_token = response.json().get("nextPageToken")
            if not next_page_token:
                break
        return all_airspaces
    except requests.exceptions.RequestException:
        return Airspace.data_cache


def update_airspace(force_download=False, countries=None):
    """
    Downloads the requested airspaces from their respective country code if it is over a month old
    """
    if countries is None:
        countries = ["de"]

    for country in countries:
        location = AIPDIR / f"{country}_asp.xml"
        url = Airspace.data_download_url.format(country)
        available = get_available_airspaces()
        try:
            data = [airspace for airspace in available if airspace[0].startswith(country)][0]
        except IndexError:
            logging.info("countries: %s not exists", ' '.join(countries))
            continue
        file_exists = location.exists()

        is_outdated = file_exists and (time.time() - os.path.getmtime(str(location.resolve()))) > 60 * 60 * 24 * 30

        if (force_download or is_outdated or not file_exists) \
                and QtWidgets.QMessageBox.question(
                    None, "Allow download",
                    f"The selected {country} airspace needs to be downloaded "
                    f"({humanfriendly.format_size(int(data[-1]))})\nIs now a good time?",
                    QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
                    QtWidgets.QMessageBox.Yes) \
                == QtWidgets.QMessageBox.Yes:
            download_progress(str(location.resolve()), url)


def get_airspaces(countries=None):
    """
    Gets the .xml files in ~/.config/msui/downloads/aip and returns all airspaces within
    """
    if countries is None:
        countries = []
    _airspaces = Airspace().data
    _airspaces_mtime = Airspace().data_mtime

    reload = False
    files = [f"{country}_asp.xml" for country in countries]
    update_airspace(countries=countries)
    files = [file for file in files if (AIPDIR / file).exists()]

    if _airspaces and len(files) == len(_airspaces_mtime):
        for file in files:
            if file not in _airspaces_mtime or \
                    os.path.getmtime(str(AIPDIR / file)) != _airspaces_mtime[file]:
                reload = True
                break
        if not reload:
            return _airspaces

    _airspaces_mtime = {}
    _airspaces = []
    for file in files:
        fpath = str(AIPDIR / file)
        root = etree.parse(fpath).getroot()
        valid_file = len(set([elem.tag for elem in root.iter()])) == 12
        if valid_file:
            airspaces = (root.find('{https://www.openaip.net}AIRSPACES'))
            names = [dat.text for dat in airspaces.findall(".//{https://www.openaip.net}NAME")]
            polygons = [dat.text for dat in airspaces.findall(".//{https://www.openaip.net}POLYGON")]
            countries = [dat.text for dat in airspaces.findall(".//{https://www.openaip.net}COUNTRY")]
            tops = []
            top_units = []
            for dat in airspaces.findall(".//{https://www.openaip.net}ALTLIMIT_TOP"):
                unit = dat[0]
                top_units.append(unit.get("UNIT"))
                _, data = dat.iter()
                tops.append(float(data.text))
            bottoms = []
            bottom_units = []
            for dat in airspaces.findall(".//{https://www.openaip.net}ALTLIMIT_BOTTOM"):
                unit = dat[0]
                bottom_units.append(unit.get("UNIT"))
                _, data = dat.iter()
                bottoms.append(float(data.text))

            for index, value in enumerate(names):
                airspace_data = {
                    "name": names[index],
                    "polygon": polygons[index],
                    "top": tops[index],
                    "top_unit": top_units[index],
                    "bottom": bottoms[index],
                    "bottom_unit": bottom_units[index],
                    "country": countries[index]
                }

                # Convert to kilometers
                airspace_data["top"] /= 3281 if airspace_data["top_unit"] == "F" else 32.81
                airspace_data["bottom"] /= 3281 if airspace_data["bottom_unit"] == "F" else 32.81
                airspace_data["top"] = round(airspace_data["top"], 2)
                airspace_data["bottom"] = round(airspace_data["bottom"], 2)
                airspace_data.pop("top_unit")
                airspace_data.pop("bottom_unit")

                airspace_data["polygon"] = [(float(data.split()[0]), float(data.split()[-1]))
                                            for data in airspace_data["polygon"].split(",")]
                _airspaces.append(airspace_data)
                _airspaces_mtime[file] = os.path.getmtime(os.path.join(AIPDIR, file))
        else:
            QtWidgets.QMessageBox.information(None, "No Airspaces data in file:", f"{file}")

    return _airspaces

1	# -- coding: utf-8 --
2	"""
3
4	mslib.utils.airdata
5	~~~~~~~~~~~~~~~~
6
7	Functions for getting and downloading airspaces and airports.
8
9	This file is part of MSS.
10
11	:copyright: Copyright 2021 May Bär
12	:copyright: Copyright 2021-2025 by the MSS team, see AUTHORS.
13	:license: APACHE-2.0, see LICENSE for details.
14
15	Licensed under the Apache License, Version 2.0 (the "License");
16	you may not use this file except in compliance with the License.
17	You may obtain a copy of the License at
18
19	http://www.apache.org/licenses/LICENSE-2.0
20
21	Unless required by applicable law or agreed to in writing, software
22	distributed under the License is distributed on an "AS IS" BASIS,
23	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
24	See the License for the specific language governing permissions and
25	limitations under the License.
26	"""
27
28
29	import csv	1✔
30	import humanfriendly	1✔
31	import os	1✔
32	import requests	1✔
33	from pathlib import Path	1✔
34	from PyQt5 import QtWidgets	1✔
35	import logging	1✔
36	import time	1✔
37
38	import defusedxml.ElementTree as etree	1✔
39	from mslib.msui.constants import MSUI_CONFIG_PATH	1✔
40
41	AIPDIR = Path(MSUI_CONFIG_PATH) / "downloads" / "aip"	1✔
42	AIPDIR.mkdir(parents=True, exist_ok=True)	1✔
43
44
45	class Airspace:	1✔
46	"""
47	This class provides airspaces/airports URLs and variables
48	"""
49	airports = []	1✔
50
51	airports_mtime = 0	1✔
52
53	data = []	1✔
54
55	data_mtime = {}	1✔
56
57	data_url = "https://storage.googleapis.com/storage/v1/b/29f98e10-a489-4c82-ae5e-489dbcd4912f/o"	1✔
58
59	data_download_url = "https://storage.googleapis.com/storage/v1/b/29f98e10-a489-4c82-ae5e-489dbcd4912f/o/" \	1✔
60	"{}_asp.xml?alt=media"
61
62	data_cache = [('ad_asp.xml', '377'), ('ae_asp.xml', '110238'), ('af_asp.xml', '377'), ('ag_asp.xml', '377'),	1✔
63	('ai_asp.xml', '377'), ('al_asp.xml', '11360'), ('am_asp.xml', '377'), ('ao_asp.xml', '377'),
64	('aq_asp.xml', '377'), ('ar_asp.xml', '975402'), ('as_asp.xml', '377'),
65	('at_asp.xml', '1059077'),
66	('au_asp.xml', '9667669'), ('aw_asp.xml', '377'), ('ax_asp.xml', '377'), ('az_asp.xml', '377'),
67	('ba_asp.xml', '76717'), ('bb_asp.xml', '377'), ('bd_asp.xml', '377'), ('be_asp.xml', '462667'),
68	('bf_asp.xml', '377'), ('bg_asp.xml', '440690'), ('bh_asp.xml', '68519'), ('bi_asp.xml', '377'),
69	('bj_asp.xml', '377'), ('bl_asp.xml', '377')
70	]
71
72
73	def download_progress(file_path, url, progress_callback=lambda f: logging.info("%sKB Downloaded", int(f))):	1✔
74	"""
75	Downloads the file at the given url to file_path and keeps track of the progress
76	"""
77	try:	1✔
78	with open(file_path, "wb+") as file:	1✔
79	logging.info("Downloading to %s. This might take a while.", file_path)	1✔
80	response = requests.get(url, stream=True, timeout=5)	1✔
81	length = response.headers.get("content-length")	1✔
82	if length is None: # no content length header	1✔
UNCOV 83	file.write(response.content)	×
84	else:
85	dl = 0	1✔
86	for data in response.iter_content(chunk_size=1024 * 1024):	1✔
87	dl += len(data)	1✔
88	file.write(data)	1✔
89	progress_callback(dl / 1024)	1✔
UNCOV 90	except requests.exceptions.RequestException:	×
91	os.remove(file_path)	×
92	QtWidgets.QMessageBox.information(None, "Download failed", f"{url} was unreachable, please try again later.")	×
93
94
95	def get_airports(force_download=False, url=None):	1✔
96	"""
97	Gets or downloads the airports.csv in ~/.config/msui/downloads/aip and returns all airports within
98	"""
99	_airports = Airspace().airports	1✔
100	_airports_mtime = Airspace().airports_mtime	1✔
101
102	if url is None:	1✔
103	url = "https://davidmegginson.github.io/ourairports-data/airports.csv"	1✔
104
105	file_exists = (Path(AIPDIR) / "airports.csv").exists()	1✔
106
107	if _airports and file_exists and \	1✔
108	os.path.getmtime(str(AIPDIR / "airports.csv")) == _airports_mtime:
UNCOV 109	return _airports	×
110
111	time_outdated = 60 * 60 * 24 * 30 # 30 days	1✔
112	is_outdated = (file_exists and	1✔
113	(time.time() - os.path.getmtime(str(Path(AIPDIR / "airports.csv")))) > time_outdated)
114
115	if (force_download or is_outdated or not file_exists) \	1✔
116	and QtWidgets.QMessageBox.question(None, "Allow download", f"You selected airports to be "
117	f"{'drawn' if not force_download else 'downloaded (~10 MB)'}." +
118	("\nThe airports file first needs to be downloaded or updated (~10 MB)."
119	if not force_download else "") + "\nIs now a good time?",
120	QtWidgets.QMessageBox.Yes \| QtWidgets.QMessageBox.No,
121	QtWidgets.QMessageBox.Yes) \
122	== QtWidgets.QMessageBox.Yes:
123	download_progress(os.path.join(AIPDIR / "airports.csv"), url)	1✔
124
125	if Path(AIPDIR / "airports.csv").exists():	1✔
126	with (AIPDIR / "airports.csv").open("r", encoding="utf8") as file:	1✔
127	_airports_mtime = (AIPDIR / "airports.csv").stat().st_mtime	1✔
128	return list(csv.DictReader(file, delimiter=","))	1✔
129
130	else:
131	return []	1✔
132
133
134	def get_available_airspaces():	1✔
135	"""
136	Gets and returns all available airspaces and their sizes from openaip
137	"""
138	try:	1✔
139	next_page_token = None	1✔
140	all_airspaces = []	1✔
141	while True:	1✔
142	params = {}	1✔
143	if next_page_token:	1✔
144	params["pageToken"] = next_page_token	1✔
145	response = requests.get(Airspace.data_url, params=params, timeout=5)	1✔
146	if response.status_code != 200:	1✔
NEW 147	return Airspace.data_cache	×
148	data = response.json()	1✔
149	for item in data.get("items", []):	1✔
150	if "_asp.xml" in item["name"] and item["size"] != "0":	1✔
151	all_airspaces.append((item["name"], item["size"]))	1✔
152	next_page_token = response.json().get("nextPageToken")	1✔
153	if not next_page_token:	1✔
154	break	1✔
155	return all_airspaces	1✔
156	except requests.exceptions.RequestException:	×
157	return Airspace.data_cache	×
158
159
160	def update_airspace(force_download=False, countries=None):	1✔
161	"""
162	Downloads the requested airspaces from their respective country code if it is over a month old
163	"""
164	if countries is None:	1✔
UNCOV 165	countries = ["de"]	×
166
167	for country in countries:	1✔
168	location = AIPDIR / f"{country}_asp.xml"	1✔
169	url = Airspace.data_download_url.format(country)	1✔
170	available = get_available_airspaces()	1✔
171	try:	1✔
172	data = [airspace for airspace in available if airspace[0].startswith(country)][0]	1✔
UNCOV 173	except IndexError:	×
174	logging.info("countries: %s not exists", ' '.join(countries))	×
175	continue
176	file_exists = location.exists()	1✔
177
178	is_outdated = file_exists and (time.time() - os.path.getmtime(str(location.resolve()))) > 60 * 60 * 24 * 30	1✔
179
180	if (force_download or is_outdated or not file_exists) \	1✔
181	and QtWidgets.QMessageBox.question(
182	None, "Allow download",
183	f"The selected {country} airspace needs to be downloaded "
184	f"({humanfriendly.format_size(int(data[-1]))})\nIs now a good time?",
185	QtWidgets.QMessageBox.Yes \| QtWidgets.QMessageBox.No,
186	QtWidgets.QMessageBox.Yes) \
187	== QtWidgets.QMessageBox.Yes:
188	download_progress(str(location.resolve()), url)	1✔
189
190
191	def get_airspaces(countries=None):	1✔
192	"""
193	Gets the .xml files in ~/.config/msui/downloads/aip and returns all airspaces within
194	"""
195	if countries is None:	1✔
UNCOV 196	countries = []	×
197	_airspaces = Airspace().data	1✔
198	_airspaces_mtime = Airspace().data_mtime	1✔
199
200	reload = False	1✔
201	files = [f"{country}_asp.xml" for country in countries]	1✔
202	update_airspace(countries=countries)	1✔
203	files = [file for file in files if (AIPDIR / file).exists()]	1✔
204
205	if _airspaces and len(files) == len(_airspaces_mtime):	1✔
UNCOV 206	for file in files:	×
207	if file not in _airspaces_mtime or \	×
208	os.path.getmtime(str(AIPDIR / file)) != _airspaces_mtime[file]:
UNCOV 209	reload = True	×
210	break	×
211	if not reload:	×
212	return _airspaces	×
213
214	_airspaces_mtime = {}	1✔
215	_airspaces = []	1✔
216	for file in files:	1✔
217	fpath = str(AIPDIR / file)	1✔
218	root = etree.parse(fpath).getroot()	1✔
219	valid_file = len(set([elem.tag for elem in root.iter()])) == 12	1✔
220	if valid_file:	1✔
221	airspaces = (root.find('{https://www.openaip.net}AIRSPACES'))	1✔
222	names = [dat.text for dat in airspaces.findall(".//{https://www.openaip.net}NAME")]	1✔
223	polygons = [dat.text for dat in airspaces.findall(".//{https://www.openaip.net}POLYGON")]	1✔
224	countries = [dat.text for dat in airspaces.findall(".//{https://www.openaip.net}COUNTRY")]	1✔
225	tops = []	1✔
226	top_units = []	1✔
227	for dat in airspaces.findall(".//{https://www.openaip.net}ALTLIMIT_TOP"):	1✔
228	unit = dat[0]	1✔
229	top_units.append(unit.get("UNIT"))	1✔
230	_, data = dat.iter()	1✔
231	tops.append(float(data.text))	1✔
232	bottoms = []	1✔
233	bottom_units = []	1✔
234	for dat in airspaces.findall(".//{https://www.openaip.net}ALTLIMIT_BOTTOM"):	1✔
235	unit = dat[0]	1✔
236	bottom_units.append(unit.get("UNIT"))	1✔
237	_, data = dat.iter()	1✔
238	bottoms.append(float(data.text))	1✔
239
240	for index, value in enumerate(names):	1✔
241	airspace_data = {	1✔
242	"name": names[index],
243	"polygon": polygons[index],
244	"top": tops[index],
245	"top_unit": top_units[index],
246	"bottom": bottoms[index],
247	"bottom_unit": bottom_units[index],
248	"country": countries[index]
249	}
250
251	# Convert to kilometers
252	airspace_data["top"] /= 3281 if airspace_data["top_unit"] == "F" else 32.81	1✔
253	airspace_data["bottom"] /= 3281 if airspace_data["bottom_unit"] == "F" else 32.81	1✔
254	airspace_data["top"] = round(airspace_data["top"], 2)	1✔
255	airspace_data["bottom"] = round(airspace_data["bottom"], 2)	1✔
256	airspace_data.pop("top_unit")	1✔
257	airspace_data.pop("bottom_unit")	1✔
258
259	airspace_data["polygon"] = [(float(data.split()[0]), float(data.split()[-1]))	1✔
260	for data in airspace_data["polygon"].split(",")]
261	_airspaces.append(airspace_data)	1✔
262	_airspaces_mtime[file] = os.path.getmtime(os.path.join(AIPDIR, file))	1✔
263	else:
264	QtWidgets.QMessageBox.information(None, "No Airspaces data in file:", f"{file}")	1✔
265
266	return _airspaces	1✔

Open-MSS / MSS / 16371583884

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous