upload.py

from influxdb_client import InfluxDBClient, Point, WritePrecision
from datetime import timedelta, datetime, timezone
from influxdb_client.client.write_api import SYNCHRONOUS

import os
import json
import glob
import logging

def get_latest_log():
    """
    retrieves the latest log file generated by Faros
    """
    log_dir_path = f'{os.getenv("SystemDrive")}/Users/{os.getlogin()}/AppData/Roaming/Afreet/Products/Faros/BeaconLogs'
    dir_names = os.listdir(log_dir_path)
    dir_dates = [int(i.replace('-', '')) for i in dir_names]
    dir_dict = dict(zip(dir_dates, dir_names))
    latest_dir_date = max(dir_dict.keys())
    latest_dir_name = dir_dict[latest_dir_date]
    os.chdir(log_dir_path+'/'+latest_dir_name)
    file_names = glob.glob('*.log')
    file_dates = [int(i.replace('-', '').replace('.log','')) for i in file_names]
    file_dict = dict(zip(file_dates, file_names))
    latest_date = max(file_dict.keys())
    latest_file = file_dict[latest_date]
    logging.info(f'checking points from {latest_file}')
    log_file = log_dir_path+'/'+latest_dir_name+'/'+latest_file
    return log_file

def ingest(file):
    date = datetime.now().replace(tzinfo=timezone.utc)
    latitude, longitude = 0.0, 0.0
    points = []
    with open(file, 'r') as f:
        lines = f.readlines()
        for l in lines:
            if l.startswith(';'):
                parts = [s.strip() for s in l[1:].split('\t')]
                for p in parts:
                    try:
                        segs = p.split('=')
                        k, v = segs[0], segs[-1]
                        if k.startswith('DATE'):
                            date = datetime.strptime(v, '%Y-%m-%d').replace(tzinfo=timezone.utc)
                        elif p.startswith('LAT'):
                            latitude = int(v)
                        elif p.startswith('LON'):
                            longitude = int(v)
                    except Exception as ex:
                        logging.info(ex)
                continue
            parts = l.split('\t')
            try:
                [time, mhz, call, snr, db, evidence, delay] = parts
                tt = datetime.strptime(time, '%H:%M:%S')
                time_point = date + timedelta(hours=tt.hour, minutes=tt.minute, seconds=tt.second)
                points.append({
                    "timestamp": time_point,
                    "mhz": int(mhz),
                    "call": call.strip(),
                    "snr": float(snr),
                    "qsb": int(db),
                    "evidence": float(evidence),
                    "valid": float(evidence) >= 1.0,
                    "delay": int(delay),
                    "lat": latitude,
                    "lon": longitude
                })
            except Exception as ex:
                logging.info(ex)
    return points

def init_db():
    client = InfluxDBClient(url=URL, token=TOKEN, bucket=BUCKET, org=ORG, port=PORT)
    return client

def upload(client: InfluxDBClient, points):
    records = []
    api = client.write_api(write_options=SYNCHRONOUS)
    for p in points:
        if not p['valid']:
            continue
        pp = Point('measurement').tag('call', p['call']).tag('mhz', p['mhz']) \
            .tag('lat', p['lat']) \
            .tag('long', p['lon']) \
            .tag('valid', p['valid']) \
            .field('snr', p['snr']) \
            .field('delay', p['delay']) \
            .field('evidence', p['evidence']) \
            .field('qsb', p['qsb']).time(time=p['timestamp'], write_precision=WritePrecision.S)
        if p['call'] in geo_hash_table:
            pp = pp.tag('geohash', geo_hash_table[p['call']]['geohash'])
        records.append(pp)
    api.write(bucket=BUCKET, org=ORG, record=records)
    logging.info(f'uploaded {len(records)} points')

if __name__ == '__main__':
    far_flux_dir = f'{os.getenv("SystemDrive")}/Users/{os.getlogin()}/AppData/Local/Programs/FarFlux/'
    os.chdir(far_flux_dir)

    with open('geohash.json') as file:
        geo_hash_table = json.load(file)
        
    with open('config.json') as file:
        config = json.load(file)

    logging.basicConfig(handlers=[logging.FileHandler(
        filename=f'upload.log', encoding='utf-8', mode='a+')],
        format='%(asctime)s >> %(message)s',
        level=logging.INFO)

    LOG_FILE = get_latest_log()
    URL = config["URL"]
    ORG = config["ORG_ID"]
    BUCKET = config["BUCKET"]
    TOKEN = config["TOKEN"]
    PORT = config["PORT"]
    points = ingest(LOG_FILE)
    client = init_db()
    upload(client, points)