Created
May 25, 2023 10:41
-
-
Save sbrodehl/60d301e4cf55e9d6731f79c88c6103bf to your computer and use it in GitHub Desktop.
Update FlarmID from United FlarmNet data.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from pathlib import Path | |
import urllib.request | |
import flarmnet | |
UNITED_FLARMNET_URL = "https://turbo87.github.io/united-flarmnet/united.fln" | |
if __name__ == '__main__': | |
import argparse | |
parser = argparse.ArgumentParser() | |
parser.add_argument( | |
"participants", | |
type=Path, | |
help="CSV file with participants to fill with ID,TYPE,CN,CALL fields (*.csv).", | |
nargs="+", | |
) | |
parser.add_argument( | |
"--fln", | |
type=Path, | |
help="FlarmNet file e.g. https://turbo87.github.io/united-flarmnet/united.fln (*.fln).", | |
) | |
args = parser.parse_args() | |
if args.fln is None: | |
args.fln = Path("united.fln") | |
if not (args.fln.exists() and args.fln.is_file()): | |
print("Downloading united-flarmnet.fln ...", end='') | |
urllib.request.urlretrieve(UNITED_FLARMNET_URL, "united.fln") | |
args.fln = Path("united.fln") | |
print(" done!") | |
assert args.fln.exists() and args.fln.is_file() | |
assert len(args.participants) > 0 | |
for fp in args.participants: | |
assert fp.exists() and fp.is_file() | |
fln_lut = {} | |
with open(args.fln) as fh: | |
reader = flarmnet.Reader(fh) | |
for record in reader.read(): | |
if record.registration is not None: | |
fln_lut[record.registration.lower()] = record | |
for fp in args.participants: | |
print(f"Updating '{fp}':") | |
new_file = [] | |
with open(fp, encoding="latin-1") as fh: | |
l = fh.readline().strip() | |
header = list(map(str.strip, l.split(','))) | |
call_id = None | |
id_id = None | |
for h_idx, h in enumerate(header): | |
if h == "CALL": | |
call_id = h_idx | |
continue | |
if h == "ID": | |
id_id = h_idx | |
continue | |
new_file.append(l + "\n") | |
for l in fh.readlines(): | |
fields = list(map(str.strip, l.strip().split(','))) | |
call_sign = fields[call_id].lower() | |
if call_sign in fln_lut: | |
r = fln_lut[call_sign] | |
fields[id_id] = r.id | |
new_file.append(",".join(fields) + "\n") | |
print(f"{call_sign} updated!") | |
else: | |
new_file.append(l.strip() + "\n") | |
with open(fp, mode="wt", encoding="latin-1") as fh: | |
fh.writelines(new_file) | |
print(f"Wrote updated file {fp}.") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment