src/ layout, functionalize for better testing and reuse

Author: scivision

.github/workflows/ci.yml

@@ -11,22 +11,26 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v2
-    - uses: actions/setup-python@v1
+    - uses: actions/setup-python@v2
       with:
         python-version: '3.x'
+
     - run: pip install .[tests,lint]
+
     - run: flake8
-    - run: mypy .
+    - run: mypy . src
+
     - run: pytest
-      working-directory: tests
+
 
   windows:
     runs-on: windows-latest
     steps:
     - uses: actions/checkout@v2
-    - uses: actions/setup-python@v1
+    - uses: actions/setup-python@v2
       with:
         python-version: '3.x'
+
     - run: pip install .[tests]
+
     - run: pytest
-      working-directory: tests

mypy.ini renamed to .mypy.ini

@@ -3,3 +3,6 @@ requires = ["setuptools", "wheel"]
 
 [tool.black]
 line-length = 132
+
+[tool.pytest.ini_options]
+addopts = "-ra -v"

csv2kml.py

@@ -1,6 +1,6 @@
 [metadata]
 name = mozloc
-version = 1.0.0
+version = 1.1.0
 author = Michael Hirsch, Ph.D.
 author_email = [email protected]
 url = https://github.com/scivision/mozilla-location-wifi
@@ -9,13 +9,12 @@ keywords =
   wifi
   geolocation
 classifiers =
-  Development Status :: 4 - Beta
+  Development Status :: 5 - Production/Stable
   Environment :: Console
   Intended Audience :: Information Technology
   Intended Audience :: System Administrators
   Operating System :: POSIX :: Linux
   Operating System :: Microsoft :: Windows
-  Programming Language :: Python :: 3.6
   Programming Language :: Python :: 3.7
   Programming Language :: Python :: 3.8
   Programming Language :: Python :: 3.9
@@ -27,14 +26,16 @@ long_description = file: README.md
 long_description_content_type = text/markdown
 
 [options]
-python_requires = >= 3.6
+python_requires = >= 3.7
 packages = find:
-scripts =
-  csv2kml.py
-  MozLoc.py
 install_requires =
   pandas
   requests
+package_dir=
+    =src
+
+[options.packages.find]
+where=src
 
 [options.extras_require]
 tests =
@@ -47,4 +48,5 @@ io =
 
 [options.entry_points]
 console_scripts =
-  MozLoc = MozLoc:main
+  MozLoc = mozloc.__main__:mozilla_location
+  mozloc_csv2kml = mozloc.__main__:csv2kml

mozloc/base.py

