From d876b1bb5fa2cfde1f9a947aacf65d769996584c Mon Sep 17 00:00:00 2001 From: Michael Clemens Date: Wed, 20 Jul 2022 15:43:24 +0200 Subject: [PATCH] first commit --- LICENSE.md | 19 +++ README.md | 70 +++++++++ setup.cfg | 31 ++++ setup.py | 3 + src/dxpager/__init__.py | 3 + src/dxpager/__main__.py | 319 ++++++++++++++++++++++++++++++++++++++++ 6 files changed, 445 insertions(+) create mode 100644 LICENSE.md create mode 100644 README.md create mode 100644 setup.cfg create mode 100644 setup.py create mode 100644 src/dxpager/__init__.py create mode 100755 src/dxpager/__main__.py diff --git a/LICENSE.md b/LICENSE.md new file mode 100644 index 0000000..15ab875 --- /dev/null +++ b/LICENSE.md @@ -0,0 +1,19 @@ +Copyright (c) 2022 Michael Clemens, DK1MI + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..9b71fae --- /dev/null +++ b/README.md @@ -0,0 +1,70 @@ +# DXPager + +This script is a bot that sends you DAPNET messages if a DX station is spotted whose entity you have not +worked/confirmed before. To achieve this, it + * parses the output of a specific dx cluster server + * downloads your LotW QSL file + * determines the DX station's country + * determines the DX station's continent + * determines if the DX station uses LotW + * determines if the DX station's country has been confirmed via LotW + * and finally - if it's a new DXCC - sends the information to your dapnet pager + +# Limitations + +The following limitations are present: + + * read-only: you can't send commands to the dx cluster server via this tool + * no filters: you need to configure your filter on the server + +# Installation + +DXPager needs Python 3 and the following libraries: + + * requests + +Furthermore, you need an account at LotW and hampager.de + +Before installing DXPager, please make sure that pip, setuptools and wheel are installed and up-to-date: + +``` +# python3 -m pip install --upgrade pip setuptools wheel +``` + +Finally, install ColorSpot with pip: + +``` +# python3 -m pip install dxpager +``` + +# Updating + +To update dxpager, execute the following command: + +``` +# python3 -m pip install --upgrade dxpager +``` + +# Usage + + * execute the application with "dxpager" + * DXPager creates a default config file and states its location (e.g. _~/.config/dxpager/dxpager.ini_) + * adapt _~/.config/dxpager/dxpager.ini_ to your needs. Important setting are: + * cluster/host and cluster/port: Change this if you want to use another cluster server + * cluster/user: Enter here your call sign + * lotw/user: Enter here your lotw user name (your call sign). Leave at "N0CALL" to disable this feature. + * lotw/password: Enter here your lotw password + * lotw/mode: Enter here the mode you would like to filter the QSL download from LotW + * dapnet_user: Enter here the hampager.de user name (your call sign) + * dapnet_pass: Enter here your hampager.de password + * dapnet_callsigns: Enter here the call sign of the receiver + * dapnet_txgroup: Adapt the tx group to your region + * execute the application again with "dxpager" + * the software now tries to download the following files and stores them into the configuration directory: + * https://www.country-files.com/bigcty/download/bigcty.zip (will be extracted) + * https://lotw.arrl.org/lotw-user-activity.csv + * https://lotw.arrl.org/lotwuser/lotwreport.adi?login={}&password={}&qso_query=1&qso_qsl=yes&qso_mode={}&qso_qsldetail=yes&qso_qslsince=1970-01-01 + +# License + +see ![LICENSE](LICENSE) diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..7a8a9d8 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,31 @@ +[metadata] +name = dxpager +version = 0.1.0 +author = Michael Clemens +author_email = dxpager@qrz.is +description = A DAPNET notification bot that alerts you if new DXCCs show up on the cluster +long_description = file: README.md +long_description_content_type = text/markdown +url = https://codeberg.org/mclemens/dxpager +project_urls = + Bug Tracker = https://codeberg.org/mclemens/dxpager/issues +classifiers = + Programming Language :: Python :: 3 + License :: OSI Approved :: MIT License + Operating System :: OS Independent + + +[options] +package_dir = + = src +packages = find: +python_requires = >=3.6 +install_requires= + requests + +[options.packages.find] +where = src + +[options.entry_points] +console_scripts = + colorspot = dxpager.__main__:main diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..f13cf21 --- /dev/null +++ b/setup.py @@ -0,0 +1,3 @@ +from setuptools import setup +if __name__ == '__main__': + setup() diff --git a/src/dxpager/__init__.py b/src/dxpager/__init__.py new file mode 100644 index 0000000..1b9bed9 --- /dev/null +++ b/src/dxpager/__init__.py @@ -0,0 +1,3 @@ +from dxpager.__main__ import DXPager + +__version__ = '0.1.0' diff --git a/src/dxpager/__main__.py b/src/dxpager/__main__.py new file mode 100755 index 0000000..86cdd5e --- /dev/null +++ b/src/dxpager/__main__.py @@ -0,0 +1,319 @@ +#!/usr/bin/env python3 + +# pylint: disable=W1401,C0303 +# pylint: disable=consider-using-f-string + + +""" ++---------------------------------------------------------------------+ +| | +| ____ _ __ ____ | +| / __ \ |/ // __ \____ _____ ____ _____ | +| / / / / // /_/ / __ `/ __ `/ _ \/ ___/ | +| / /_/ / |/ ____/ /_/ / /_/ / __/ / | +| /_____/_/|_/_/ \__,_/\__, /\___/_/ | +| -= DK1MI =- /____/ | +| | +| | +| A DAPNET bot that alerts you when stations pop up on a DX cluster | +| that are new ones | +| | +| Author: Michael Clemens, DK1MI (dxpager@qrz.is) | +| | +| Documentation: Please see the README.md file | +| License: Please see the LICENSE file | +| Repository: https://codeberg.org/mclemens/dxpager | +| | ++---------------------------------------------------------------------+ +""" + +import sys +import csv +import re +#import random +import os +from os.path import exists +import configparser +import zipfile +from telnetlib import Telnet +from pathlib import Path +import json +import requests +from requests.auth import HTTPBasicAuth + + +class DXPager(): + """DXPager class""" + + def __init__(self): + """initialize things""" + + self.print_banner() + + self.config = configparser.ConfigParser() + self.home_dir = str(Path.home()) + self.config_dir = self.home_dir + "/.config/dxpager/" + # Check if config directory exists and else create it + Path(self.config_dir).mkdir(parents=True, exist_ok=True) + self.config_file = os.path.expanduser(self.config_dir + 'dxpager.ini') + self.read_config(self.config, self.config_file) + + self.check_files() + + if self.config['lotw']['user'] != "N0CALL" and self.check_lotw_confirmed: + self.confirmed_entities = self.get_confirmed_entities() + + if self.check_cty: + with open(self.config_dir + self.config['files']['cty'], encoding='us-ascii') as csvfile: + self.cty = list(csv.reader(csvfile, delimiter=',')) + + if self.check_lotw_activity: + with open(self.config_dir + self.config['files']['lotw_activity'], encoding='us-ascii') as csvfile: + self.lotw_activity = list(csv.reader(csvfile, delimiter=',')) + + + @staticmethod + def print_banner(): + """print an awesome banner""" + print(" ____ _ __ ____ ") + print(" / __ \ |/ // __ \____ _____ ____ _____ ") + print(" / / / / // /_/ / __ `/ __ `/ _ \/ ___/ ") + print(" / /_/ / |/ ____/ /_/ / /_/ / __/ / ") + print(" /_____/_/|_/_/ \__,_/\__, /\___/_/ ") + print(" -= DK1MI =- /____/ ") + print("") + + + @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['dapnet'] = { + 'dapnet_user': 'N0CALL', + 'dapnet_pass': 'xxxxxxxxxxxxxxxxxxxx', + 'dapnet_url': 'http://www.hampager.de:8080/calls', + 'dapnet_callsigns': 'N0CALL', + 'dapnet_txgroup': 'dl-all'} + config['files'] = { + 'cty': 'cty.csv', + 'cty_url': 'https://www.country-files.com/bigcty/download/bigcty.zip', + 'lotw_confirmed': 'lotw.adi', + 'lotw_activity': 'lotw-user-activity.csv', + 'lotw_activity_url': 'https://lotw.arrl.org/lotw-user-activity.csv'} + config['lotw'] = { + 'user': 'N0CALL', + 'password': 'CHANGEME', + 'mode': 'ssb'} + + with open(file_name, 'w', encoding='us-ascii') 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 + + @staticmethod + def download_file(url, local_filename): + """downloads a file via HTTP and saves it to a defined file""" + with requests.get(url, stream=True) as request: + request.raise_for_status() + with open(local_filename, 'wb') as file: + for chunk in request.iter_content(chunk_size=8192): + file.write(chunk) + return local_filename + + + def check_files(self): + """Checks if all necessary files are in the file system. + Downloads all files and unzips them (if necessary)""" + + # check for lotw qsl information file + if self.config['lotw']['user'] != "N0CALL": + self.check_lotw_confirmed = exists(self.config_dir + self.config['files']['lotw_confirmed']) + if not self.check_lotw_confirmed: + print("The file " + self.config_dir + self.config['files']['lotw_confirmed'] + " is missing.") + user = self.config['lotw']['user'] + password = self.config['lotw']['password'] + mode = self.config['lotw']['mode'] + url = "https://lotw.arrl.org/lotwuser/lotwreport.adi?login={}&password={}"\ + "&qso_query=1&qso_qsl=yes&qso_mode={}&qso_qsldetail=yes&"\ + "qso_qslsince=1970-01-01".format(user, password, mode) + print("Trying to download " + url) + self.download_file(url, self.config_dir + self.config['files']['lotw_confirmed']) + self.check_lotw_confirmed = exists(self.config_dir + self.config['files']['lotw_confirmed']) + if self.check_lotw_confirmed: + print("File successfully downloaded") + else: + print("something went wrong while downloading " + url) + else: + self.check_lotw_confirmed = False + + # check for cty.csv file + self.check_cty = exists(self.config_dir + self.config['files']['cty']) + if not self.check_cty: + url = self.config['files']['cty_url'] + print("The file " + self.config_dir + self.config['files']['cty'] + " is missing.") + print("Trying to download " + url) + zip_name = self.download_file(url, self.config_dir + "bigcty.zip" ) + with zipfile.ZipFile(zip_name, 'r') as zip_ref: + zip_ref.extract("cty.csv", path=self.config_dir) + os.remove(zip_name) + self.check_cty = exists(self.config_dir + 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_dir + self.config['files']['lotw_activity']) + if not self.check_lotw_activity: + url = self.config['files']['lotw_activity_url'] + print("The file " + self.config_dir + self.config['files']['lotw_activity'] + " is missing.") + print("Trying to download " + url) + self.download_file(url, self.config_dir + self.config['files']['lotw_activity']) + self.check_lotw_activity = exists(self.config_dir + self.config['files']['lotw_activity']) + if self.check_lotw_activity: + print("File successfully downloaded") + else: + print("something went wrong while downloading " + url) + + + def get_confirmed_entities(self): + """Reads the file downloaded from LotW with all confirmed QSOs, + extracts all confirmed DXCCs and puts them into a list""" + ret = [] + with open(self.config_dir + self.config['files']['lotw_confirmed'], \ + encoding='us-ascii') as file: + for row in file: + if re.search("")[2].lower().rstrip() + if dxcc not in ret: + ret.append(dxcc) + return ret + + + def check_lotw(self, call): + """Reads the LotW user activity file and returns the date + of the last upload date if a specific call sign""" + ret = "" + for row in self.lotw_activity: + if call == row[0]: + ret = row[1] + return ret + return ret + + + def get_cty_row(self, call): + """Parses all CTY records, tries to find the DXCC entity of a + specific call sign and returns the line as a list of strings""" + done = False + while not done: + for row in self.cty: + entities = row[9].replace(";", "").replace("=", "").split(" ") + # TODO: Check if it is a speciall call (=) and mark it in the list + for prefix in entities: + if call == prefix: + return row + call = call[:-1] + if call == "": + return ["-", "-", "-", "-", "-", "-", "-"] + return None + + + def get_spots(self): + """Connects to the specified telnet dx cluster, performs a login, grabs the + output row by row, enriches it with data""" + 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 + try: + #line = line_enc.decode('ascii') + line = line_enc.decode('utf-8') + except: + print("Error while encoding the following line:") + print(line_enc) + # Enters the call sign if requested + if "enter your call" in line: + b_user = str.encode(self.config['cluster']['user']+"\n") + telnet.write(b_user) + # Detects the beginning of the stream + elif " Hello " in line: + print(line) + # This is true for every line representing a spot + elif "DX de" in line or "Dx de" in line: + try: + # Extract all necessary fields from the line and store them + # into different variables. + #print(line) + 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] + time = re.search(' (\d{4})Z', line).group(1) + comment = re.search(call_dx + ' +(.+?) +' + time, line).group(1) + + # If the CTY file is available, further information will be + # gathered from it, e.g. continent, country, dxcc ID + if self.check_cty: + cty_details = self.get_cty_row(call_dx) + else: + cty_details = ["-","-","-","-","-","-","-","-","-","-"] + + areaname = cty_details[1] + continent = cty_details[3] + + # If the LotW user activity file is available and the call + # sign in question is actually a LotW user, we'll add a (L) + # to the message + if self.check_lotw_activity and self.check_lotw(call_dx): + lotw = " (L) " + else: + lotw = " " + + # Removes the trailing .0 from a frequency for better readability + freq = freq.replace('.0', '') + + msg = "{}: {} de {} {} {}({}){}{}"\ + .format(time, call_dx, call_de, freq, areaname, continent, lotw, comment) + dapnet_json = json.dumps({"text": msg, "callSignNames": \ + [self.config['dapnet']['dapnet_callsigns']], \ + "transmitterGroupNames": [self.config['dapnet']['dapnet_txgroup']], \ + "emergency": False}) + # If the DX station's entity hasn't been worked/confirmed via + # LotW yet, the message will be sent to the dapnet API + if self.check_lotw_confirmed and self.config['lotw']['user'] != "N0CALL" \ + and cty_details[2] not in self.confirmed_entities: + response = requests.post(self.config['dapnet']['dapnet_url'], \ + data=dapnet_json, auth=HTTPBasicAuth(\ + self.config['dapnet']['dapnet_user'],\ + self.config['dapnet']['dapnet_pass'])) + print("!!! Sent to DAPNET: {} Response: {}".format(msg, response)) + else: + print(" Not sent to DAPNET: {}".format(msg)) + + except AttributeError: + print(line) + +def main(): + """main routine""" + try: + dx_pager = DXPager() + dx_pager.get_spots() + except KeyboardInterrupt: + sys.exit(0) + +if __name__ == "__main__": + try: + sys.exit(main()) + except EOFError: + pass