mirror of
https://codeberg.org/mclemens/colorspot.git
synced 2024-06-11 20:40:54 +00:00
299 lines
11 KiB
Python
Executable File
299 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
|
|
# Infos on colors: https://pypi.org/project/colored/
|
|
# Country list extracted from http://www.arrl.org/files/file/dxcclist.txt
|
|
|
|
|
|
import sys
|
|
import csv
|
|
import re
|
|
import json
|
|
import os
|
|
import time as bla
|
|
from telnetlib import Telnet
|
|
from colored import fg, bg, attr
|
|
import configparser
|
|
from collections import defaultdict
|
|
import random
|
|
import requests
|
|
from os.path import exists
|
|
import zipfile
|
|
|
|
class ColorSpot():
|
|
|
|
def __init__(self):
|
|
self.version = "0.1.0"
|
|
|
|
self.print_banner()
|
|
|
|
self.config = configparser.ConfigParser()
|
|
self.config_file = os.path.expanduser('~/.colorspot.ini')
|
|
self.read_config(self.config, self.config_file)
|
|
|
|
self.check_files()
|
|
if self.check_lotw_confirmed:
|
|
self.confirmed_entities = self.get_confirmed_entities()
|
|
|
|
|
|
if self.check_cty:
|
|
self.cty = list(csv.reader(open(self.config['files']['cty'], "r"), delimiter=","))
|
|
|
|
@staticmethod
|
|
def rnd_col():
|
|
r = lambda: random.randint(0,255)
|
|
return'#%02X%02X%02X' % (r(),r(),r())
|
|
|
|
@staticmethod
|
|
def download_file(url, local_filename):
|
|
#local_filename = url.split('/')[-1]
|
|
# NOTE the stream=True parameter below
|
|
with requests.get(url, stream=True) as r:
|
|
r.raise_for_status()
|
|
with open(local_filename, 'wb') as f:
|
|
for chunk in r.iter_content(chunk_size=8192):
|
|
# If you have chunk encoded response uncomment if
|
|
# and set chunk_size parameter to None.
|
|
#if chunk:
|
|
f.write(chunk)
|
|
return local_filename
|
|
|
|
def check_files(self):
|
|
# check for lotw qsl information file
|
|
self.check_lotw_confirmed = exists(self.config['files']['lotw_confirmed'])
|
|
if not self.check_lotw_confirmed:
|
|
print("The file " + self.config['files']['lotw_confirmed'] + " is missing.")
|
|
|
|
|
|
# check for cty.csv file
|
|
self.check_cty = exists(self.config['files']['cty'])
|
|
if not self.check_cty:
|
|
url = "https://www.country-files.com/bigcty/download/bigcty.zip"
|
|
print("The file " + self.config['files']['cty'] + " is missing.")
|
|
print("Trying to download " + url)
|
|
zip_name = self.download_file(url, "bigcty.zip" )
|
|
with zipfile.ZipFile(zip_name, 'r') as zip_ref:
|
|
zip_ref.extract("cty.csv")
|
|
os.remove(zip_name)
|
|
self.check_cty = exists(self.config['files']['cty'])
|
|
if self.check_cty:
|
|
print("File successfully downloaded and extracted.")
|
|
else:
|
|
print("something went wrong while downloading " + url)
|
|
|
|
# check for lotw user activity file
|
|
self.check_lotw_activity = exists(self.config['files']['lotw_activity'])
|
|
if not self.check_lotw_activity:
|
|
url = "https://lotw.arrl.org/lotw-user-activity.csv"
|
|
print("The file " + self.config['files']['lotw_activity'] + " is missing.")
|
|
print("Trying to download " + url)
|
|
file_name = self.download_file(url, self.config['files']['lotw_activity'])
|
|
self.check_lotw_activity = exists(self.config['files']['lotw_activity'])
|
|
if self.check_lotw_activity:
|
|
print("File successfully downloaded")
|
|
else:
|
|
print("something went wrong while downloading " + url)
|
|
|
|
|
|
|
|
def print_banner(self):
|
|
"""print an awesome banner"""
|
|
ver = self.version
|
|
# print the banner
|
|
print(fg(self.rnd_col())+" ___ _ ___ _ ")
|
|
print(fg(self.rnd_col())+" / __|___| |___ _ _/ __|_ __ ___| |_ ")
|
|
print(fg(self.rnd_col())+" | (__/ _ \ / _ \ '_\__ \ '_ \/ _ \ _|")
|
|
print(fg(self.rnd_col())+" \___\___/_\___/_| |___/ .__/\___/\__|")
|
|
print(fg(self.rnd_col())+" -= DK1MI =- |_| ")
|
|
print("")
|
|
print(attr('reset'))
|
|
|
|
@staticmethod
|
|
def read_config(config, file_name):
|
|
"""reads the configuration from the config file or
|
|
creates a default config file if none could be found"""
|
|
if os.path.isfile(file_name):
|
|
config.read(file_name)
|
|
else:
|
|
config = configparser.ConfigParser()
|
|
config['cluster'] = {
|
|
'host': 'dxc.nc7j.com',
|
|
'port': '7373',
|
|
'user': 'N0CALL',
|
|
'timeout': '100'}
|
|
config['files'] = {
|
|
'cty': 'cty.csv',
|
|
'lotw_confirmed': 'lotw.adi',
|
|
'lotw_activity': 'lotw-user-activity.csv'}
|
|
config['lotw'] = {
|
|
'user': 'N0CALL',
|
|
'password': 'XXXXXXXXX',
|
|
'mode': 'ssb'}
|
|
config['band_colors'] = {
|
|
"145": "white",
|
|
"144": "white",
|
|
"50": "white",
|
|
"29": "yellow",
|
|
"28": "yellow",
|
|
"24": "red",
|
|
"21": "orchid",
|
|
"18": "green",
|
|
"14": "steel_blue_3",
|
|
"10": "orange_1",
|
|
"7": "cyan",
|
|
"5": "white",
|
|
"3": "light_cyan",
|
|
"1": "white",
|
|
'alert_bg': 'indian_red_1a',
|
|
'alert_fg': 'white',
|
|
'default_bg': 'black'}
|
|
config['cont_colors'] = {
|
|
"AF": "light_salmon_3b",
|
|
"AN": "white",
|
|
"AS": "orange_red_1",
|
|
"EU": "cyan",
|
|
"NA": "steel_blue_3",
|
|
"OC": "orchid",
|
|
"SA": "light_goldenrod_2a"}
|
|
config['colors']= {
|
|
'use_colors': 'yes',
|
|
'color_by' : 'continent',
|
|
'alert_bg': 'indian_red_1a',
|
|
'alert_fg': 'white',
|
|
'default_bg': 'black'}
|
|
|
|
with open(file_name, 'w') as configfile:
|
|
config.write(configfile)
|
|
print("\nNo configuration file found. A new configuration file has been created.")
|
|
print("\nPlease edit the file " + file_name + " and restart the application.\n" )
|
|
sys.exit()
|
|
return config
|
|
|
|
def get_confirmed_entities(self):
|
|
ret = []
|
|
#TODO: download file and/or tell user what to do
|
|
file = open(self.config['files']['lotw_confirmed'], "r")
|
|
for row in file:
|
|
if re.search("<DXCC:", row):
|
|
dxcc = row.partition(">")[2].lower().rstrip()
|
|
if dxcc not in ret:
|
|
ret.append(dxcc)
|
|
return ret
|
|
|
|
def check_lotw(self, call):
|
|
ret = ""
|
|
#TODO: download file and/or tell user what to do
|
|
csv_file = csv.reader(open(self.config['files']['lotw_activity'], "r"), delimiter=",")
|
|
#loop through the csv file
|
|
for row in csv_file:
|
|
if call == row[0]:
|
|
ret = row[1]
|
|
return ret
|
|
return ret
|
|
|
|
def get_cty_row(self, call):
|
|
done = False
|
|
while not done:
|
|
for row in self.cty:
|
|
entities = row[9].replace(";", "").replace("=", "").split(" ")
|
|
# TODO: schauen ob = davor und match -> als special call anzeigen
|
|
for prefix in entities:
|
|
if call == prefix:
|
|
return row
|
|
call = call[:-1]
|
|
if call == "":
|
|
return ["-", "-", "-", "-", "-", "-", "-"]
|
|
return None
|
|
|
|
|
|
def get_spots(self):
|
|
with Telnet(self.config['cluster']['host'], int(self.config['cluster']['port']), \
|
|
int(self.config['cluster']['timeout'])) as telnet:
|
|
while True:
|
|
line_enc = telnet.read_until(b"\n") # Read one line
|
|
line = line_enc.decode('ascii')
|
|
if "enter your call" in line:
|
|
b_user = str.encode(self.config['cluster']['user']+"\n")
|
|
telnet.write(b_user)
|
|
elif " Hello " in line:
|
|
print(fg("grey_27") + line + attr("reset"))
|
|
foreground = "white"
|
|
background = "grey_27"
|
|
sep = fg("grey_27")+'|'+fg(foreground)
|
|
row = ["DE", sep, "Freq", sep, "DX", \
|
|
sep, "Country", sep, "C", sep, "L", sep, "Comment", sep, "Time"]
|
|
|
|
print(bg(background) + fg(foreground) + \
|
|
'{:9.9} {:<1} {:>7.7} {:<1} {:<10.10} {:<1} {:<16.16} {:<1} {:<2.2} {:<1} {:<1.1} {:<1} {:<30.30} {:<1} {:<4.4}'.format(*row) + attr("reset"))
|
|
b_cmd = str.encode("sh/dx/50 @\n")
|
|
telnet.write(b_cmd)
|
|
elif "DX de" in line or "Dx de" in line:
|
|
try:
|
|
band_col = ""
|
|
call_de = re.search('D(X|x) de (.+?): ', line).group(2)
|
|
freq = re.search(': +(.+?) ', line).group(1)
|
|
call_dx = re.search(freq + ' +(.+?) ', line).group(1)
|
|
time = re.search('[^ ]*$', line).group(0)[0:4]
|
|
comment = re.search(call_dx + ' +(.+?) +' + time, line).group(1)
|
|
|
|
if self.check_cty:
|
|
cty_details = self.get_cty_row(call_dx)
|
|
else:
|
|
cty_details = ["-","-","-","-","-","-","-","-","-","-"]
|
|
|
|
areaname = cty_details[1]
|
|
continent = cty_details[3]
|
|
|
|
if self.check_lotw_activity and self.check_lotw(call_dx):
|
|
lotw = "✓"
|
|
else:
|
|
lotw = ""
|
|
|
|
try:
|
|
if self.config['colors']['color_by'] == "band":
|
|
foreground = self.config['band_colors'][freq[:-5]]
|
|
elif self.config['colors']['color_by'] == "continent":
|
|
foreground = self.config['cont_colors'][continent]
|
|
else:
|
|
foreground = "white"
|
|
except Exception:
|
|
foreground = "white"
|
|
|
|
freq = freq.replace('.0', '')
|
|
|
|
if self.check_lotw_confirmed and cty_details[2] not in self.confirmed_entities:
|
|
background = self.config['colors']['alert_bg']
|
|
foreground = self.config['colors']['alert_fg']
|
|
else:
|
|
background = self.config['colors']['default_bg']
|
|
|
|
sep = fg("grey_27")+'|'+fg(foreground)
|
|
|
|
row = [call_de, sep, freq, sep, call_dx, \
|
|
sep, areaname, sep, continent, sep, lotw, sep, comment, sep, time]
|
|
|
|
print(bg(background) + fg(foreground) + \
|
|
'{:9.9} {:<1} {:>7.7} {:<1} {:<10.10} {:<1} {:<16.16} {:<1} {:<2.2} {:<1} {:<1.1} {:<1} {:<30.30} {:<1} {:<4.4}'.format(*row) + attr("reset"))
|
|
|
|
except AttributeError:
|
|
print(line)
|
|
|
|
#####################################################
|
|
# Main Routine #
|
|
#####################################################
|
|
def main():
|
|
try:
|
|
color_spot = ColorSpot()
|
|
color_spot.get_spots()
|
|
except KeyboardInterrupt:
|
|
sys.exit(0) # or 1, or whatever
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except EOFError:
|
|
pass
|
|
|
|
|
|
|