@@ -7,11 +7,11 @@
 
 Don't abuse the API or you'll get banned (excessive polling rate)
 """
-from mozloc import log_wifi_loc
+from .base import log_wifi_loc
 from argparse import ArgumentParser
 
 
-def main():
+def mozilla_location():
     """
     output: lat lon [deg] accuracy [m]
     """
@@ -20,10 +20,23 @@ def main():
     p.add_argument(
         "-T", "--cadence", help="how often to ping [sec]. Some laptops cannot go faster than 30 sec.", default=60, type=float
     )
+    p.add_argument(
+        "-url",
+        help="Mozilla location services URL--don't use this default test key",
+        default="https://location.services.mozilla.com/v1/geolocate?key=test",
+    )
     p = p.parse_args()
 
-    log_wifi_loc(p.cadence, p.logfile)
+    log_wifi_loc(p.cadence, p.url, p.logfile)
+
 
+def csv2kml():
+    """convert logged positions to KML"""
+    from .utils import csv2kml
+
+    p = ArgumentParser()
+    p.add_argument("logfn", help="csv logfile to read")
+    p.add_argument("kmlfn", help="kml filename to write")
+    p = p.parse_args()
 
-if __name__ == "__main__":
-    main()
+    csv2kml(p.logfn, p.kmlfn)

mozloc/config.py

@@ -0,0 +1,45 @@
+from time import sleep
+from pathlib import Path
+import logging
+
+from .modules import get_cli, cli_config_check
+from .web import get_loc_mozilla
+
+HEADER = "time                lat        lon         accuracy NumBSSIDs"
+
+
+def log_wifi_loc(cadence_sec: float, mozilla_url: str, logfile: Path = None):
+
+    if logfile:
+        logfile = Path(logfile).expanduser()
+        with logfile.open("a") as f:
+            f.write(HEADER + "\n")
+
+    print(f"updating every {cadence_sec} seconds")
+    print(HEADER)
+
+    if not cli_config_check():
+        raise ConnectionError("Could not connect to WiFi hardware")
+    # nmcli errored for less than about 0.2 sec.
+    sleep(0.5)
+    while True:
+        dat = get_cli()
+        if len(dat) < 2:
+            logging.warning(f"cannot locate since at least 2 BSSIDs required\n{dat}")
+            sleep(cadence_sec)
+            continue
+
+        loc = get_loc_mozilla(dat, mozilla_url)
+        if loc is None:
+            logging.warning(f"Did not get location from {len(dat)} BSSIDs")
+            sleep(cadence_sec)
+            continue
+
+        stat = f'{loc["t"].isoformat(timespec="seconds")} {loc["lat"]} {loc["lng"]} {loc["accuracy"]:.1f} {loc["N"]:02d}'
+        print(stat)
+
+        if logfile:
+            with logfile.open("a") as f:
+                f.write(stat + "\n")
+
+        sleep(cadence_sec)

pyproject.toml

@@ -1,15 +1,11 @@
 """ Network Manager CLI (nmcli) functions """
 import subprocess
-import typing
 import logging
 import shutil
 import pandas
+import typing as T
 from io import StringIO
 from time import sleep
-import requests
-from datetime import datetime
-
-from .config import URL
 
 NMCLI = shutil.which("nmcli")
 if not NMCLI:
@@ -20,22 +16,36 @@
 NMSCAN = [NMCLI, "device", "wifi", "rescan"]
 
 
-def cli_config_check():
+def cli_config_check() -> bool:
     # %% check that NetworkManager CLI is available and WiFi is active
-    ret = subprocess.check_output([NMCLI, "-t", "radio", "wifi"], universal_newlines=True, timeout=1.0).strip().split(":")
+    ret = subprocess.run([NMCLI, "-t", "radio", "wifi"], stdout=subprocess.PIPE, text=True, timeout=2)
+
+    if ret.returncode != 0:
+        return False
 
-    if "enabled" not in ret and "disabled" in ret:
-        raise ConnectionError("must enable WiFi, perhaps via nmcli radio wifi on")
+    stdout = ret.stdout.strip().split(":")
+    if "enabled" in stdout:
+        return True
 
+    if "disabled" in stdout:
+        logging.error(
+            """must enable WiFi, perhaps via:
+nmcli radio wifi on"""
+        )
+        return False
 
-def get_cli() -> typing.Dict[str, typing.Any]:
+    logging.error("could not determine WiFi state.")
+    return False
+
+
+def get_cli() -> T.List[T.Dict[str, T.Any]]:
 
     ret = subprocess.run(NMCMD, timeout=1.0)
     if ret.returncode != 0:
-        raise ConnectionError(f"could not connect with NetworkManager for WiFi")
+        raise ConnectionError("could not connect with NetworkManager for WiFi")
     sleep(0.5)  # nmcli errored for less than about 0.2 sec.
     # takes several seconds to update, so do it now.
-    ret = subprocess.run(NMSCAN, timeout=1.0, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
+    ret = subprocess.run(NMSCAN, timeout=1.0, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
     if ret.returncode != 0:
         logging.error(f"consider slowing scan cadence.  {ret.stderr}")
 
@@ -52,30 +62,9 @@ def get_cli() -> typing.Dict[str, typing.Any]:
     )
     # %% optout
     dat = dat[~dat["ssid"].str.endswith("_nomap")]
-    if dat.shape[0] < 2:
-        logging.warning("cannot locate since at least 2 BSSIDs required")
-        return None
+
     # %% cleanup
     dat["ssid"] = dat["ssid"].str.replace("nan", "")
     dat["macAddress"] = dat["macAddress"].str.replace(r"\\:", ":")
-    # %% JSON
-    jdat = dat.to_json(orient="records")
-    jdat = '{ "wifiAccessPoints":' + jdat + "}"
-    logging.debug(jdat)
-    # %% cloud MLS
-    try:
-        req = requests.post(URL, data=jdat)
-        if req.status_code != 200:
-            logging.error(req.text)
-            return None
-    except requests.exceptions.ConnectionError as e:
-        logging.error(f"no network connection.  {e}")
-        return None
-    # %% process MLS response
-    jres = req.json()
-    loc = jres["location"]
-    loc["accuracy"] = jres["accuracy"]
-    loc["N"] = dat.shape[0]  # number of BSSIDs used
-    loc["t"] = datetime.now()
 
-    return loc
+    return dat