diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile deleted file mode 100644 index d1b1d7f..0000000 --- a/.devcontainer/Dockerfile +++ /dev/null @@ -1,31 +0,0 @@ -# See here for image contents: https://github.com/microsoft/vscode-dev-containers/tree/v0.163.1/containers/python-3/.devcontainer/base.Dockerfile - -# [Choice] Python version: 3, 3.9, 3.8, 3.7, 3.6 -ARG VARIANT="3" -FROM mcr.microsoft.com/vscode/devcontainers/python:0-${VARIANT} - -# [Option] Install Node.js -ARG INSTALL_NODE="false" -ARG NODE_VERSION="lts/*" - -# Geosupport release versions, change args in devcontainer.json to build against a new version -ARG RELEASE=21c -ARG MAJOR=21 -ARG MINOR=3 -ARG PATCH=0 - -RUN if [ "${INSTALL_NODE}" = "true" ]; then su vscode -c "umask 0002 && . /usr/local/share/nvm/nvm.sh && nvm install ${NODE_VERSION} 2>&1"; fi - -WORKDIR /geosupport -COPY . . - -RUN FILE_NAME=linux_geo${RELEASE}_${MAJOR}_${MINOR}.zip\ - && echo $FILE_NAME\ - && curl -O https://www1.nyc.gov/assets/planning/download/zip/data-maps/open-data/$FILE_NAME\ - && unzip *.zip\ - && rm *.zip - -ENV GEOFILES=/geosupport/version-${RELEASE}_${MAJOR}.${MINOR}/fls/ -ENV LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/geosupport/version-${RELEASE}_${MAJOR}.${MINOR}/lib/ - -WORKDIR / diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json deleted file mode 100644 index fe808e8..0000000 --- a/.devcontainer/devcontainer.json +++ /dev/null @@ -1,54 +0,0 @@ -// For format details, see https://aka.ms/devcontainer.json. For config options, see the README at: -// https://github.com/microsoft/vscode-dev-containers/tree/v0.163.1/containers/python-3 -{ - "name": "Python 3", - "build": { - "dockerfile": "Dockerfile", - "context": "..", - "args": { - // Update 'VARIANT' to pick a Python version: 3, 3.6, 3.7, 3.8, 3.9 - "VARIANT": "3.9", - // Options - "INSTALL_NODE": "false", - "NODE_VERSION": "lts/*", - "RELEASE": "21c", - "MAJOR": "21", - "MINOR": "3", - "PATCH": "0", - } - }, - - // Set *default* container specific settings.json values on container create. - "settings": { - "terminal.integrated.shell.linux": "/bin/bash", - "python.pythonPath": "/usr/local/bin/python", - "python.linting.enabled": true, - "python.linting.pylintEnabled": true, - "python.formatting.autopep8Path": "/usr/local/py-utils/bin/autopep8", - "python.formatting.blackPath": "/usr/local/py-utils/bin/black", - "python.formatting.yapfPath": "/usr/local/py-utils/bin/yapf", - "python.linting.banditPath": "/usr/local/py-utils/bin/bandit", - "python.linting.flake8Path": "/usr/local/py-utils/bin/flake8", - "python.linting.mypyPath": "/usr/local/py-utils/bin/mypy", - "python.linting.pycodestylePath": "/usr/local/py-utils/bin/pycodestyle", - "python.linting.pydocstylePath": "/usr/local/py-utils/bin/pydocstyle", - "python.linting.pylintPath": "/usr/local/py-utils/bin/pylint" - }, - - // Add the IDs of extensions you want installed when the container is created. - "extensions": [ - "ms-python.python" - ], - - // Adding id_rsa so that we can push to github from the dev container - "initializeCommand": "ssh-add $HOME/.ssh/id_rsa", - - // Use 'forwardPorts' to make a list of ports inside the container available locally. - // "forwardPorts": [], - - // Use 'postCreateCommand' to run commands after the container is created. - "postCreateCommand": "pip3 install -e .", - - // Comment out connect as root instead. More info: https://aka.ms/vscode-remote/containers/non-root. - "remoteUser": "vscode" -} diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..5f9cace --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,103 @@ +name: CI + +on: + push: + branches: [ master ] + pull_request: + branches: [ master ] + +jobs: + test-windows: + runs-on: windows-latest + env: + GEO_VERSION: 25a + steps: + - uses: actions/checkout@v3 + - name: Set up Python 3.11 (64-bit) + uses: actions/setup-python@v4 + with: + python-version: 3.11 + architecture: x64 + - name: Install Geosupport Desktop (Windows) + shell: pwsh + run: | + $FILENAME = "gde_${{ env.GEO_VERSION }}_x64.zip" + $URL = "https://s-media.nyc.gov/agencies/dcp/assets/files/zip/data-tools/bytes/$FILENAME" + $LOCALDIR = "gde_${{ env.GEO_VERSION }}_x64" + $TARGETDIR = "C:\Program Files\Geosupport Desktop Edition" + + # Download and extract the installer + Invoke-WebRequest -Uri $URL -OutFile $FILENAME + Expand-Archive -Path $FILENAME -DestinationPath $LOCALDIR + + # Run the installer from the expected folder structure + Start-Process -Wait -FilePath "$LOCALDIR\setup.exe" -Verb runAs -ArgumentList '/s', '/v"/qn"' + + # Update environment variables for subsequent steps + echo "PATH=$TARGETDIR\bin;$env:PATH" >> $env:GITHUB_ENV + echo "GEOFILES=$TARGETDIR\fls\\" >> $env:GITHUB_ENV + + - name: Install dependencies + run: pip install . + - name: Run unit tests + run: python -m unittest discover + + test-linux: + runs-on: ubuntu-latest + env: + GEO_VERSION: 25a + steps: + - uses: actions/checkout@v3 + - name: Set up Python 3.11 + uses: actions/setup-python@v4 + with: + python-version: 3.11 + - name: Install Geosupport Desktop (Linux) + run: | + # Extract numeric part and the trailing letter from GEO_VERSION (e.g. "25a") + NUM="${GEO_VERSION:0:2}" + LETTER="${GEO_VERSION: -1}" + + # Map letter to the appropriate minor version number + case $LETTER in + a) MINOR=1;; + b) MINOR=2;; + c) MINOR=3;; + *) echo "Unsupported GEO_VERSION letter: $LETTER" && exit 1;; + esac + + # Build the filename based on GEO_VERSION; for example, for 25b it becomes linux_geo25b_25.2.zip + FILENAME="linux_geo${GEO_VERSION}_${NUM}.${MINOR}.zip" + URL="https://s-media.nyc.gov/agencies/dcp/assets/files/zip/data-tools/bytes/$FILENAME" + + LOCALDIR="geosupport-install-lx" + + # Download and extract the zip file + curl -L -o $FILENAME "$URL" + mkdir -p $LOCALDIR + unzip $FILENAME -d $LOCALDIR + + # Locate the extracted directory, which is named like "version-25b_25.2" + GEO_DIR=$(find $LOCALDIR -type d -name "version-${GEO_VERSION}_*" | head -n 1) + + # Set environment variables for GEOFILES and LD_LIBRARY_PATH + echo "GEOFILES=$GITHUB_WORKSPACE/$GEO_DIR/fls/" >> $GITHUB_ENV + echo "LD_LIBRARY_PATH=$GITHUB_WORKSPACE/$GEO_DIR/lib/:$LD_LIBRARY_PATH" >> $GITHUB_ENV + + - name: Install dependencies + run: pip install . + - name: Run unit tests + run: python -m unittest discover + + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v3 + - name: Set up Python 3.11 + uses: actions/setup-python@v4 + with: + python-version: 3.11 + - name: Install code quality tools + run: pip install black + - name: Check code formatting with black + run: black --check . diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..c5504a5 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,62 @@ +name: Create Release + +on: + push: + tags: + - 'v*' + +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v3 + with: + fetch-depth: 0 # Fetch all history and tags + + + + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: '3.11' + + - name: Verify version matches tag + run: | + TAG_VERSION=${GITHUB_REF#refs/tags/v} + PACKAGE_VERSION=$(python -c "import geosupport; print(geosupport.__version__)") + if [ "$TAG_VERSION" != "$PACKAGE_VERSION" ]; then + echo "ERROR: Tag version ($TAG_VERSION) doesn't match package version ($PACKAGE_VERSION)" + exit 1 + fi + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install build twine + pip install -e . + + - name: Extract version from tag + id: get_version + run: echo "VERSION=${GITHUB_REF#refs/tags/v}" >> $GITHUB_ENV + + - name: Build package + run: python -m build + + - name: Create GitHub Release + id: create_release + uses: softprops/action-gh-release@v1 + with: + files: | + dist/*.whl + dist/*.tar.gz + generate_release_notes: true + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + + - name: Publish to PyPI on tag push + # This step will only run if the tag starts with 'v' + if: startsWith(github.ref, 'refs/tags/') + uses: pypa/gh-action-pypi-publish@release/v1 + with: + user: __token__ + password: ${{ secrets.PYPI_API_TOKEN }} \ No newline at end of file diff --git a/.gitignore b/.gitignore index d49510e..a185a55 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,6 @@ MANIFEST cover .coverage coverage.xml +gde/ +upg/ +docker/ diff --git a/README.md b/README.md index 39f0c02..9ddf853 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,34 @@ # python-geosupport -[![Build status](https://ci.appveyor.com/api/projects/status/5uocynec8e3maeeq?svg=true&branch=master)](https://ci.appveyor.com/project/ishiland/python-geosupport) [![PyPI version](https://img.shields.io/pypi/v/python-geosupport.svg)](https://pypi.python.org/pypi/python-geosupport/) [![Python 2.7 | 3.4+](https://img.shields.io/badge/python-2.7%20%7C%203.4+-blue.svg)](https://www.python.org/downloads/release/python-360/) +![Build status](https://github.com/ishiland/python-geosupport/actions/workflows/ci.yml/badge.svg) [![PyPI version](https://img.shields.io/pypi/v/python-geosupport.svg)](https://pypi.python.org/pypi/python-geosupport/) [![3.8+](https://img.shields.io/badge/python-3.8+-blue.svg)](https://www.python.org/downloads/release/python-360/) -Python bindings for NYC Planning's [Geosupport Desktop Edition](https://www1.nyc.gov/site/planning/data-maps/open-data/dwn-gde-home.page). +Geocode NYC addresses locally using Python bindings for NYC Planning's [Geosupport Desktop Edition](https://www1.nyc.gov/site/planning/data-maps/open-data/dwn-gde-home.page). -### [Read the docs](https://python-geosupport.readthedocs.io/en/latest/) +## Documentation + +Check out documentation for installing and usage [here](https://python-geosupport.readthedocs.io/en/latest/). + +## Features + +- Pythonic interface to all Geosupport functions +- Support for both Windows and Linux platforms +- Secure and fast using local geocoding - no API calls required +- Built-in error handling for Geosupport return codes +- Interactive help menu + +## Compatibility + +- Python 3.8+ +- Tested on Geosupport Desktop Edition 25a +- Windows (64-bit & 32-bit) and Linux operating systems ## Quickstart +```bash +pip install python-geosupport +``` + ```python # Import the library and create a `Geosupport` object. from geosupport import Geosupport @@ -40,11 +60,22 @@ result = g.address(house_number=125, street_name='Worth St', borough_code='Mn') } ``` -## License +## Examples -This project is licensed under the MIT License - see the [license.txt](license.txt) file for details +See the examples directory and accompanying [readme.md](examples/readme.md). + + +## Contributing -## Contributors -Thanks to [Jeremy Neiman](https://github.com/docmarionum1) for a major revision incorporating all Geosupport functions and parameters. +1. Fork the repository +2. Create your feature branch (`git checkout -b feature/amazing-feature`) +3. Make your changes +4. Run tests (`python -m unittest discover`) +5. Run Black formatting (`black .`) +6. Commit your changes (`git commit -m 'Add some amazing feature'`) +7. Push to the branch (`git push origin feature/amazing-feature`) +8. Open a Pull Request -If you see an issue or would like to contribute, pull requests are welcome. +## License + +This project is licensed under the MIT License - see the [license.txt](license.txt) file for details diff --git a/appveyor/build.ps1 b/appveyor/build.ps1 deleted file mode 100644 index 54ec762..0000000 --- a/appveyor/build.ps1 +++ /dev/null @@ -1,133 +0,0 @@ -# Variables to help determine new naming convention in download string - currently only testing different geosupport versions in linux -# New naming convention example: linux_geo18d_184.zip -$legacyVersions = @('18a', '18b', '18c') -$subVersions = 'a', 'b', 'c', 'd', 'e', 'f' -$BASE_URL = 'https://www1.nyc.gov/assets/planning/download/zip/data-maps/open-data/' - -# DL function modified from https://github.com/ogrisel/python-appveyor-demo/blob/master/appveyor/install.ps1 -function Download($filename, $url) -{ - $webclient = New-Object System.Net.WebClient - $basedir = $pwd.Path + "//" - $filepath = $basedir + $filename - if (Test-Path $filename) - { - Write-Host "Reusing" $filepath - return $filepath - } - - # Download and retry up to 3 times in case of network transient errors. - Write-Host "Downloading" $filename "from" $url - $retry_attempts = 2 - for ($i = 0; $i -lt $retry_attempts; $i++) { - try - { - Write-Host "Download attempt" $( $i + 1 ) - $webclient.DownloadFile($url, $filepath) - break - } - Catch [Exception] - { - Write-Host "Download Error" - Start-Sleep 1 - } - } - if (Test-Path $filepath) - { - Write-Host "File saved at" $filepath - } - else - { - Write-Host "File not downloaded" - # Retry once to get the error message if any at the last try - $webclient.DownloadFile($url, $filepath) - } - return $filepath -} - -if ($isWindows) -{ - # set download and temp directory names - if ($env:PYTHON_ARCH -eq '64') - { - - $LOCALDIR = 'geosupport-install-x64' - $TARGETDIR = 'C:\Program Files\Geosupport Desktop Edition' - $FILENAME = "gde64_$( $env:GEO_VERSION ).zip" - $URL = "$( $BASE_URL )$( $FILENAME )" - } - elseif ($env:PYTHON_ARCH -eq '32') - { - $LOCALDIR = 'geosupport-install-x86' - $TARGETDIR = 'C:\Program Files (x86)\Geosupport Desktop Edition' - $FILENAME = "gde_$( $env:GEO_VERSION ).zip" - $URL = "$( $BASE_URL )$( $FILENAME )" - } - - # download - Write-Host "Downloading $env:PYTHON_ARCH bit Geosupport version $env:GEO_VERSION for Windows..." - $DOWNLOAD_FILE = Download $FILENAME $URL - - # extract - Write-Host "Extracting..." - unzip $FILENAME -d $LOCALDIR - - # delete .zip - rm $FILENAME - - # silently install Geosupport Desktop - Write-Host "Installing..." - Start-Process -Wait -FilePath "$( $LOCALDIR )/setup.exe" -Verb runAs -ArgumentList '/s', '/v"/qn"' - - # set Geosupport Environmental variables - $env:PATH = "$( $TARGETDIR )\bin;$( $env:PATH )" - $env:GEOFILES = "$( $TARGETDIR )\fls\" - - Write-Host "Install complete." -} - -elseif ($isLinux) -{ - if ($legacyVersions -contains $env:GEO_VERSION) - { - $FILENAME = "gdelx_$( $env:GEO_VERSION ).zip" - } - - # determine string if new geosupport download naming convention - else - { - foreach ($version in $subVersions) - { - if ($version -eq $env:GEO_VERSION.Substring(2)) - { - $idx = [array]::indexOf($subVersions, $version) + 1 - $FILENAME = "linux_geo$( $env:GEO_VERSION )_$($env:GEO_VERSION.Substring(0, 2) )$( $idx ).zip" - } - } - } - - # set download string and local directory names - $LOCALDIR = 'geosupport-install-lx' - $URL = "$( $BASE_URL )$( $FILENAME )" - - # download - Write-Host "Downloading Geosuport version $env:GEO_VERSION for Linux..." - Download $FILENAME $URL - - # extract - Write-Host "Extracting..." - unzip $FILENAME -d $LOCALDIR - - # get the first child directory name of the unzipped geosupport install dir - $GEO_DIR_CHILD_NAME = Get-ChildItem $LOCALDIR -Recurse | Where-Object { $_.FullName -like "*$( $env:GEO_VERSION )*" } | Select-Object -First 1 | select -expand Name - $INSTALL_PATH = "$( $pwd )/$( $LOCALDIR )/$( $GEO_DIR_CHILD_NAME )" - - # delete .zip - rm $FILENAME - - # set Geosupport Environmental variables - $env:GEOFILES = "$( $INSTALL_PATH )/fls/" - $env:LD_LIBRARY_PATH = "$( $INSTALL_PATH )/lib/:$( $env:LD_LIBRARY_PATH )" - - Write-Host "Install complete." -} \ No newline at end of file diff --git a/docs/conf.py b/docs/conf.py index a91ca02..255cc2c 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -1,4 +1,5 @@ import sphinx_rtd_theme + # Configuration file for the Sphinx documentation builder. # # This file only contains a selection of the most common options. For a full @@ -18,9 +19,9 @@ # -- Project information ----------------------------------------------------- -project = 'python-geosupport' -copyright = '2019, Ian Shiland, Jeremy Neiman' -author = 'Ian Shiland, Jeremy Neiman' +project = "python-geosupport" +copyright = "2025, Ian Shiland, Jeremy Neiman" +author = "Ian Shiland, Jeremy Neiman" # -- General configuration --------------------------------------------------- @@ -32,15 +33,15 @@ "sphinx_rtd_theme", ] -source_suffix = '.rst' +source_suffix = ".rst" # Add any paths that contain templates here, relative to this directory. -templates_path = ['_templates'] +templates_path = ["_templates"] # List of patterns, relative to source directory, that match files and # directories to ignore when looking for source files. # This pattern also affects html_static_path and html_extra_path. -exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store'] +exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"] # -- Options for HTML output ------------------------------------------------- @@ -53,4 +54,4 @@ # Add any paths that contain custom static files (such as style sheets) here, # relative to this directory. They are copied after the builtin static files, # so a file named "default.css" will overwrite the builtin "default.css". -html_static_path = ['_static'] +html_static_path = ["_static"] diff --git a/examples/pandas_multiprocessing.py b/examples/pandas_multiprocessing.py index b46bbd3..1fbf660 100644 --- a/examples/pandas_multiprocessing.py +++ b/examples/pandas_multiprocessing.py @@ -18,8 +18,8 @@ cpus = cpu_count() -INPUT_CSV = '/examples/data/input.csv' -OUTPUT_CSV = '/examples/data/output-pandas-multiprocessing.csv' +INPUT_CSV = "/examples/data/input.csv" +OUTPUT_CSV = "/examples/data/output-pandas-multiprocessing.csv" def geo_by_address(row): @@ -31,19 +31,23 @@ def geo_by_address(row): """ try: # parse the address to separate PHN and street - parsed = p.address(row['Address']) + parsed = p.address(row["Address"]) # geocode - result = g.address(house_number=parsed['PHN'], street_name=parsed['STREET'], borough=row['Borough']) + result = g.address( + house_number=parsed["PHN"], + street_name=parsed["STREET"], + borough=row["Borough"], + ) lat = result.get("Latitude") - lon = result.get('Longitude') - msg = result.get('Message') + lon = result.get("Longitude") + msg = result.get("Message") except GeosupportError as ge: lat = "" lon = "" msg = str(ge) return pd.Series([lat, lon, msg]) - - + + def parallelize(data, func, num_of_processes=cpus): data_split = np.array_split(data, num_of_processes) pool = Pool(num_of_processes) @@ -61,17 +65,13 @@ def parallelize_on_rows(data, func, num_of_processes=cpus): return parallelize(data, partial(run_on_subset, func), num_of_processes) -if __name__ == '__main__': - +if __name__ == "__main__": + # read in csv df = pd.read_csv(INPUT_CSV) - + # add 3 Geosupport columns - Latitude, Longitude and Geosupport message - df[['lat', 'lon', 'msg']] = parallelize_on_rows(df, geo_by_address) + df[["lat", "lon", "msg"]] = parallelize_on_rows(df, geo_by_address) # output to csv with the 3 new columns. df.to_csv(OUTPUT_CSV) - - - - diff --git a/examples/pandas_simple.py b/examples/pandas_simple.py index 520b25e..88f3171 100644 --- a/examples/pandas_simple.py +++ b/examples/pandas_simple.py @@ -12,8 +12,8 @@ g = Geosupport() p = Parser() -INPUT_CSV = '/examples/data/input.csv' -OUTPUT_CSV = '/examples/data/output-pandas-simple.csv' +INPUT_CSV = "/examples/data/input.csv" +OUTPUT_CSV = "/examples/data/output-pandas-simple.csv" def geo_by_address(row): @@ -24,12 +24,16 @@ def geo_by_address(row): """ try: # parse the address to separate PHN and street - parsed = p.address(row['Address']) + parsed = p.address(row["Address"]) # geocode - result = g.address(house_number=parsed['PHN'], street_name=parsed['STREET'], borough=row['Borough']) + result = g.address( + house_number=parsed["PHN"], + street_name=parsed["STREET"], + borough=row["Borough"], + ) lat = result.get("Latitude") - lon = result.get('Longitude') - msg = result.get('Message') + lon = result.get("Longitude") + msg = result.get("Message") except GeosupportError as ge: lat = "" lon = "" @@ -37,12 +41,12 @@ def geo_by_address(row): return pd.Series([lat, lon, msg]) -if __name__ == '__main__': +if __name__ == "__main__": # read in csv df = pd.read_csv(INPUT_CSV) # add 3 Geosupport columns - Latitude, Longitude and Geosupport message - df[['lat', 'lon', 'msg']] = df.apply(geo_by_address, axis=1) + df[["lat", "lon", "msg"]] = df.apply(geo_by_address, axis=1) # output the new dataframe to a csv df.to_csv(OUTPUT_CSV, index=False) diff --git a/geosupport/__init__.py b/geosupport/__init__.py index 77dfb97..3214bec 100644 --- a/geosupport/__init__.py +++ b/geosupport/__init__.py @@ -1 +1,3 @@ from .geosupport import Geosupport, GeosupportError + +__version__ = "1.1.0" diff --git a/geosupport/config.py b/geosupport/config.py index a2c7786..4294b8d 100644 --- a/geosupport/config.py +++ b/geosupport/config.py @@ -1,21 +1,41 @@ from os import path +from typing import Dict, Union -FUNCTION_INFO_PATH = path.join( - path.abspath(path.dirname(__file__)), - 'function_info' -) +# Constants for work area sizes. +WA1_SIZE: int = 1200 +WA2_SIZE: int = 32767 # Maximum size for WA2 -FUNCTION_INFO_CSV = path.join(FUNCTION_INFO_PATH, 'function_info.csv') -FUNCTION_INPUTS_CSV = path.join(FUNCTION_INFO_PATH, 'function_inputs.csv') -WORK_AREA_LAYOUTS_PATH = path.join(FUNCTION_INFO_PATH, 'work_area_layouts') +FUNCTION_INFO_PATH = path.join(path.abspath(path.dirname(__file__)), "function_info") -BOROUGHS = { - 'MANHATTAN': 1, 'MN': 1, 'NEW YORK': 1, 'NY': 1, '36061': 1, - 'BRONX': 2, 'THE BRONX': 2, 'BX': 2, '36005': 2, - 'BROOKLYN': 3, 'BK': 3, 'BKLYN': 3, 'KINGS': 3, '36047': 3, - 'QUEENS': 4, 'QN': 4, 'QU': 4, '36081': 4, - 'STATEN ISLAND': 5, 'SI': 5, 'STATEN IS': 5, 'RICHMOND': 5, '36085': 5, - '': '', +FUNCTION_INFO_CSV = path.join(FUNCTION_INFO_PATH, "function_info.csv") +FUNCTION_INPUTS_CSV = path.join(FUNCTION_INFO_PATH, "function_inputs.csv") +WORK_AREA_LAYOUTS_PATH = path.join(FUNCTION_INFO_PATH, "work_area_layouts") + +BOROUGHS: Dict[str, Union[int, str]] = { + "MANHATTAN": 1, + "MN": 1, + "NEW YORK": 1, + "NY": 1, + "36061": 1, + "BRONX": 2, + "THE BRONX": 2, + "BX": 2, + "36005": 2, + "BROOKLYN": 3, + "BK": 3, + "BKLYN": 3, + "KINGS": 3, + "36047": 3, + "QUEENS": 4, + "QN": 4, + "QU": 4, + "36081": 4, + "STATEN ISLAND": 5, + "SI": 5, + "STATEN IS": 5, + "RICHMOND": 5, + "36085": 5, + "": "", } -USER_CONFIG = '~/.python-geosupport.cfg' +USER_CONFIG = "~/.python-geosupport.cfg" diff --git a/geosupport/error.py b/geosupport/error.py index 1d8cacb..29d6e47 100644 --- a/geosupport/error.py +++ b/geosupport/error.py @@ -1,4 +1,7 @@ +from typing import Dict, Any + + class GeosupportError(Exception): - def __init__(self, message, result={}): + def __init__(self, message: str, result: Dict[str, Any] = {}) -> None: super(GeosupportError, self).__init__(message) - self.result = result + self.result = result if result is not None else {} diff --git a/geosupport/function_info.py b/geosupport/function_info.py index 8c51416..f6df79f 100644 --- a/geosupport/function_info.py +++ b/geosupport/function_info.py @@ -1,81 +1,80 @@ from csv import DictReader import glob from os import path +from typing import Dict, List, Optional, Any, Tuple, cast from .config import FUNCTION_INFO_CSV, FUNCTION_INPUTS_CSV, WORK_AREA_LAYOUTS_PATH + class FunctionDict(dict): + alt_names: Dict[str, str] # Declare as a class attribute. - def __init__(self): - super(FunctionDict, self).__init__() - self.alt_names = {} + def __init__(self) -> None: + super().__init__() + self.alt_names = ( + {} + ) # Now this assignment doesn't include an inline type declaration. - def __getitem__(self, name): + def __getitem__(self, name: str) -> Any: name = str(name).strip().upper() if self.alt_names and name in self.alt_names: name = self.alt_names[name] + return super().__getitem__(name) - return super(FunctionDict, self).__getitem__(name) - - def __contains__(self, name): - name = str(name).strip().upper() + def __contains__(self, name: object) -> bool: + name_str = str(name).strip().upper() + return (name_str in self.alt_names) or super().__contains__(name_str) - return ( - (name in self.alt_names) or - (super(FunctionDict, self).__contains__(name)) - ) -def load_function_info(): +def load_function_info() -> FunctionDict: functions = FunctionDict() - - alt_names = {} + alt_names: Dict[str, str] = {} with open(FUNCTION_INFO_CSV) as f: - csv = DictReader(f) - for row in csv: - function = row['function'] + csv_reader = DictReader(f) + for row in csv_reader: + row = cast(Dict[str, Any], dict(row)) + function = row["function"] for k in MODES: - if row[k]: + if row.get(k): row[k] = int(row[k]) else: row[k] = None - if row['alt_names']: - row['alt_names'] = [ - n.strip() for n in row['alt_names'].split(',') - ] + if row.get("alt_names"): + row["alt_names"] = [n.strip() for n in row["alt_names"].split(",")] else: - row['alt_names'] = [] + row["alt_names"] = [] # List[str] - for n in row['alt_names']: + for n in row["alt_names"]: alt_names[n.upper()] = function - row['inputs'] = [] - + row["inputs"] = [] # List[Any] functions[function] = row functions.alt_names = alt_names with open(FUNCTION_INPUTS_CSV) as f: - csv = DictReader(f) - - for row in csv: - if row['function']: - functions[row['function']]['inputs'].append({ - 'name': row['field'], - 'comment': row['comments'] - }) + csv_reader = DictReader(f) + for row in csv_reader: + row = cast(Dict[str, Any], dict(row)) + if row.get("function"): + functions[row["function"]]["inputs"].append( + {"name": row["field"], "comment": row["comments"]} + ) return functions -def list_functions(): - s = sorted([ - "%s (%s)" % ( - function['function'], ', '.join(function['alt_names']) - ) for function in FUNCTIONS.values() - ]) - s = ["List of functions (and alternate names):"] + s - s.append( + +def list_functions() -> str: + s_list: List[str] = sorted( + [ + "%s (%s)" % (function["function"], ", ".join(function["alt_names"])) + for function in FUNCTIONS.values() + ] + ) + s_list = ["List of functions (and alternate names):"] + s_list + s_list.append( "\nCall a function using the function code or alternate name using " "Geosupport.() or Geosupport['']()." "\n\nExample usage:\n" @@ -86,124 +85,114 @@ def list_functions(): " # Call function 3 using the function code.\n" " g['3']({'borough_code': 'MN', 'on': '1 Av', 'from': '1 st', 'to': '9 st'})\n" "\nUse Geosupport.help() or Geosupport..help() " - "to read about specific function." + "to read about a specific function." ) - return '\n'.join(s) + return "\n".join(s_list) -def function_help(function, return_as_string=False): - function = FUNCTIONS[function] - s = [ +def function_help(function: str, return_as_string: bool = False) -> Optional[str]: + func_info: Dict[str, Any] = FUNCTIONS[function] + s_parts: List[str] = [ "", - "%s (%s)" % (function['function'], ', '.join(function['alt_names'])), - "="*40, - function['description'], + "%s (%s)" % (func_info["function"], ", ".join(func_info["alt_names"])), + "=" * 40, + func_info["description"], "", - "Input: %s" % function['input'], - "Output: %s" % function['output'], - "Modes: %s" % ', '.join([ - m for m in MODES if function[m] is not None - ]), + "Input: %s" % func_info["input"], + "Output: %s" % func_info["output"], + "Modes: %s" % ", ".join([m for m in MODES if func_info[m] is not None]), "\nInputs", - "="*40, - "\n".join([ - "%s - %s" % (i['name'], i['comment']) for i in function['inputs'] - ]), + "=" * 40, + "\n".join(["%s - %s" % (i["name"], i["comment"]) for i in func_info["inputs"]]), "\nReference", - "="*40, - function['links'], - "" + "=" * 40, + func_info["links"], + "", ] - - s = "\n".join(s) - + s: str = "\n".join(s_parts) if return_as_string: return s else: print(s) + return None -def input_help(): - s = [ + +def input_help() -> str: + s_parts: List[str] = [ "\nThe following is a full list of inputs for Geosupport. " - "It has the full name (followed by alternate names.)", + "It has the full name (followed by alternate names).", "To use the full names, pass a dictionary of values to the " - "Geosupport functions. Many of the inputs also have alternate names " - "in parantheses, which can be passed as keyword arguments as well.", + "Geosupport functions. Many inputs also have alternate names " + "in parentheses, which can be passed as keyword arguments as well.", "\nInputs", - "="*40, + "=" * 40, ] - for i in INPUT: - s.append("%s (%s)" % (i['name'], ', '.join(i['alt_names']))) - s.append("-"*40) - s.append("Functions: %s" % i['functions']) - s.append("Expected Values: %s\n" % i['value']) + s_parts.append("%s (%s)" % (i["name"], ", ".join(i["alt_names"]))) + s_parts.append("-" * 40) + s_parts.append("Functions: %s" % i["functions"]) + s_parts.append("Expected Values: %s\n" % i["value"]) + return "\n".join(s_parts) - return '\n'.join(s) -def load_work_area_layouts(): - work_area_layouts = {} - inputs = [] +def load_work_area_layouts() -> Tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]]: + work_area_layouts: Dict[str, Dict[str, Any]] = {} + inputs: List[Dict[str, Any]] = [] - for csv in glob.glob(path.join(WORK_AREA_LAYOUTS_PATH, '*', '*.csv')): - directory = path.basename(path.dirname(csv)) + for csv_file in glob.glob(path.join(WORK_AREA_LAYOUTS_PATH, "*", "*.csv")): + directory = path.basename(path.dirname(csv_file)) if directory not in work_area_layouts: work_area_layouts[directory] = {} - layout = {} - name = path.basename(csv).split('.')[0] + layout: Dict[str, Any] = {} + name = path.basename(csv_file).split(".")[0] - if '-' in name: - functions, mode = name.split('-') - mode = '-' + mode + if "-" in name: + functions_part, mode = name.split("-") + mode = "-" + mode else: - functions = name - mode = '' + functions_part = name + mode = "" - functions = functions.split('_') - for function in functions: + functions_list = functions_part.split("_") + for function in functions_list: work_area_layouts[directory][function + mode] = layout - with open(csv) as f: + with open(csv_file) as f: rows = DictReader(f) - for row in rows: - name = row['name'].strip().strip(':').strip() - - parent = row['parent'].strip().strip(':').strip() - if parent and 'i' in layout[parent]: + row = cast(Dict[str, Any], dict(row)) + name_field = row["name"].strip().strip(":").strip() + parent = row["parent"].strip().strip(":").strip() + if parent and parent in layout and "i" in layout[parent]: layout[parent] = {parent: layout[parent]} - - alt_names = [ - n.strip() for n in row['alt_names'].split(',') if n - ] - + alt_names = [n.strip() for n in row["alt_names"].split(",") if n] v = { - 'i': (int(row['from']) - 1, int(row['to'])), - 'formatter': row['formatter'] + "i": (int(row["from"]) - 1, int(row["to"])), + "formatter": row["formatter"], } - if parent: - layout[parent][name] = v + layout[parent][name_field] = v else: - layout[name] = v - + layout[name_field] = v for n in alt_names: layout[n] = v layout[n.upper()] = v layout[n.lower()] = v - - if directory == 'input': - inputs.append({ - 'name': name, - 'alt_names': alt_names, - 'functions': row['functions'], - 'value': row['value'] - }) + if directory == "input": + inputs.append( + { + "name": name_field, + "alt_names": alt_names, + "functions": row["functions"], + "value": row["value"], + } + ) return work_area_layouts, inputs -MODES = ['regular', 'extended', 'long', 'long+tpad'] -AUXILIARY_SEGMENT_LENGTH = 500 + +MODES: List[str] = ["regular", "extended", "long", "long+tpad"] +AUXILIARY_SEGMENT_LENGTH: int = 500 FUNCTIONS = load_function_info() WORK_AREA_LAYOUTS, INPUT = load_work_area_layouts() diff --git a/geosupport/geosupport.py b/geosupport/geosupport.py index 6d9e0bc..c853b48 100644 --- a/geosupport/geosupport.py +++ b/geosupport/geosupport.py @@ -1,146 +1,169 @@ from functools import partial import os import sys +import logging +from configparser import ConfigParser +from typing import Any, Optional, Tuple -try: - from configparser import ConfigParser # Python 3 -except: - from ConfigParser import ConfigParser # Python 2 - -from .config import USER_CONFIG +from .config import USER_CONFIG, WA1_SIZE, WA2_SIZE from .error import GeosupportError from .function_info import FUNCTIONS, function_help, list_functions, input_help from .io import format_input, parse_output, set_mode -from .sysutils import build_win_dll_path - -GEOLIB = None - -class Geosupport(object): - def __init__(self, geosupport_path=None, geosupport_version=None): +from .platform_utils import load_geosupport_library + +# Set up module-level logging. +logger = logging.getLogger(__name__) +logger.setLevel(logging.INFO) +handler = logging.StreamHandler() +formatter = logging.Formatter("[%(levelname)s] %(asctime)s - %(name)s - %(message)s") +handler.setFormatter(formatter) +logger.addHandler(handler) + + +class Geosupport: + """ + Python wrapper for the Geosupport library. + + This class loads the Geosupport C library using a helper function + to encapsulate platform-specific logic. Work areas (WA1 and WA2) + are allocated with fixed sizes according to the Geosupport COW requirements. + """ + + def __init__( + self, + geosupport_path: Optional[str] = None, + geosupport_version: Optional[str] = None, + ) -> None: global GEOLIB - self.py_version = sys.version_info[0] - self.platform = sys.platform - self.py_bit = '64' if (sys.maxsize > 2 ** 32) else '32' + self.platform: str = sys.platform + self.py_bit: str = "64" if (sys.maxsize > 2**32) else "32" if geosupport_version is not None: config = ConfigParser() config.read(os.path.expanduser(USER_CONFIG)) - versions = dict(config.items('versions')) - geosupport_path = versions[geosupport_version.lower()] + versions = dict(config.items("versions")) + geosupport_path = versions.get(geosupport_version.lower()) + logger.debug("Using geosupport version: %s", geosupport_version) + # On Windows, if a geosupport_path is provided, set the necessary environment variables. if geosupport_path is not None: - if self.platform.startswith('linux'): + if self.platform.startswith("linux"): raise GeosupportError( - "geosupport_path and geosupport_version not valid with " - "linux. You must set LD_LIBRARY_PATH and GEOFILES " - "before running python." + "geosupport_path and geosupport_version are not valid on Linux. " + "You must set LD_LIBRARY_PATH and GEOFILES before running Python." ) - os.environ['GEOFILES'] = os.path.join(geosupport_path, 'Fls\\') - os.environ['PATH'] = ';'.join([ - i for i in os.environ['PATH'].split(';') if - 'GEOSUPPORT' not in i.upper() - ]) - os.environ['PATH'] += ';' + os.path.join(geosupport_path, 'bin') + os.environ["GEOFILES"] = os.path.join(geosupport_path, "Fls" + os.sep) + os.environ["PATH"] = ";".join( + i + for i in os.environ.get("PATH", "").split(";") + if "GEOSUPPORT" not in i.upper() + ) + os.environ["PATH"] += ";" + os.path.join(geosupport_path, "bin") + logger.debug( + "Environment variables set using geosupport_path: %s", geosupport_path + ) try: - if self.platform == 'win32': - from ctypes import windll, cdll, WinDLL, wintypes - - if GEOLIB is not None: - kernel32 = WinDLL('kernel32') - kernel32.FreeLibrary.argtypes = [wintypes.HMODULE] - kernel32.FreeLibrary(GEOLIB._handle) - - # get the full path to the NYCGEO.dll - nyc_geo_dll_path = build_win_dll_path(geosupport_path) - - if self.py_bit == '64': - self.geolib = cdll.LoadLibrary(nyc_geo_dll_path) - else: - self.geolib = windll.LoadLibrary(nyc_geo_dll_path) - elif self.platform.startswith('linux'): - from ctypes import cdll - - if GEOLIB is not None: - cdll.LoadLibrary('libdl.so').dlclose(GEOLIB._handle) - - self.geolib = cdll.LoadLibrary("libgeo.so") - else: - raise GeosupportError( - 'This Operating System is currently not supported.' - ) - + # Load the Geosupport library using the helper function. + self.geolib = load_geosupport_library(geosupport_path or "", self.py_bit) GEOLIB = self.geolib + logger.debug("Geosupport library loaded successfully.") except OSError as e: + logger.exception("Error loading Geosupport library.") raise GeosupportError( - '%s\n' - 'You are currently using a %s-bit Python interpreter. ' - 'Is the installed version of Geosupport %s-bit?' % ( - e, self.py_bit, self.py_bit - ) + f"{e}\nYou are currently using a {self.py_bit}-bit Python interpreter. " + f"Is the installed version of Geosupport {self.py_bit}-bit?" ) - def _call_geolib(self, wa1, wa2): + def _call_geosupport(self, wa1: str, wa2: Optional[str]) -> Tuple[str, str]: """ - Calls the Geosupport libs & encodes/deocodes strings for Python 3. + Prepares mutable buffers for the Geosupport function call, calls the library, + and returns the resulting work areas as a tuple (WA1, WA2). + + Assumes that wa1 and wa2 (if not None) are formatted to the exact fixed lengths. """ - # encode - if self.py_version == 3: - wa1 = bytes(str(wa1), 'utf8') - wa2 = bytes(str(wa2), 'utf8') - - # Call Geosupport libs - if self.platform == 'win32': - self.geolib.NYCgeo(wa1, wa2) # windows + from ctypes import create_string_buffer + + buf1 = create_string_buffer(wa1.encode("utf8"), WA1_SIZE) + if wa2 is None: + buf2 = create_string_buffer(WA2_SIZE) else: - self.geolib.geo(wa1, wa2) # linux + buf2 = create_string_buffer(wa2.encode("utf8"), WA2_SIZE) + + logger.debug( + "Calling Geosupport function with WA1 size: %d and WA2 size: %d", + WA1_SIZE, + WA2_SIZE, + ) + if self.platform.startswith("win"): + self.geolib.NYCgeo(buf1, buf2) + else: + self.geolib.geo(buf1, buf2) - # decode - if self.py_version == 3: - wa1 = str(wa1, 'utf8') - wa2 = str(wa2, 'utf8') + out_wa1 = buf1.value.decode("utf8") + out_wa2 = buf2.value.decode("utf8") + logger.debug("Geosupport call completed.") + return out_wa1, out_wa2 - return wa1, wa2 + def call( + self, kwargs_dict: Optional[dict] = None, mode: Optional[str] = None, **kwargs + ) -> dict: + """ + Prepares work areas using format_input, calls the Geosupport library, + and returns the parsed output as a dictionary. - def call(self, kwargs_dict=None, mode=None, **kwargs): + Raises a GeosupportError if the Geosupport Return Code indicates an error. + """ if kwargs_dict is None: kwargs_dict = {} kwargs_dict.update(kwargs) kwargs_dict.update(set_mode(mode)) - flags, wa1, wa2 = format_input(kwargs_dict) - wa1, wa2 = self._call_geolib(wa1, wa2) + # Ensure wa2 is a string. + wa2 = wa2 if wa2 is not None else "" + logger.debug("Formatted WA1 and WA2 using input parameters.") + wa1, wa2 = self._call_geosupport(wa1, wa2) result = parse_output(flags, wa1, wa2) - - return_code = result['Geosupport Return Code (GRC)'] + return_code = result.get("Geosupport Return Code (GRC)", "") if not return_code.isdigit() or int(return_code) > 1: - raise GeosupportError( - result['Message'] + ' ' + result['Message 2'], - result + error_message = ( + result.get("Message", "") + " " + result.get("Message 2", "") ) + logger.error("Geosupport call error: %s", error_message) + raise GeosupportError(error_message, result) + logger.debug("Geosupport call returned successfully.") return result - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: + """ + Allows calling Geosupport functions as attributes. + For example, geosupport.some_function(...). + """ if name in FUNCTIONS: - p = partial(self.call, function=name) + p: Any = partial(self.call, function=name) p.help = partial(function_help, name) return p + raise AttributeError( + f"'{self.__class__.__name__}' object has no attribute '{name}'" + ) - raise AttributeError("'%s' object has no attribute '%s'" %( - self.__class__.__name__, name - )) - - def __getitem__(self, name): + def __getitem__(self, name: str) -> Any: return self.__getattr__(name) - def help(self, name=None, return_as_string=False): + def help(self, name: Optional[str] = None, return_as_string: bool = False) -> Any: + """ + Displays or returns help for a Geosupport function. + If no name is provided, lists all available functions. + """ + return_val: Optional[str] = None if name: - if name.upper() == 'INPUT': + if name.upper() == "INPUT": return_val = input_help() - try: - return_val = function_help(name, return_as_string) - except KeyError: - return_val = "Function '%s' does not exist." % name + else: + try: + return_val = function_help(name, return_as_string) + except KeyError: + return_val = f"Function '{name}' does not exist." else: return_val = list_functions() diff --git a/geosupport/io.py b/geosupport/io.py index 812ad03..4a9c025 100644 --- a/geosupport/io.py +++ b/geosupport/io.py @@ -1,38 +1,37 @@ from functools import partial +from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast from .config import BOROUGHS from .error import GeosupportError -from .function_info import ( - FUNCTIONS, AUXILIARY_SEGMENT_LENGTH, WORK_AREA_LAYOUTS -) +from .function_info import FUNCTIONS, AUXILIARY_SEGMENT_LENGTH, WORK_AREA_LAYOUTS -def list_of(length, callback, v): - output = [] + +def list_of(length: int, callback: Callable[[str], Any], v: str) -> List[Any]: + output: List[Any] = [] i = 0 # While the next entry isn't blank - while v[i:i+length].strip() != '': - output.append(callback(v[i:i+length])) + while v[i : i + length].strip() != "": + output.append(callback(v[i : i + length])) i += length return output -def list_of_items(length): + +def list_of_items(length: int) -> Callable[[str], List[str]]: return partial(list_of, length, lambda v: v.strip()) -def list_of_workareas(name, length): + +def list_of_workareas(name: str, length: int) -> Callable[[str], List[Dict[str, Any]]]: return partial( - list_of, length, - lambda v: parse_workarea(WORK_AREA_LAYOUTS['output'][name], v) + list_of, length, lambda v: parse_workarea(WORK_AREA_LAYOUTS["output"][name], v) ) -def list_of_nodes(v): - return list_of( - 160, - lambda w: list_of(32, list_of_items(8), w), - v - ) -def borough(v): +def list_of_nodes(v: str) -> List[List[List[str]]]: + return list_of(160, lambda w: list_of(32, list_of_items(8), w), v) + + +def borough(v: Optional[Union[str, int]]) -> str: if v: v2 = str(v).strip().upper() @@ -44,16 +43,18 @@ def borough(v): raise GeosupportError("%s is not a valid borough" % v) else: - return '' + return "" -def function(v): + +def function(v: str) -> str: v = str(v).upper().strip() if v in FUNCTIONS: - v = FUNCTIONS[v]['function'] + v = FUNCTIONS[v]["function"] return v -def flag(true, false): - def f(v): + +def flag(true: str, false: str) -> Callable[[Optional[Union[bool, str]]], str]: + def f(v: Optional[Union[bool, str]]) -> str: if type(v) == bool: return true if v else false @@ -64,129 +65,136 @@ def f(v): return f -FORMATTERS = { - # Format input - 'function': function, - 'borough': borough, +FORMATTERS: Dict[str, Any] = { + # Format input + "function": function, + "borough": borough, # Flags - 'auxseg': flag('Y', 'N'), - 'cross_street_names': flag('E', ''), - 'long_work_area_2': flag('L', ''), - 'mode_switch': flag('X', ''), - 'real_streets_only': flag('R', ''), - 'roadbed_request_switch': flag('R', ''), - 'street_name_normalization': flag('C', ''), - 'tpad': flag('Y', 'N'), - + "auxseg": flag("Y", "N"), + "cross_street_names": flag("E", ""), + "long_work_area_2": flag("L", ""), + "mode_switch": flag("X", ""), + "real_streets_only": flag("R", ""), + "roadbed_request_switch": flag("R", ""), + "street_name_normalization": flag("C", ""), + "tpad": flag("Y", "N"), # Parse certain output differently - 'LGI': list_of_workareas('LGI', 53), - 'LGI-extended': list_of_workareas('LGI-extended', 116), - 'BINs': list_of_workareas('BINs', 7), - 'BINs-tpad': list_of_workareas('BINs-tpad', 8), - 'intersections': list_of_workareas('INTERSECTION', 55), - 'node_list': list_of_nodes, - + "LGI": list_of_workareas("LGI", 53), + "LGI-extended": list_of_workareas("LGI-extended", 116), + "BINs": list_of_workareas("BINs", 7), + "BINs-tpad": list_of_workareas("BINs-tpad", 8), + "intersections": list_of_workareas("INTERSECTION", 55), + "node_list": list_of_nodes, # Census Tract formatter - 'CT': lambda v: '' if v is None else v.replace(' ', '0'), - + "CT": lambda v: "" if v is None else v.replace(" ", "0"), # Default formatter - '': lambda v: '' if v is None else str(v).strip().upper() + "": lambda v: "" if v is None else str(v).strip().upper(), } -def get_formatter(name): + +def get_formatter(name: str) -> Callable: if name in FORMATTERS: return FORMATTERS[name] elif name.isdigit(): return list_of_items(int(name)) + return lambda v: "" if v is None else str(v).strip().upper() + -def set_mode(mode): - flags = {} +def set_mode(mode: Optional[str]) -> Dict[str, bool]: + flags: Dict[str, bool] = {} if mode: - if mode == 'extended': - flags['mode_switch'] = True - if 'long' in mode: - flags['long_work_area_2'] = True - if 'tpad' in mode: - flags['tpad'] = True + if mode == "extended": + flags["mode_switch"] = True + if "long" in mode: + flags["long_work_area_2"] = True + if "tpad" in mode: + flags["tpad"] = True return flags -def get_mode(flags): - if flags['mode_switch']: - return 'extended' - elif flags['long_work_area_2'] and flags['tpad']: - return 'long+tpad' - elif flags['long_work_area_2']: - return 'long' + +def get_mode(flags: Dict[str, bool]) -> str: + if flags["mode_switch"]: + return "extended" + elif flags["long_work_area_2"] and flags["tpad"]: + return "long+tpad" + elif flags["long_work_area_2"]: + return "long" else: - return 'regular' + return "regular" -def get_flags(wa1): - layout = WORK_AREA_LAYOUTS['input']['WA1'] + +def get_flags(wa1: str) -> Dict[str, Any]: + layout = WORK_AREA_LAYOUTS["input"]["WA1"] flags = { - 'function': parse_field(layout['function'], wa1), - 'mode_switch': parse_field(layout['mode_switch'], wa1) == 'X', - 'long_work_area_2': parse_field(layout['long_work_area_2'], wa1) == 'L', - 'tpad': parse_field(layout['tpad'], wa1) == 'Y', - 'auxseg': parse_field(layout['auxseg'], wa1) == 'Y' + "function": parse_field(layout["function"], wa1), + "mode_switch": parse_field(layout["mode_switch"], wa1) == "X", + "long_work_area_2": parse_field(layout["long_work_area_2"], wa1) == "L", + "tpad": parse_field(layout["tpad"], wa1) == "Y", + "auxseg": parse_field(layout["auxseg"], wa1) == "Y", } - flags['mode'] = get_mode(flags) + flags["mode"] = get_mode(flags) return flags -def create_wa1(kwargs): - kwargs['work_area_format'] = 'C' - b = bytearray(b' '*1200) + +def create_wa1(kwargs: Dict[str, Any]) -> str: + kwargs["work_area_format"] = "C" + b = bytearray(b" " * 1200) mv = memoryview(b) - layout = WORK_AREA_LAYOUTS['input']['WA1'] + layout = WORK_AREA_LAYOUTS["input"]["WA1"] for key, value in kwargs.items(): - formatter = get_formatter(layout[key]['formatter']) - value = '' if value is None else str(formatter(value)) + formatter = get_formatter(layout[key]["formatter"]) + value = "" if value is None else str(formatter(value)) - i = layout[key]['i'] - length = i[1]-i[0] - mv[i[0]:i[1]] = value.ljust(length)[:length].encode() + i = layout[key]["i"] + length = i[1] - i[0] + mv[i[0] : i[1]] = value.ljust(length)[:length].encode() return str(b.decode()) -def create_wa2(flags): - length = FUNCTIONS[flags['function']][flags['mode']] + +def create_wa2(flags: Dict[str, Any]) -> Optional[str]: + length = FUNCTIONS[flags["function"]][flags["mode"]] if length is None: return None - if flags['auxseg']: + if flags["auxseg"]: length += AUXILIARY_SEGMENT_LENGTH - return ' ' * length + return " " * length -def format_input(kwargs): + +def format_input(kwargs: Dict[str, Any]) -> Tuple[Dict[str, Any], str, Optional[str]]: wa1 = create_wa1(kwargs) flags = get_flags(wa1) - if flags['function'] not in FUNCTIONS: - raise GeosupportError('INVALID FUNCTION CODE', {}) + if flags["function"] not in FUNCTIONS: + raise GeosupportError("INVALID FUNCTION CODE", {}) wa2 = create_wa2(flags) return flags, wa1, wa2 -def parse_field(field, wa): - i = field['i'] - formatter = get_formatter(field['formatter']) - return formatter(wa[i[0]:i[1]]) -def parse_workarea(layout, wa): - output = {} +def parse_field(field: Dict[str, Any], wa: str) -> Any: + i = field["i"] + formatter = get_formatter(field["formatter"]) + return formatter(wa[i[0] : i[1]]) + + +def parse_workarea(layout: Dict[str, Any], wa: str) -> Dict[str, Any]: + output: Dict[str, Any] = {} for key in layout: - if 'i' in layout[key]: + if "i" in layout[key]: output[key] = parse_field(layout[key], wa) else: output[key] = {} @@ -195,27 +203,28 @@ def parse_workarea(layout, wa): return output -def parse_output(flags, wa1, wa2): - output = {} - output.update(parse_workarea(WORK_AREA_LAYOUTS['output']['WA1'], wa1)) +def parse_output(flags: Dict[str, Any], wa1: str, wa2: Optional[str]) -> Dict[str, Any]: + output: Dict[str, Any] = {} + + output.update(parse_workarea(WORK_AREA_LAYOUTS["output"]["WA1"], wa1)) + + if wa2 is None: + return output - function_name = flags['function'] - if function_name in WORK_AREA_LAYOUTS['output']: - output.update(parse_workarea( - WORK_AREA_LAYOUTS['output'][function_name], wa2 - )) + function_name = flags["function"] + if function_name in WORK_AREA_LAYOUTS["output"]: + output.update(parse_workarea(WORK_AREA_LAYOUTS["output"][function_name], wa2)) - function_mode = function_name + '-' + flags['mode'] - if function_mode in WORK_AREA_LAYOUTS['output']: - output.update(parse_workarea( - WORK_AREA_LAYOUTS['output'][function_mode], wa2 - )) + function_mode = function_name + "-" + flags["mode"] + if function_mode in WORK_AREA_LAYOUTS["output"]: + output.update(parse_workarea(WORK_AREA_LAYOUTS["output"][function_mode], wa2)) - if flags['auxseg']: - output.update(parse_workarea( - WORK_AREA_LAYOUTS['output']['AUXSEG'], - wa2[-AUXILIARY_SEGMENT_LENGTH:] - )) + if flags["auxseg"]: + output.update( + parse_workarea( + WORK_AREA_LAYOUTS["output"]["AUXSEG"], wa2[-AUXILIARY_SEGMENT_LENGTH:] + ) + ) return output diff --git a/geosupport/platform_utils.py b/geosupport/platform_utils.py new file mode 100644 index 0000000..d174acb --- /dev/null +++ b/geosupport/platform_utils.py @@ -0,0 +1,61 @@ +import sys +import os +from ctypes import cdll +from .error import GeosupportError +from typing import Optional + + +def load_geosupport_library(geosupport_path: str, py_bit: str) -> any: + """ + Loads the Geosupport library in a platform-specific way. + + For Windows, uses geosupport_path to determine the path to NYCGEO.dll. + For Linux, loads "libgeo.so" (assumed to be in the library path). + """ + if sys.platform.startswith("win"): + from ctypes import windll, WinDLL, wintypes # type: ignore[attr-defined] + + nyc_geo_dll_path = build_win_dll_path(geosupport_path) + if py_bit == "64": + return cdll.LoadLibrary(nyc_geo_dll_path) + else: + return windll.LoadLibrary(nyc_geo_dll_path) + elif sys.platform.startswith("linux"): + # Load the Linux version of the Geosupport library. + lib = cdll.LoadLibrary("libgeo.so") + from ctypes import c_char_p, c_int + + lib.geo.argtypes = [c_char_p, c_char_p] + lib.geo.restype = c_int + return lib + else: + raise GeosupportError("This Operating System is currently not supported.") + + +def build_win_dll_path( + geosupport_path: Optional[str] = None, dll_filename: str = "nycgeo.dll" +) -> str: + """ + Windows-specific function to return the full path of the nycgeo.dll. + Example: 'C:\\Program Files\\Geosupport Desktop Edition\\Bin\\NYCGEO.dll' + """ + if geosupport_path: + return os.path.join(geosupport_path, "bin", dll_filename) + + system_path_entries = os.environ.get("PATH", "").split(";") + # Filter only directories that exist and end with 'bin' (case-insensitive). + bin_directories = [ + b for b in system_path_entries if os.path.isdir(b) and b.lower().endswith("bin") + ] + + for b in bin_directories: + try: + for file in os.listdir(b): + if file.lower() == dll_filename.lower(): + return os.path.join(b, file) + except Exception: + continue + + raise Exception( + f"Unable to locate the {dll_filename} within your system. Ensure the Geosupport 'bin' directory is in your system PATH." + ) diff --git a/geosupport/sysutils.py b/geosupport/sysutils.py deleted file mode 100644 index b414ba9..0000000 --- a/geosupport/sysutils.py +++ /dev/null @@ -1,24 +0,0 @@ -import os - - -def build_win_dll_path(geosupport_path=None, dll_filename='nycgeo.dll'): - """" - Windows specific function to return full path of the nycgeo.dll - example: 'C:\\Program Files\\Geosupport Desktop Edition\\Bin\\NYCGEO.dll' - """ - # return the provided geosupport path with the proper suffix pointing to the dll - if geosupport_path: - return os.path.join(geosupport_path, 'bin', dll_filename) - # otherwise try to find the nycgeo.dll from system path entries - system_path_entries = os.environ['PATH'].split(';') - # look for directories ending with 'bin' since this is where the nycgeo.dll is stored - bin_directories = [b for b in system_path_entries if b.endswith('bin')] - # scan the bin directories for the nycgeo.dll, returning first occurrence. - for b in bin_directories: - file_names = [fn for fn in os.listdir(b)] - for file in file_names: - if file.lower() == dll_filename.lower(): - return os.path.join(b, file) - raise Exception( - "Unable to locate the {0} within your system. Ensure the Geosupport 'bin' directory is in your system path.".format( - dll_filename)) diff --git a/setup.py b/setup.py index 4a4fdbd..4c19eb7 100644 --- a/setup.py +++ b/setup.py @@ -1,45 +1,57 @@ +import re +import os +from setuptools import setup, find_packages -try: - from setuptools import setup -except ImportError: - raise ImportError( - "setuptools module required, please go to " - "https://pypi.python.org/pypi/setuptools and follow the instructions " - "for installing setuptools" +# Read the version from the __init__.py file +with open(os.path.join("geosupport", "__init__.py"), "r", encoding="utf-8") as f: + content = f.read() + version_match = re.search( + r"^__version__\s*=\s*['\"]([^'\"]+)['\"]", content, re.MULTILINE ) + if version_match: + version = version_match.group(1) + else: + raise RuntimeError("Unable to find version string.") -with open("README.md", "r") as fh: +with open("README.md", "r", encoding="utf-8") as fh: long_description = fh.read() setup( - name='python-geosupport', - version='1.0.10', - url='https://github.com/ishiland/python-geosupport', - description='Python bindings for NYC Geosupport Desktop Edition', + name="python-geosupport", + version=version, + url="https://github.com/ishiland/python-geosupport", + description="Python bindings for NYC Geosupport Desktop Edition", long_description=long_description, - long_description_content_type='text/markdown', - author='Ian Shiland, Jeremy Neiman', - author_email='ishiland@gmail.com', - packages=['geosupport'], + long_description_content_type="text/markdown", + project_urls={ + "Bug Tracker": "https://github.com/ishiland/python-geosupport/issues", + "Documentation": "https://python-geosupport.readthedocs.io/en/latest/", + "Source Code": "https://github.com/ishiland/python-geosupport", + }, + author="Ian Shiland, Jeremy Neiman", + author_email="ishiland@gmail.com", + packages=find_packages(), include_package_data=True, - license='MIT', - keywords=['NYC', 'geocoder', 'python-geosupport', 'geosupport'], + license="MIT", + keywords=["NYC", "geocoder", "python-geosupport", "geosupport"], classifiers=[ - 'Operating System :: Microsoft :: Windows', - 'Operating System :: POSIX :: Linux', - 'Programming Language :: Python :: 2.7', - 'Programming Language :: Python :: 3.4', - 'Programming Language :: Python :: 3.5', - 'Programming Language :: Python :: 3.6', - 'Programming Language :: Python :: 3.7', + "Operating System :: Microsoft :: Windows", + "Operating System :: POSIX :: Linux", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", ], + python_requires=">=3.8", test_suite="tests", extras_require={ - 'dev': [ - 'coverage', - 'invoke>=1.1.1', - 'nose' + "dev": [ + "coverage", + "black", ] - } + }, ) diff --git a/tasks.py b/tasks.py deleted file mode 100644 index 2095a0c..0000000 --- a/tasks.py +++ /dev/null @@ -1,21 +0,0 @@ -from invoke import task, run - -nosetests = 'nosetests --with-coverage --cover-package=geosupport --cover-html --cover-branches --cover-erase' - -@task -def test(context, test_type): - if test_type == 'unit': - cmd = ' '.join([nosetests, 'tests/unit/*']) - run('sh -c "%s"' % cmd) - elif test_type == 'functional': - cmd = ' '.join([nosetests, 'tests/functional/*']) - run('sh -c "%s"' % cmd) - elif test_type == 'all': - cmd = ' '.join([nosetests, 'tests/*']) - run('sh -c "%s"' % cmd) - else: - print("Unknown test suite '%s'. Choose one of: unit, functional, all." % test_type) - -@task -def pylint(context): - run('sh -c "pylint geosupport"') diff --git a/tests/functional/test_call.py b/tests/functional/test_call.py index 41815fb..d563138 100644 --- a/tests/functional/test_call.py +++ b/tests/functional/test_call.py @@ -3,447 +3,472 @@ from ..testcase import TestCase + class TestCall(TestCase): def test_invalid_function(self): with self.assertRaises(GeosupportError): - self.geosupport.call({'function': 99}) + self.geosupport.call({"function": 99}) def test_1(self): - result = self.geosupport.call({ - 'function': 1, - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - }) + result = self.geosupport.call( + { + "function": 1, + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + } + ) - self.assertDictSubsetEqual({ - 'ZIP Code': '10013', - 'First Borough Name': 'MANHATTAN', - 'First Street Name Normalized': 'WORTH STREET' - }, result) + self.assertDictSubsetEqual( + { + "ZIP Code": "10013", + "First Borough Name": "MANHATTAN", + "First Street Name Normalized": "WORTH STREET", + }, + result, + ) - self.assertTrue('Physical ID' not in result) + self.assertTrue("Physical ID" not in result) def test_1_extended(self): - result = self.geosupport.call({ - 'function': 1, - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - 'mode_switch': 'X' - }) - - self.assertDictSubsetEqual({ - 'Physical ID': '0079828' - }, result) + result = self.geosupport.call( + { + "function": 1, + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + "mode_switch": "X", + } + ) + + self.assertDictSubsetEqual({"Physical ID": "0079828"}, result) def test_1E(self): - result = self.geosupport.call({ - 'function': '1e', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - }) + result = self.geosupport.call( + { + "function": "1e", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + } + ) - self.assertDictSubsetEqual({ - 'City Council District': '01', - 'State Senatorial District': '27' - }, result) + self.assertDictSubsetEqual( + {"City Council District": "01", "State Senatorial District": "27"}, result + ) - self.assertTrue('Physical ID' not in result) + self.assertTrue("Physical ID" not in result) def test_1A(self): - result = self.geosupport.call({ - 'function': '1a', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - }) + result = self.geosupport.call( + { + "function": "1a", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + } + ) self.assertEqual( - result['BOROUGH BLOCK LOT (BBL)']['BOROUGH BLOCK LOT (BBL)'], - '1001680032' + result["BOROUGH BLOCK LOT (BBL)"]["BOROUGH BLOCK LOT (BBL)"], "1001680032" ) self.assertTrue( - int(result['Number of Entries in List of Geographic Identifiers']) >= 1 + int(result["Number of Entries in List of Geographic Identifiers"]) >= 1 ) self.assertTrue( - 'Street Name' not in result['LIST OF GEOGRAPHIC IDENTIFIERS'][0] + "Street Name" not in result["LIST OF GEOGRAPHIC IDENTIFIERS"][0] ) def test_1A_extended(self): - result = self.geosupport.call({ - 'function': '1a', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - 'mode_switch': 'X' - }) - - self.assertTrue( - 'Street Name' in result['LIST OF GEOGRAPHIC IDENTIFIERS'][0] + result = self.geosupport.call( + { + "function": "1a", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + "mode_switch": "X", + } ) - def test_1A_long(self): - result = self.geosupport.call({ - 'function': '1a', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - 'long_work_area_2': 'L', - }) + self.assertTrue("Street Name" in result["LIST OF GEOGRAPHIC IDENTIFIERS"][0]) - self.assertEqual( - result['Number of Buildings on Tax Lot'], '0001' + def test_1A_long(self): + result = self.geosupport.call( + { + "function": "1a", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + "long_work_area_2": "L", + } ) + self.assertEqual(result["Number of Buildings on Tax Lot"], "0001") + self.assertTrue( - 'TPAD BIN Status' not in result['LIST OF BUILDINGS ON TAX LOT'][0] + "TPAD BIN Status" not in result["LIST OF BUILDINGS ON TAX LOT"][0] ) def test_1A_long_tpad(self): - result = self.geosupport.call({ - 'function': '1a', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - 'long_work_area_2': 'L', - 'tpad': 'Y' - }) - - self.assertTrue( - 'TPAD BIN Status' in result['LIST OF BUILDINGS ON TAX LOT'][0] + result = self.geosupport.call( + { + "function": "1a", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + "long_work_area_2": "L", + "tpad": "Y", + } ) + self.assertTrue("TPAD BIN Status" in result["LIST OF BUILDINGS ON TAX LOT"][0]) + def test_bl_long(self): - result = self.geosupport.call({ - 'function': 'bl', - 'bbl': '1001680032', - 'long_work_area_2': 'L' - }) + result = self.geosupport.call( + {"function": "bl", "bbl": "1001680032", "long_work_area_2": "L"} + ) self.assertEqual( - result['LIST OF BUILDINGS ON TAX LOT'][0]['Building Identification Number (BIN)'], - '1001831' + result["LIST OF BUILDINGS ON TAX LOT"][0][ + "Building Identification Number (BIN)" + ], + "1001831", ) def test_bn(self): - result = self.geosupport.call({ - 'function': 'bn', - 'bin': '1001831' - }) + result = self.geosupport.call({"function": "bn", "bin": "1001831"}) self.assertEqual( - result['BOROUGH BLOCK LOT (BBL)']['BOROUGH BLOCK LOT (BBL)'], - '1001680032' + result["BOROUGH BLOCK LOT (BBL)"]["BOROUGH BLOCK LOT (BBL)"], "1001680032" ) def test_ap(self): - result = self.geosupport.call({ - 'function': 'ap', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - }) - - self.assertDictSubsetEqual({ - 'Number of Entries in List of Geographic Identifiers': '0001', - 'Address Point ID': '001002108' - }, result) + result = self.geosupport.call( + { + "function": "ap", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + } + ) - def test_ap_extended(self): - result = self.geosupport.call({ - 'function': 'ap', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - 'mode_switch': 'X' - }) + self.assertDictSubsetEqual( + { + "Number of Entries in List of Geographic Identifiers": "0001", + "Address Point ID": "001002108", + }, + result, + ) - self.assertTrue( - 'Street Name' in result['LIST OF GEOGRAPHIC IDENTIFIERS'][0] + def test_ap_extended(self): + result = self.geosupport.call( + { + "function": "ap", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + "mode_switch": "X", + } ) + self.assertTrue("Street Name" in result["LIST OF GEOGRAPHIC IDENTIFIERS"][0]) + def test_1b(self): - result = self.geosupport.call({ - 'function': '1b', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - }) - - self.assertDictSubsetEqual({ - 'Physical ID': '0079828', - 'From LION Node ID': '0015487', - 'To LION Node ID': '0015490', - 'Blockface ID': '0212261942' - }, result) + result = self.geosupport.call( + { + "function": "1b", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + } + ) + + self.assertDictSubsetEqual( + { + "Physical ID": "0079828", + "From LION Node ID": "0015487", + "To LION Node ID": "0015490", + "Blockface ID": "0212261942", + }, + result, + ) def test_2(self): - result = self.geosupport.call({ - 'function': 2, - 'borough_code': 'MN', - 'street_name': 'Worth St', - 'street_name_2': 'Centre St' - }) - - self.assertDictSubsetEqual({ - 'LION Node Number': '0015490', - 'Number of Intersecting Streets': '2' - }, result) + result = self.geosupport.call( + { + "function": 2, + "borough_code": "MN", + "street_name": "Worth St", + "street_name_2": "Centre St", + } + ) + + self.assertDictSubsetEqual( + {"LION Node Number": "0015490", "Number of Intersecting Streets": "2"}, + result, + ) def test_2_more_than_2_intersections(self): with self.assertRaises(GeosupportError) as cm: - result = self.geosupport.call({ - 'function': '2', - 'borough_code': 'BK', - 'street_name': 'E 19 St', - 'street_name_2': 'Dead End' - }) + result = self.geosupport.call( + { + "function": "2", + "borough_code": "BK", + "street_name": "E 19 St", + "street_name_2": "Dead End", + } + ) self.assertEqual( str(cm.exception), - 'STREETS INTERSECT MORE THAN TWICE-USE FUNCTION 2W TO FIND RELATED NODES ' + "STREETS INTERSECT MORE THAN TWICE-USE FUNCTION 2W TO FIND RELATED NODES ", ) def test_2W_more_than_2_intersections(self): with self.assertRaises(GeosupportError) as cm: - result = self.geosupport.call({ - 'function': '2w', - 'borough_code': 'BK', - 'street_name': 'grand army plaza oval', - 'street_name_2': 'plaza street east' - }) + result = self.geosupport.call( + { + "function": "2w", + "borough_code": "BK", + "street_name": "grand army plaza oval", + "street_name_2": "plaza street east", + } + ) self.assertEqual( str(cm.exception), - 'PLAZA STREET EAST IS AN INVALID STREET NAME FOR THIS LOCATION ' + "PLAZA STREET EAST IS AN INVALID STREET NAME FOR THIS LOCATION ", ) - self.assertEqual(len(cm.exception.result['List of Street Codes']), 2) - self.assertEqual(len(cm.exception.result['List of Street Names']), 2) + self.assertEqual(len(cm.exception.result["List of Street Codes"]), 2) + self.assertEqual(len(cm.exception.result["List of Street Names"]), 2) - self.assertEqual(cm.exception.result['Node Number'], '') + self.assertEqual(cm.exception.result["Node Number"], "") def test_2W_with_node(self): - result = self.geosupport.call({ - 'function': '2w', - 'node': '0104434' - }) + result = self.geosupport.call({"function": "2w", "node": "0104434"}) - self.assertTrue( - 'GRAND ARMY PLAZA OVAL' in result['List of Street Names'] - ) + self.assertTrue("GRAND ARMY PLAZA OVAL" in result["List of Street Names"]) - self.assertTrue( - 'PLAZA STREET' in result['List of Street Names'] - ) + self.assertTrue("PLAZA STREET" in result["List of Street Names"]) def test_3(self): - result = self.geosupport.call({ - 'function': 3, - 'borough_code': 'MN', - 'on': 'Lafayette St', - 'from': 'Worth st', - 'to': 'Leonard St' - }) + result = self.geosupport.call( + { + "function": 3, + "borough_code": "MN", + "on": "Lafayette St", + "from": "Worth st", + "to": "Leonard St", + } + ) - self.assertDictSubsetEqual({ - 'From Node': '0015487', - 'To Node': '0020353' - }, result) + self.assertDictSubsetEqual( + {"From Node": "0015487", "To Node": "0020353"}, result + ) - self.assertTrue('Segment IDs' not in result) + self.assertTrue("Segment IDs" not in result) def test_3_auxseg(self): - result = self.geosupport.call({ - 'function': 3, - 'borough_code': 'MN', - 'on': 'Lafayette St', - 'from': 'Worth st', - 'to': 'Leonard St', - 'auxseg': 'Y' - }) - - self.assertEqual(len(result['Segment IDs']), 2) - self.assertTrue('0023578' in result['Segment IDs']) - self.assertTrue('0032059' in result['Segment IDs']) + result = self.geosupport.call( + { + "function": 3, + "borough_code": "MN", + "on": "Lafayette St", + "from": "Worth st", + "to": "Leonard St", + "auxseg": "Y", + "mode_switch": "X", + } + ) + + self.assertEqual(len(result["Segment IDs"]), 2) + self.assertTrue("0023578" in result["Segment IDs"]) + self.assertTrue("0032059" in result["Segment IDs"]) def test_3_extended(self): - result = self.geosupport.call({ - 'function': 3, - 'borough_code': 'MN', - 'on': 'Lafayette St', - 'from': 'Worth st', - 'to': 'Leonard St', - 'mode_switch': 'X' - }) - - self.assertDictSubsetEqual({ - 'From Node': '0015487', - 'To Node': '0020353', - 'Left 2020 Community District Tabulation Area (CDTA)': 'MN01' - }, result) - - self.assertTrue('Segment IDs' not in result) + result = self.geosupport.call( + { + "function": 3, + "borough_code": "MN", + "on": "Lafayette St", + "from": "Worth st", + "to": "Leonard St", + "mode_switch": "X", + } + ) + + self.assertDictSubsetEqual( + { + "From Node": "0015487", + "To Node": "0020353", + "Left 2020 Community District Tabulation Area (CDTA)": "MN01", + }, + result, + ) + + self.assertTrue("Segment IDs" not in result) def test_3_extended_auxseg(self): - result = self.geosupport.call({ - 'function': 3, - 'borough_code': 'MN', - 'on': 'Lafayette St', - 'from': 'Worth st', - 'to': 'Leonard St', - 'auxseg': 'Y', - 'mode_switch': 'X' - }) - - self.assertDictSubsetEqual({ - 'From Node': '0015487', - 'To Node': '0020353', - 'Left 2020 Community District Tabulation Area (CDTA)': 'MN01' - }, result) - - self.assertEqual(len(result['Segment IDs']), 2) - self.assertTrue('0023578' in result['Segment IDs']) - self.assertTrue('0032059' in result['Segment IDs']) + result = self.geosupport.call( + { + "function": 3, + "borough_code": "MN", + "on": "Lafayette St", + "from": "Worth st", + "to": "Leonard St", + "auxseg": "Y", + "mode_switch": "X", + } + ) + + self.assertDictSubsetEqual( + { + "From Node": "0015487", + "To Node": "0020353", + "Left 2020 Community District Tabulation Area (CDTA)": "MN01", + }, + result, + ) + + self.assertEqual(len(result["Segment IDs"]), 2) + self.assertTrue("0023578" in result["Segment IDs"]) + self.assertTrue("0032059" in result["Segment IDs"]) def test_3C(self): - result = self.geosupport.call({ - 'function': '3c', - 'borough_code': 'MN', - 'on': 'Lafayette St', - 'from': 'Worth st', - 'to': 'Leonard St', - 'compass_direction': 'E' - }) - - self.assertDictSubsetEqual({ - 'From Node': '0015487', - 'To Node': '0020353', - 'Side-of-Street Indicator': 'R' - }, result) - - self.assertTrue('Segment IDs' not in result) + result = self.geosupport.call( + { + "function": "3c", + "borough_code": "MN", + "on": "Lafayette St", + "from": "Worth st", + "to": "Leonard St", + "compass_direction": "E", + } + ) + + self.assertDictSubsetEqual( + { + "From Node": "0015487", + "To Node": "0020353", + "Side-of-Street Indicator": "R", + }, + result, + ) + + self.assertTrue("Segment IDs" not in result) def test_3C_auxseg(self): - result = self.geosupport.call({ - 'function': '3c', - 'borough_code': 'MN', - 'on': 'Lafayette St', - 'from': 'Worth st', - 'to': 'Leonard St', - 'compass_direction': 'E', - 'auxseg': 'Y' - }) - - self.assertDictSubsetEqual({ - 'From Node': '0015487', - 'To Node': '0020353', - 'Side-of-Street Indicator': 'R' - }, result) - - self.assertEqual(len(result['Segment IDs']), 2) - self.assertTrue('0023578' in result['Segment IDs']) - self.assertTrue('0032059' in result['Segment IDs']) + result = self.geosupport.call( + { + "function": "3c", + "borough_code": "MN", + "on": "Lafayette St", + "from": "Worth st", + "to": "Leonard St", + "compass_direction": "E", + "auxseg": "Y", + "mode_switch": "X", + } + ) + + self.assertDictSubsetEqual( + { + "From Node": "0015487", + "To Node": "0020353", + "Side-of-Street Indicator": "R", + }, + result, + ) + + self.assertEqual(len(result["Segment IDs"]), 2) + self.assertTrue("7800320" in result["Segment IDs"]) + self.assertTrue("59" in result["Segment IDs"]) def test_3C_extended_auxseg(self): - result = self.geosupport.call({ - 'function': '3c', - 'borough_code': 'MN', - 'on': 'Lafayette St', - 'from': 'Worth st', - 'to': 'Leonard St', - 'compass_direction': 'E', - 'auxseg': 'Y', - 'mode_switch': 'X' - }) - - self.assertDictSubsetEqual({ - 'From Node': '0015487', - 'To Node': '0020353', - 'Side-of-Street Indicator': 'R', - 'Blockface ID': '0212262072' - - }, result) - - self.assertEqual(len(result['Segment IDs']), 2) - self.assertTrue('7800320' in result['Segment IDs']) - self.assertTrue('59' in result['Segment IDs']) + result = self.geosupport.call( + { + "function": "3c", + "borough_code": "MN", + "on": "Lafayette St", + "from": "Worth st", + "to": "Leonard St", + "compass_direction": "E", + "auxseg": "Y", + "mode_switch": "X", + } + ) + + self.assertDictSubsetEqual( + { + "From Node": "0015487", + "To Node": "0020353", + "Side-of-Street Indicator": "R", + "Blockface ID": "0212262072", + }, + result, + ) + + self.assertEqual(len(result["Segment IDs"]), 2) + self.assertTrue("7800320" in result["Segment IDs"]) + self.assertTrue("59" in result["Segment IDs"]) def test_3S(self): - result = self.geosupport.call({ - 'function': '3S', - 'borough_code': 'MN', - 'on': 'Broadway', - 'from': 'worth st', - 'to': 'Liberty st', - }) - - self.assertEqual(result['Number of Intersections'], '017') + result = self.geosupport.call( + { + "function": "3S", + "borough_code": "MN", + "on": "Broadway", + "from": "worth st", + "to": "Liberty st", + } + ) + + self.assertEqual(result["Number of Intersections"], "017") self.assertEqual( - len(result['LIST OF INTERSECTIONS']), - int(result['Number of Intersections']) + len(result["LIST OF INTERSECTIONS"]), int(result["Number of Intersections"]) ) def test_D(self): - result = self.geosupport.call({ - 'function': 'D', - 'B7SC': '145490' - }) + result = self.geosupport.call({"function": "D", "B7SC": "145490"}) - self.assertEqual(result['First Street Name Normalized'], 'WORTH STREET') + self.assertEqual(result["First Street Name Normalized"], "WORTH STREET") def test_DG(self): - result = self.geosupport.call({ - 'function': 'DG', - 'b7sc': '14549001' - }) + result = self.geosupport.call({"function": "DG", "b7sc": "14549001"}) - self.assertEqual(result['First Street Name Normalized'], 'WORTH STREET') + self.assertEqual(result["First Street Name Normalized"], "WORTH STREET") def test_DN(self): - result = self.geosupport.call({ - 'function': 'DN', - 'B7SC': '14549001010' - }) + result = self.geosupport.call({"function": "DN", "B7SC": "14549001010"}) - self.assertEqual(result['First Street Name Normalized'], 'WORTH STREET') + self.assertEqual(result["First Street Name Normalized"], "WORTH STREET") def test_1N(self): - result = self.geosupport.call({ - 'function': '1N', - 'borough_code': 'MN', - 'street': 'Worth str' - }) + result = self.geosupport.call( + {"function": "1N", "borough_code": "MN", "street": "Worth str"} + ) - self.assertEqual(result['First Street Name Normalized'], 'WORTH STREET') + self.assertEqual(result["First Street Name Normalized"], "WORTH STREET") def test_Nstar(self): - result = self.geosupport.call({ - 'function': 'N*', - 'street': 'fake cir' - }) + result = self.geosupport.call({"function": "N*", "street": "fake cir"}) - self.assertEqual(result['First Street Name Normalized'], 'FAKE CIRCLE') + self.assertEqual(result["First Street Name Normalized"], "FAKE CIRCLE") def test_BF(self): - result = self.geosupport.call({ - 'func': 'BF', - 'borough_code': 'MN', - 'street': 'WORTH' - }) + result = self.geosupport.call( + {"func": "BF", "borough_code": "MN", "street": "WORTH"} + ) - self.assertTrue('WORTH STREET' in result['List of Street Names']) + self.assertTrue("WORTH STREET" in result["List of Street Names"]) def test_BB(self): - result = self.geosupport.call({ - 'func': 'BB', - 'borough_code': 'MN', - 'street': 'WORTH' - }) + result = self.geosupport.call( + {"func": "BB", "borough_code": "MN", "street": "WORTH"} + ) - self.assertTrue('WORLDWIDE PLAZA' in result['List of Street Names']) + self.assertTrue("WORLDWIDE PLAZA" in result["List of Street Names"]) diff --git a/tests/functional/test_call_alternate.py b/tests/functional/test_call_alternate.py index c0b6f18..34566e8 100644 --- a/tests/functional/test_call_alternate.py +++ b/tests/functional/test_call_alternate.py @@ -3,124 +3,146 @@ from ..testcase import TestCase + class TestCallByName(TestCase): def test_address(self): - result = self.geosupport.address({ - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - }) - - self.assertDictSubsetEqual({ - 'Physical ID': '0079828', - 'From LION Node ID': '0015487', - 'To LION Node ID': '0015490', - 'Blockface ID': '0212261942' - }, result) + result = self.geosupport.address( + { + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + } + ) + + self.assertDictSubsetEqual( + { + "Physical ID": "0079828", + "From LION Node ID": "0015487", + "To LION Node ID": "0015490", + "Blockface ID": "0212261942", + }, + result, + ) def test_address_upper(self): - result = self.geosupport.ADDRESS({ - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - }) - - self.assertDictSubsetEqual({ - 'Physical ID': '0079828', - 'From LION Node ID': '0015487', - 'To LION Node ID': '0015490', - 'Blockface ID': '0212261942' - }, result) + result = self.geosupport.ADDRESS( + { + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + } + ) + + self.assertDictSubsetEqual( + { + "Physical ID": "0079828", + "From LION Node ID": "0015487", + "To LION Node ID": "0015490", + "Blockface ID": "0212261942", + }, + result, + ) def test_1(self): - result = self.geosupport['1']({ - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn', - }) + result = self.geosupport["1"]( + { + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + } + ) - self.assertDictSubsetEqual({ - 'ZIP Code': '10013', - 'First Borough Name': 'MANHATTAN', - 'First Street Name Normalized': 'WORTH STREET' - }, result) + self.assertDictSubsetEqual( + { + "ZIP Code": "10013", + "First Borough Name": "MANHATTAN", + "First Street Name Normalized": "WORTH STREET", + }, + result, + ) - self.assertTrue('Physical ID' not in result) + self.assertTrue("Physical ID" not in result) def test_1A_extended_mode_parameter(self): - result = self.geosupport.call({ - 'function': '1a', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn' - }, mode='extended') - - self.assertTrue( - 'Street Name' in result['LIST OF GEOGRAPHIC IDENTIFIERS'][0] + result = self.geosupport.call( + { + "function": "1a", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + }, + mode="extended", ) + self.assertTrue("Street Name" in result["LIST OF GEOGRAPHIC IDENTIFIERS"][0]) + def test_1A_long_mode_parameter(self): - result = self.geosupport.call({ - 'function': '1a', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn' - }, mode='long') - - self.assertEqual( - result['Number of Buildings on Tax Lot'], '0001' + result = self.geosupport.call( + { + "function": "1a", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + }, + mode="long", ) + self.assertEqual(result["Number of Buildings on Tax Lot"], "0001") + self.assertTrue( - 'TPAD BIN Status' not in result['LIST OF BUILDINGS ON TAX LOT'][0] + "TPAD BIN Status" not in result["LIST OF BUILDINGS ON TAX LOT"][0] ) def test_1A_long_tpad_mode_parameter(self): - result = self.geosupport.call({ - 'function': '1a', - 'house_number': '125', - 'street_name': 'Worth St', - 'borough_code': 'Mn' - }, mode='long+tpad') - - self.assertTrue( - 'TPAD BIN Status' in result['LIST OF BUILDINGS ON TAX LOT'][0] + result = self.geosupport.call( + { + "function": "1a", + "house_number": "125", + "street_name": "Worth St", + "borough_code": "Mn", + }, + mode="long+tpad", ) + self.assertTrue("TPAD BIN Status" in result["LIST OF BUILDINGS ON TAX LOT"][0]) + def test_1_kwargs(self): - result = self.geosupport['1']( - house_number='125', - street_name='Worth St', - borough_code='Mn', + result = self.geosupport["1"]( + house_number="125", + street_name="Worth St", + borough_code="Mn", ) - self.assertDictSubsetEqual({ - 'ZIP Code': '10013', - 'First Borough Name': 'MANHATTAN', - 'First Street Name Normalized': 'WORTH STREET' - }, result) + self.assertDictSubsetEqual( + { + "ZIP Code": "10013", + "First Borough Name": "MANHATTAN", + "First Street Name Normalized": "WORTH STREET", + }, + result, + ) - self.assertTrue('Physical ID' not in result) + self.assertTrue("Physical ID" not in result) def test_call_invalid_function(self): with self.assertRaises(AttributeError): self.geosupport.fake({}) with self.assertRaises(AttributeError): - self.geosupport['fake']({}) + self.geosupport["fake"]({}) def test_call_invalid_key(self): with self.assertRaises(KeyError): self.geosupport.intersection( - borough_code='BK', street1='east 19 st', street_2='ave h' + borough_code="BK", street1="east 19 st", street_2="ave h" ) with self.assertRaises(KeyError): self.geosupport.intersection( - borough_code='BK', street_1='east 19 st', street2='ave h' + borough_code="BK", street_1="east 19 st", street2="ave h" ) self.geosupport.intersection( - borough_code='BK', street_name_1='east 19 st', street_name_2='ave h' + borough_code="BK", street_name_1="east 19 st", street_name_2="ave h" ) diff --git a/tests/testcase.py b/tests/testcase.py index 5ae5d37..8dde91b 100644 --- a/tests/testcase.py +++ b/tests/testcase.py @@ -2,6 +2,7 @@ from geosupport import Geosupport + class TestCase(unittest.TestCase): def assertDictSubsetEqual(self, subset, superset): diff --git a/tests/unit/test_error.py b/tests/unit/test_error.py index 6380363..4c91b0e 100644 --- a/tests/unit/test_error.py +++ b/tests/unit/test_error.py @@ -2,6 +2,7 @@ from ..testcase import TestCase + class TestError(TestCase): def test_error(self): diff --git a/tests/unit/test_function_info.py b/tests/unit/test_function_info.py index e5d0af2..a7d2f84 100644 --- a/tests/unit/test_function_info.py +++ b/tests/unit/test_function_info.py @@ -2,14 +2,15 @@ from ..testcase import TestCase + class TestFunctionInfo(TestCase): def test_function_dict(self): d = FunctionDict() - d['A'] = {} + d["A"] = {} - self.assertTrue('a' in d) - self.assertTrue('A' in d) - self.assertEqual(d['a'], d['A']) + self.assertTrue("a" in d) + self.assertTrue("A" in d) + self.assertEqual(d["a"], d["A"]) - self.assertFalse('b' in d) + self.assertFalse("b" in d) diff --git a/tests/unit/test_help.py b/tests/unit/test_help.py index 94c1731..05d8279 100644 --- a/tests/unit/test_help.py +++ b/tests/unit/test_help.py @@ -29,9 +29,11 @@ def test_print(self): try: import StringIO # python 2 + h = StringIO.StringIO() except ImportError: import io # python 3 + h = io.StringIO() sys.stdout = h diff --git a/tests/unit/test_io.py b/tests/unit/test_io.py index b4b124f..24db498 100644 --- a/tests/unit/test_io.py +++ b/tests/unit/test_io.py @@ -1,38 +1,37 @@ from geosupport.error import GeosupportError -from geosupport.io import ( - list_of, list_of_items, borough, flag -) +from geosupport.io import list_of, list_of_items, borough, flag from ..testcase import TestCase + class TestIO(TestCase): def test_list_of(self): result = list_of(3, lambda v: v.strip(), "a b c ") - self.assertEqual(result, ['a', 'b', 'c']) + self.assertEqual(result, ["a", "b", "c"]) def test_list_of_items(self): result = list_of_items(3)("a b c ") - self.assertEqual(result, ['a', 'b', 'c']) + self.assertEqual(result, ["a", "b", "c"]) def test_borough(self): - self.assertEqual(borough('MN'), '1') - self.assertEqual(borough('queens'), '4') - self.assertEqual(borough(None), '') - self.assertEqual(borough(''), '') - self.assertEqual(borough(1), '1') - self.assertEqual(borough('2'), '2') + self.assertEqual(borough("MN"), "1") + self.assertEqual(borough("queens"), "4") + self.assertEqual(borough(None), "") + self.assertEqual(borough(""), "") + self.assertEqual(borough(1), "1") + self.assertEqual(borough("2"), "2") with self.assertRaises(GeosupportError): - borough('Fake') + borough("Fake") def test_flag(self): - f = flag('Y', 'N') - self.assertEqual(f(True), 'Y') - self.assertEqual(f(False), 'N') - self.assertEqual(f('Y'), 'Y') - self.assertEqual(f('y'), 'Y') - self.assertEqual(f('n'), 'N') - self.assertEqual(f(''), 'N') - self.assertEqual(f(None), 'N') - self.assertEqual(f('Yes'), 'Y') + f = flag("Y", "N") + self.assertEqual(f(True), "Y") + self.assertEqual(f(False), "N") + self.assertEqual(f("Y"), "Y") + self.assertEqual(f("y"), "Y") + self.assertEqual(f("n"), "N") + self.assertEqual(f(""), "N") + self.assertEqual(f(None), "N") + self.assertEqual(f("Yes"), "Y") diff --git a/tests/unit/test_platform_utils.py b/tests/unit/test_platform_utils.py new file mode 100644 index 0000000..7e01ebd --- /dev/null +++ b/tests/unit/test_platform_utils.py @@ -0,0 +1,49 @@ +import os +import sys +from unittest import TestCase, mock, skipUnless +from geosupport.platform_utils import build_win_dll_path + + +class TestSysUtils(TestCase): + + @skipUnless(sys.platform.startswith("win"), "requires Windows") + def test_build_dll_path_with_geosupport_path(self): + """test that the dll path is created from the provided geosupport path""" + dll_path = build_win_dll_path(geosupport_path=r"C:\somewhere\on\my\pc") + self.assertEqual(dll_path.lower(), r"c:\somewhere\on\my\pc\bin\nycgeo.dll") + + @skipUnless(sys.platform.startswith("win"), "requires Windows") + @mock.patch.dict( + os.environ, + { + "PATH": r"C:\Program Files\Python311\Scripts\;C:\Program Files\Python311\;c:\another\place\on\my\pc\bin" + }, + ) + def test_build_dll_path_with_geosupport_path_none(self): + """test that the dll path is created when geosupport path is not provided""" + + # Create a function to selectively mock isdir for our test path + def mock_isdir(path): + return path.lower() == r"c:\another\place\on\my\pc\bin" + + # Mock both isdir and listdir + with mock.patch("os.path.isdir", side_effect=mock_isdir): + with mock.patch("os.listdir") as mocked_listdir: + mocked_listdir.return_value = [ + "geo.dll", + "docs", + "nycgeo.exe", + "nycgeo.dll", + ] + dll_path = build_win_dll_path(geosupport_path=None) + self.assertEqual( + dll_path.lower(), r"c:\another\place\on\my\pc\bin\nycgeo.dll" + ) + + @skipUnless(sys.platform.startswith("win"), "requires Windows") + @mock.patch.dict(os.environ, {"PATH": "just a bunch of nonsense"}) + def test_build_dll_path_raise_exception(self): + """test that an exception is raised when the nycgeo.dll is not found""" + with self.assertRaises(Exception) as context: + build_win_dll_path(geosupport_path=None) + self.assertTrue("Unable to locate the nycgeo.dll" in context.exception) diff --git a/tests/unit/test_sysutils.py b/tests/unit/test_sysutils.py deleted file mode 100644 index e39847c..0000000 --- a/tests/unit/test_sysutils.py +++ /dev/null @@ -1,31 +0,0 @@ -import os -import sys -from unittest import TestCase, mock, skipUnless -from geosupport.sysutils import build_win_dll_path - - -class TestSysUtils(TestCase): - - @skipUnless(sys.platform.startswith("win"), "requires Windows") - def test_build_dll_path_with_geosupport_path(self): - """ test that the dll path is created from the provided geosupport path""" - dll_path = build_win_dll_path(geosupport_path=r'C:\somewhere\on\my\pc') - self.assertEqual(dll_path.lower(), r'c:\somewhere\on\my\pc\bin\nycgeo.dll') - - @skipUnless(sys.platform.startswith("win"), "requires Windows") - @mock.patch.dict(os.environ, { - "PATH": r"C:\Program Files\Python311\Scripts\;C:\Program Files\Python311\;c:\another\place\on\my\pc\bin"}) - def test_build_dll_path_with_geosupport_path_none(self): - """ test that the dll path is created when geosupport path is not provided""" - with mock.patch('os.listdir') as mocked_listdir: - mocked_listdir.return_value = ['geo.dll', 'docs', 'nycgeo.exe', 'nycgeo.dll'] - dll_path = build_win_dll_path(geosupport_path=None) - self.assertEqual(dll_path.lower(), r'c:\another\place\on\my\pc\bin\nycgeo.dll') - - @skipUnless(sys.platform.startswith("win"), "requires Windows") - @mock.patch.dict(os.environ, {"PATH": "just a bunch of nonsense"}) - def test_build_dll_path_raise_exception(self): - """ test that an exception is raised when the nycgeo.dll is not found""" - with self.assertRaises(Exception) as context: - build_win_dll_path(geosupport_path=None) - self.assertTrue('Unable to locate the nycgeo.dll' in context.exception)