diff --git a/.github/workflows/debian.yml b/.github/workflows/debian.yml index 73806e3..bc170fb 100644 --- a/.github/workflows/debian.yml +++ b/.github/workflows/debian.yml @@ -30,9 +30,10 @@ jobs: - name: Build Debian/Apt bdist_deb run: | + export REPO_NAME=$(echo ${github.repository} | awk -F"/" '{print $2}') python3 setup.py --command-packages=stdeb.command bdist_deb ls -al deb_dist/ - cp deb_dist/python3-aprscot_*_all.deb deb_dist/python3-aprscot_latest_all.deb + cp deb_dist/python3-${REPO_NAME}_*_all.deb deb_dist/python3-${REPO_NAME}_latest_all.deb - uses: actions/upload-artifact@master with: diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish_to_pypi.yml similarity index 55% rename from .github/workflows/python-publish.yml rename to .github/workflows/python-publish_to_pypi.yml index 0236000..2c719cf 100644 --- a/.github/workflows/python-publish.yml +++ b/.github/workflows/python-publish_to_pypi.yml @@ -1,9 +1,10 @@ # This workflow will upload a Python Package using Twine when a release is created -# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries +name: Publish package to PyPI -name: Publish package to PYPI - -on: push +on: + push: + tags: + - '*' jobs: deploy: @@ -12,23 +13,18 @@ jobs: steps: - uses: actions/checkout@v2 - - name: Set up Python uses: actions/setup-python@v2 with: python-version: '3.x' - - name: Install dependencies run: | - python -m pip install --upgrade pip - pip install setuptools wheel twine - + python3 -m pip install --upgrade pip + python3 -m pip install setuptools wheel twine - name: Build run: | - python setup.py sdist bdist_wheel - + python3 setup.py sdist bdist_wheel - name: Publish package - if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags') uses: pypa/gh-action-pypi-publish@release/v1 with: user: __token__ diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml deleted file mode 100644 index d12614d..0000000 --- a/.github/workflows/python-test.yml +++ /dev/null @@ -1,66 +0,0 @@ -# This workflow will install Python dependencies, run tests and lint with a -# single version of Python -# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions - -name: Lint, Style & Test - -on: - push: - branches: [ main ] - paths-ignore: - - ".gitignore" - - "*.md" - - "*.rst" - - "LICENSE" - - "requirements_test.txt" - pull_request: - branches: [ main ] - paths-ignore: - - ".gitignore" - - "*.md" - - "*.rst" - - "LICENSE" - - "requirements_test.txt" - -jobs: - test: - runs-on: ${{ matrix.os }} - strategy: - matrix: - os: [ubuntu-latest] - python-version: [3.6, 3.7, 3.8, 3.9, pypy3] - - steps: - - uses: actions/checkout@v2 - - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v2 - with: - python-version: ${{ matrix.python-version }} - - - name: Display Python version - run: python -c "import sys; print(sys.version)" - - - name: Upgrade PIP - run: | - python -m pip install --upgrade pip - - - name: Install Python testing dependencies - run: | - make install_requirements_test - - - name: Install Python program in 'develop' mode - run: | - make install_requirements develop - - - name: Lint with pylint - run: | - make lint - - - name: Style-check with flake8 - run: | - make flake8 - - - name: Test with pytest - run: | - make pytest diff --git a/.github/workflows/python-test_and_lint.yml b/.github/workflows/python-test_and_lint.yml new file mode 100644 index 0000000..455c927 --- /dev/null +++ b/.github/workflows/python-test_and_lint.yml @@ -0,0 +1,38 @@ +name: Lint & Test Code + +on: + push: + branches: [ main ] + pull_request: + branches: [ main ] + +jobs: + build: + + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: [3.6, 3.7, 3.8, 3.9] + + steps: + - uses: actions/checkout@v2 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v2 + with: + python-version: ${{ matrix.python-version }} + - name: Install test requirements + run: | + make install_test_requirements + - name: Install package itself (editable) + run: | + make editable + - name: Lint with pylint + run: | + make pylint + - name: Lint with flake8 + run: | + make flake8 + - name: Test with pytest-cov + run: | + make test_cov diff --git a/Makefile b/Makefile index c1ac92c..e68fbb9 100644 --- a/Makefile +++ b/Makefile @@ -1,76 +1,78 @@ -# Makefile for Python APRS Cursor-on-Target Gateway. # -# Source:: https://github.com/ampledata/aprscot +# Copyright 2022 Greg Albrecht +# +# 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. +# # Author:: Greg Albrecht W2GMD # Copyright:: Copyright 2022 Greg Albrecht # License:: Apache License, Version 2.0 # - +this_app = aprscot .DEFAULT_GOAL := all +all: editable -all: develop - -install_requirements: - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi +develop: + python3 setup.py develop -install_requirements_test: - pip install -r requirements_test.txt +editable: + python3 -m pip install -e . -develop: - pip install -e . +install_test_requirements: + python3 -m pip install -r requirements_test.txt install: - python setup.py install + python3 setup.py install uninstall: - pip uninstall -y aprscot + python3 -m pip uninstall -y $(this_app) reinstall: uninstall install -remember_test: - @echo - @echo "Hello from the Makefile..." - @echo "Don't forget to run: 'make install_requirements_test'" - @echo +publish: + python3 setup.py publish clean: @rm -rf *.egg* build dist *.py[oc] */*.py[co] cover doctest_pypi.cfg \ nosetests.xml pylint.log output.xml flake8.log tests.log \ - test-result.xml htmlcov fab.log .coverage */__pycache__/ - -# Publishing: + test-result.xml htmlcov fab.log .coverage __pycache__ \ + */__pycache__ -build: remember_test - python3 -m build --sdist --wheel - -twine_check: remember_test build - twine check dist/* - -upload: remember_test build - twine upload dist/* - -publish: build twine_check upload - -# Tests: - -pep8: remember_test - # flake8 --max-complexity 12 --exit-zero *.py */*.py - # stop the build if there are Python syntax errors or undefined names - flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics - # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide - flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics +pep8: + flake8 --max-line-length=88 --extend-ignore=E203 --exit-zero $(this_app)/*.py flake8: pep8 -lint: remember_test +lint: pylint --msg-template="{path}:{line}: [{msg_id}({symbol}), {obj}] {msg}" \ - -r n *.py */*.py || exit 0 + --max-line-length=88 -r n $(this_app)/*.py || exit 0 pylint: lint -pytest: remember_test +checkmetadata: + python3 setup.py check -s --restructuredtext + +mypy: + mypy --strict . + +pytest: pytest -test: lint pep8 pytest +test: editable install_test_requirements pytest + +test_cov: + pytest --cov=$(this_app) + +black: + black . diff --git a/aprscot/__init__.py b/aprscot/__init__.py index fb69009..2c2d653 100644 --- a/aprscot/__init__.py +++ b/aprscot/__init__.py @@ -1,26 +1,43 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # -*- coding: utf-8 -*- - -# APRS Cursor-on-Target Gateway. +# +# Copyright 2022 Greg Albrecht +# +# 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. +# +# Author:: Greg Albrecht W2GMD +# """ APRS Cursor-on-Target Gateway. ~~~~ - :author: Greg Albrecht W2GMD :copyright: Copyright 2022 Greg Albrecht :license: Apache License, Version 2.0 :source: - """ -from .constants import (LOG_FORMAT, LOG_LEVEL, DEFAULT_APRSIS_PORT, # NOQA - DEFAULT_COT_TYPE, DEFAULT_COT_STALE, - DEFAULT_APRSIS_HOST, DEFAULT_APRSIS_CALLSIGN, - DEFAULT_APRSIS_FILTER) - -from .functions import aprs_to_cot # NOQA +from .constants import ( # NOQA + LOG_FORMAT, + LOG_LEVEL, + DEFAULT_APRSIS_PORT, + DEFAULT_COT_TYPE, + DEFAULT_COT_STALE, + DEFAULT_APRSIS_HOST, + DEFAULT_APRSIS_CALLSIGN, + DEFAULT_APRSIS_FILTER, +) + +from .functions import aprs_to_cot, create_tasks # NOQA from .classes import APRSWorker # NOQA diff --git a/aprscot/classes.py b/aprscot/classes.py index 977a1b6..dbeb95c 100644 --- a/aprscot/classes.py +++ b/aprscot/classes.py @@ -1,94 +1,93 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # -*- coding: utf-8 -*- - -"""APRS Cursor-on-Target Class Definitions.""" +# +# Copyright 2022 Greg Albrecht +# +# 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. +# +# Author:: Greg Albrecht W2GMD +# + +"""APRSCOT Class Definitions.""" import asyncio import aprslib.parsing -import pytak +import pytak import aprscot __author__ = "Greg Albrecht W2GMD " __copyright__ = "Copyright 2022 Greg Albrecht" __license__ = "Apache License, Version 2.0" -__source__ = "https://github.com/ampledata/aprscot" -class APRSWorker(pytak.MessageWorker): +class APRSWorker(pytak.QueueWorker): """APRS Cursor-on-Target Worker Class.""" - def __init__(self, event_queue: asyncio.Queue, config: dict) -> None: - super().__init__(event_queue) - self.config = config - - # APRS Parameters: - self.passcode = "-1" - - self.callsign: str = config["aprscot"].get( - "CALLSIGN", aprscot.DEFAULT_APRSIS_CALLSIGN) - - self.aprs_filter: str = config["aprscot"].get( - "APRSIS_FILTER", aprscot.DEFAULT_APRSIS_FILTER) - - # Figure out APRS Host: - aprs_host: str = config["aprscot"].get( - "APRS_HOST", aprscot.DEFAULT_APRSIS_HOST) - aprs_port: str = config["aprscot"].get( - "APRS_PORT", aprscot.DEFAULT_APRSIS_PORT) - - if ':' in aprs_host: - aprs_host, aprs_port = aprs_host.split(':') - - self.aprs_host = aprs_host - self.aprs_port = aprs_port - self._logger.info( - "Using APRS-IS server: %s:%s", self.aprs_host, self.aprs_port) - - async def handle_message(self, message: bytes) -> None: + async def handle_data(self, data: bytes) -> None: """Handles messages from APRS Worker.""" - self._logger.debug("APRS message='%s'", message) + self._logger.debug("APRS data='%s'", data) + frame = None # Skip control messages from APRS-IS: - if b"# " in message[:2]: - self._logger.info("APRS-IS: '%s'", message) + if b"# " in data[:2]: + self._logger.info("APRS-IS: '%s'", data) return # Some APRS Frame types are not supported by aprslib yet: try: - aprs_frame = aprslib.parsing.parse(message) + frame = aprslib.parsing.parse(data) except aprslib.exceptions.UnknownFormat as exc: - self._logger.warning("Unhandled APRS Frame: '%s'", message) + self._logger.warning("Unhandled APRS Frame: '%s'", data) return except aprslib.exceptions.ParseError as exc2: - self._logger.warning("Invalid Format: '%s'", message) + self._logger.warning("Invalid Format: '%s'", data) return - # self._logger.debug("aprs_frame=%s", aprs_frame) + if not frame: + return - event = aprscot.aprs_to_cot(aprs_frame, self.config) + event: bytes = aprscot.aprs_to_cot(frame, self.config) if not event: - self._logger.warning( - "Empty CoT Event for APRS Frame: '%s'", aprs_frame.get("raw")) + self._logger.warning("Empty COT for frame: '%s'", frame.get("raw")) return - await self._put_event_queue(event) + await self.put_queue(event) - async def run(self): # pylint: disable=arguments-differ + async def run(self, number_of_iterations=-1): """Runs this Thread, Reads from Pollers.""" - self._logger.info("Running APRSWorker") + self._logger.info("Running %s", self.__class__) - reader, writer = await asyncio.open_connection( - self.aprs_host, int(self.aprs_port)) + aprs_host: str = self.config.get("APRS_HOST", aprscot.DEFAULT_APRSIS_HOST) + aprs_port: str = self.config.get("APRS_PORT", aprscot.DEFAULT_APRSIS_PORT) + if ":" in aprs_host: + aprs_host, aprs_port = aprs_host.split(":") + self._logger.info("Using APRS-IS server: %s:%s", aprs_host, aprs_port) + + reader, writer = await asyncio.open_connection(aprs_host, int(aprs_port)) + + # APRS Parameters: + passcode = "-1" + callsign: str = self.config.get("CALLSIGN", aprscot.DEFAULT_APRSIS_CALLSIGN) + aprs_filter: str = self.config.get( + "APRSIS_FILTER", aprscot.DEFAULT_APRSIS_FILTER + ) - _login = ( - f"user {self.callsign} pass {self.passcode} vers aprscot v5.0.0") + _login = f"user {callsign} pass {passcode} vers aprscot v6.0.0" - if self.aprs_filter: - self._logger.info("Using APRS Filter: '%s'", self.aprs_filter) - _login = f"{_login} filter {self.aprs_filter}" + if aprs_filter: + self._logger.info("Using APRS Filter: '%s'", aprs_filter) + _login = f"{_login} filter {aprs_filter}" _login = f"{_login}\r\n" b_login = bytes(_login, "UTF-8") @@ -96,6 +95,6 @@ async def run(self): # pylint: disable=arguments-differ await writer.drain() while 1: - frame = await reader.readline() - if frame: - await self.handle_message(frame) + data = await reader.readline() + if data: + await self.handle_data(data) diff --git a/aprscot/commands.py b/aprscot/commands.py index 8ad0bb7..86dc44a 100644 --- a/aprscot/commands.py +++ b/aprscot/commands.py @@ -1,84 +1,29 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # -*- coding: utf-8 -*- - -"""APRS Cursor-on-Target Gateway Commands.""" - -import argparse -import asyncio -import configparser -import logging -import sys -import urllib +# +# Copyright 2022 Greg Albrecht +# +# 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. +# +# Author:: Greg Albrecht W2GMD +# + +"""PyTAK Command Line.""" import pytak -import aprscot - -# Python 3.6 support: -if sys.version_info[:2] >= (3, 7): - from asyncio import get_running_loop -else: - from asyncio import _get_running_loop as get_running_loop # NOQA pylint: disable=no-name-in-module - - -__author__ = 'Greg Albrecht W2GMD ' -__copyright__ = 'Copyright 2022 Greg Albrecht' -__license__ = 'Apache License, Version 2.0' -__source__ = 'https://github.com/ampledata/aprscot' - - -async def main(config): - """Main program function. Sets Queues, Executes Workers.""" - tx_queue: asyncio.Queue = asyncio.Queue() - rx_queue: asyncio.Queue = asyncio.Queue() - - cot_url: urllib.parse.ParseResult = urllib.parse.urlparse( - config["aprscot"].get("COT_URL")) - - # Create our CoT Event Queue Worker - reader, writer = await pytak.protocol_factory(cot_url) - write_worker = pytak.EventTransmitter(tx_queue, writer) - read_worker = pytak.EventReceiver(rx_queue, reader) - - message_worker = aprscot.APRSWorker(tx_queue, config) - - await tx_queue.put(pytak.hello_event("aprscot")) - - done, _ = await asyncio.wait( - {message_worker.run(), read_worker.run(), write_worker.run()}, - return_when=asyncio.FIRST_COMPLETED) - - for task in done: - print(f"Task completed: {task}") - - -def cli(): - """Command Line interface for APRS Cursor-on-Target Gateway.""" - - # Get cli arguments: - parser = argparse.ArgumentParser() - - parser.add_argument("-c", "--CONFIG_FILE", dest="CONFIG_FILE", - default="config.ini", type=str) - namespace = parser.parse_args() - cli_args = {k: v for k, v in vars(namespace).items() if v is not None} - - # Read config file: - config = configparser.ConfigParser() - - config_file = cli_args.get("CONFIG_FILE") - logging.info("Reading configuration from %s", config_file) - config.read(config_file) - - if sys.version_info[:2] >= (3, 7): - asyncio.run(main(config), debug=config["aprscot"].getboolean("DEBUG")) - else: - loop = asyncio.get_event_loop() - try: - loop.run_until_complete(main(config)) - finally: - loop.close() +__author__ = "Greg Albrecht W2GMD " +__copyright__ = "Copyright 2022 Greg Albrecht" +__license__ = "Apache License, Version 2.0" -if __name__ == "__main__": - cli() +# PyTAK CLI tool boilerplate: +pytak.cli(__name__.split(".")[0]) diff --git a/aprscot/constants.py b/aprscot/constants.py index 0c15b88..46a9584 100644 --- a/aprscot/constants.py +++ b/aprscot/constants.py @@ -1,28 +1,44 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- - -"""APRS Cursor-on-Target Gatway: Constants.""" +# +# Copyright 2022 Greg Albrecht +# +# 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. +# +# Author:: Greg Albrecht W2GMD +# + +"""APRSCOT Constants.""" import logging import os import re -__author__ = 'Greg Albrecht W2GMD ' -__copyright__ = 'Copyright 2022 Greg Albrecht' -__license__ = 'Apache License, Version 2.0' -__source__ = 'https://github.com/ampledata/aprscot' +__author__ = "Greg Albrecht W2GMD " +__copyright__ = "Copyright 2022 Greg Albrecht" +__license__ = "Apache License, Version 2.0" -if bool(os.environ.get('DEBUG')): +if bool(os.environ.get("DEBUG")): LOG_LEVEL = logging.DEBUG LOG_FORMAT = logging.Formatter( - ('%(asctime)s aprscot %(levelname)s %(name)s.%(funcName)s:%(lineno)d ' - ' - %(message)s')) - logging.debug('aprscot Debugging Enabled via DEBUG Environment Variable.') + ( + "%(asctime)s aprscot %(levelname)s %(name)s.%(funcName)s:%(lineno)d " + " - %(message)s" + ) + ) + logging.debug("aprscot Debugging Enabled via DEBUG Environment Variable.") else: LOG_LEVEL = logging.INFO - LOG_FORMAT = logging.Formatter( - ('%(asctime)s aprscot - %(message)s')) + LOG_FORMAT = logging.Formatter(("%(asctime)s aprscot - %(message)s")) # 3833.55N/12248.93W LL_REX = re.compile( @@ -31,8 +47,8 @@ DEFAULT_APRSIS_PORT: int = 14580 DEFAULT_APRSIS_HOST: str = "rotate.aprs.net" -DEFAULT_APRSIS_CALLSIGN: str = "APRSCOT" +DEFAULT_APRSIS_CALLSIGN: str = "SUNSET" DEFAULT_APRSIS_FILTER: str = "m/50" DEFAULT_COT_TYPE: str = "a-f-G-I-U-T-r" -DEFAULT_COT_STALE: str = 3600 +DEFAULT_COT_STALE: str = "3600" diff --git a/aprscot/functions.py b/aprscot/functions.py index 7cf67c9..49c964f 100644 --- a/aprscot/functions.py +++ b/aprscot/functions.py @@ -1,14 +1,29 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # -*- coding: utf-8 -*- - -"""APRS Cursor-on-Target Gateway Functions.""" - -import datetime +# +# Copyright 2022 Greg Albrecht +# +# 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. +# +# Author:: Greg Albrecht W2GMD +# + +"""APRSCOT Functions.""" import xml.etree.ElementTree as ET -import pytak +from configparser import ConfigParser +from typing import Set, Union +import pytak import aprscot @@ -18,38 +33,55 @@ __source__ = "https://github.com/ampledata/aprscot" -def aprs_to_cot_xml(aprs_frame: dict, config: dict) -> ET.Element: # NOQA pylint: disable=too-many-locals,too-many-statements - """Converts an APRS Frame to a Cursor-on-Target Event.""" - time = datetime.datetime.now(datetime.timezone.utc) +def create_tasks( + config: ConfigParser, clitool: pytak.CLITool +) -> Set[pytak.Worker,]: + """ + Creates specific coroutine task set for this application. + + Parameters + ---------- + config : `ConfigParser` + Configuration options & values. + clitool : `pytak.CLITool` + A PyTAK Worker class instance. + + Returns + ------- + `set` + Set of PyTAK Worker classes for this application. + """ + return set([aprscot.APRSWorker(clitool.tx_queue, config)]) + - lat = aprs_frame.get("latitude") - lon = aprs_frame.get("longitude") +def aprs_to_cot_xml( + frame: dict, config: Union[dict, None] = None +) -> Union[ + ET.Element, None +]: # NOQA pylint: disable=too-many-locals,too-many-statements + """Converts an APRS Frame to a Cursor-on-Target Event.""" + lat = frame.get("latitude") + lon = frame.get("longitude") if not lat or not lon: return None - callsign = aprs_frame.get("from") - name = callsign + config: dict = config or {} + cot_stale = int(config.get("COT_STALE", pytak.DEFAULT_COT_STALE)) - if "aprscot" in config: - aprscot_conf = config["aprscot"] - else: - aprscot_conf = {} + callsign = frame.get("from").replace(" ", "") + name = callsign - cot_type = aprscot_conf.get("COT_TYPE", aprscot.DEFAULT_COT_TYPE) - _cot_stale = aprscot_conf.get("COT_STALE", aprscot.DEFAULT_COT_STALE) + cot_uid = f"APRS.{callsign}" + cot_type = config.get("COT_TYPE", aprscot.DEFAULT_COT_TYPE) - if 'sections' in config and callsign in config.sections(): - cs_conf = aprscot_conf[callsign] + if "sections" in config and callsign in config.sections(): + cs_conf = config[callsign] cot_type = cs_conf.get("COT_TYPE", cot_type) - _cot_stale = cs_conf.get("COT_STALE", _cot_stale) + cot_stale = cs_conf.get("COT_STALE", cot_stale) name = cs_conf.get("COT_NAME", name) cot_icon = cs_conf.get("COT_ICON") - cot_stale = (datetime.datetime.now(datetime.timezone.utc) + - datetime.timedelta( - seconds=int(_cot_stale))).strftime(pytak.ISO_8601_UTC) - point = ET.Element("point") point.set("lat", str(lat)) point.set("lon", str(lon)) @@ -77,40 +109,34 @@ def aprs_to_cot_xml(aprs_frame: dict, config: dict) -> ET.Element: # NOQA pylin track.set("course", "9999999.0") detail = ET.Element("detail") - detail.set("uid", name) + detail.set("uid", cot_uid) detail.append(uid) detail.append(contact) detail.append(track) remarks = ET.Element("remarks") - comment = aprs_frame.get("comment") + comment = frame.get("comment") if comment: remarks.text = comment detail.append(remarks) -# event.stale = time + datetime.timedelta(seconds=stale) - root = ET.Element("event") root.set("version", "2.0") root.set("type", cot_type) - root.set("uid", f"APRS.{callsign}".replace(" ", "")) + root.set("uid", cot_uid) root.set("how", "h-g-i-g-o") - root.set("time", time.strftime(pytak.ISO_8601_UTC)) - root.set("start", time.strftime(pytak.ISO_8601_UTC)) - root.set("stale", cot_stale) + root.set("time", pytak.cot_time()) + root.set("start", pytak.cot_time()) + root.set("stale", pytak.cot_time(cot_stale)) + root.append(point) root.append(detail) return root -def aprs_to_cot(aprs_frame: dict, config: dict) -> str: - """ - Converts an APRS Frame to a Cursor-on-Target Event, as a String. - """ - cot_str: str = "" - cot_xml: ET.Element = aprs_to_cot_xml(aprs_frame, config) - if cot_xml: - cot_str = ET.tostring(cot_xml) - return cot_str +def aprs_to_cot(frame: dict, config: Union[dict, None] = None) -> Union[bytes, None]: + """Wrapper that returns COT as an XML string.""" + cot: Union[ET.Element, None] = aprs_to_cot_xml(frame, config) + return ET.tostring(cot) if cot else None diff --git a/requirements_test.txt b/requirements_test.txt index ffbd4a3..9120a8a 100644 --- a/requirements_test.txt +++ b/requirements_test.txt @@ -1,12 +1,5 @@ -# Test Requirements -# -# Source:: https://github.com/ampledata/aprscot -# Author:: Greg Albrecht W2GMD -# Copyright:: Copyright 2022 Greg Albrecht -# License:: Apache License, Version 2.0 - - -flake8 +pytest-asyncio +pytest-cov pylint -twine -pytest +flake8 +black diff --git a/setup.py b/setup.py index a566cee..0e1c83c 100644 --- a/setup.py +++ b/setup.py @@ -1,5 +1,20 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- +# +# Copyright 2022 Greg Albrecht +# +# 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. +# +# Author:: Greg Albrecht W2GMD +# """ Setup for the APRS Cursor-on-Target Gateway. @@ -13,7 +28,7 @@ import setuptools __title__ = "aprscot" -__version__ = "5.0.5" +__version__ = "6.0.0" __author__ = "Greg Albrecht W2GMD " __copyright__ = "Copyright 2022 Greg Albrecht" __license__ = "Apache License, Version 2.0" @@ -36,6 +51,7 @@ def publish(): packages=[__title__], package_dir={__title__: __title__}, url=f"https://github.com/ampledata/{__title__}", + entry_points={"console_scripts": [f"{__title__} = {__title__}.commands"]}, description="APRS Cursor-on-Target Gateway.", author="Greg Albrecht", author_email="oss@undef.net", @@ -45,17 +61,11 @@ def publish(): long_description_content_type="text/x-rst", zip_safe=False, include_package_data=True, - install_requires=[ - "aprslib", - "pytak >= 3.0.0" - ], + install_requires=["aprslib", "pytak >= 5.0.0"], classifiers=[ "Topic :: Communications :: Ham Radio", "Programming Language :: Python", - "License :: OSI Approved :: Apache Software License" - ], - keywords=[ - "Ham Radio", "APRS", "Cursor on Target", "ATAK", "TAK", "CoT" + "License :: OSI Approved :: Apache Software License", ], - entry_points={"console_scripts": ["aprscot = aprscot.commands:cli"]} + keywords=["Ham Radio", "APRS", "Cursor on Target", "ATAK", "TAK", "CoT"], ) diff --git a/tests/test_functions.py b/tests/test_functions.py index 566a15a..ec518af 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -27,16 +27,16 @@ class FunctionsTestCase(unittest.TestCase): - def test_aprs_to_cot_xml(self): """ Tests that aprs_to_cot decodes an APRS Frame into a Cursor-on-Target message. """ test_frame = ( - 'SUNSET>APRS,TCPIP*,qAC,T2SP:@145502z3745.60N/12229.85W_000/' - '000g000t060P000h99b00030W2GMD Outer Sunset, SF IGate/Digipeater ' - 'http://w2gmd.org') + "SUNSET>APRS,TCPIP*,qAC,T2SP:@145502z3745.60N/12229.85W_000/" + "000g000t060P000h99b00030W2GMD Outer Sunset, SF IGate/Digipeater " + "http://w2gmd.org" + ) parsed_frame = aprslib.parse(test_frame) cot_frame = aprscot.functions.aprs_to_cot_xml(parsed_frame, {}) @@ -45,5 +45,5 @@ def test_aprs_to_cot_xml(self): self.assertEqual(cot_frame.get("uid"), "APRS.SUNSET") -if __name__ == '__main__': +if __name__ == "__main__": unittest.main()