Skip to content

Commit

Permalink
remove pystac
Browse files Browse the repository at this point in the history
  • Loading branch information
jonhealy1 committed Sep 1, 2022
1 parent 95d1d05 commit 1756c6a
Show file tree
Hide file tree
Showing 2 changed files with 2 additions and 171 deletions.
165 changes: 1 addition & 164 deletions stac_validator/utilities.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,7 @@
import functools
import json
import os
from enum import Enum
from typing import Optional, cast
from urllib.parse import ParseResult as URLParseResult
from urllib.parse import urljoin, urlparse, urlunparse
from urllib.parse import urlparse
from urllib.request import urlopen

import requests # type: ignore
Expand Down Expand Up @@ -93,163 +90,3 @@ def link_request(
else:
initial_message["request_invalid"].append(link["href"])
initial_message["format_invalid"].append(link["href"])


""" Utility functions below this comment have been copied from
https://github.com/stac-utils/pystac without any editing.
They were added to handle relative schema paths."""


def safe_urlparse(href: str) -> URLParseResult:
"""Wrapper around :func:`urllib.parse.urlparse` that returns consistent results for
both Windows and UNIX file paths.
For Windows paths, this function will include the drive prefix (e.g. ``"D:\\"``) as
part of the ``path`` of the :class:`urllib.parse.ParseResult` rather than as the
``scheme`` for consistency with handling of UNIX/LINUX file paths.
Args:
href (str) : The HREF to parse. May be a local file path or URL.
Returns:
urllib.parse.ParseResult : The named tuple representing the parsed HREF.
"""
parsed = urlparse(href)
if parsed.scheme != "" and href.lower().startswith("{}:\\".format(parsed.scheme)):
return URLParseResult(
scheme="",
netloc="",
path="{}:{}".format(
# We use this more complicated formulation because parsed.scheme
# converts to lower-case
href[: len(parsed.scheme)],
parsed.path,
),
params=parsed.params,
query=parsed.query,
fragment=parsed.fragment,
)
else:
return parsed


def _make_absolute_href_url(
parsed_source: URLParseResult,
parsed_start: URLParseResult,
start_is_dir: bool = False,
) -> str:

# If the source is already absolute, just return it
if parsed_source.scheme != "":
return urlunparse(parsed_source)

# If the start path is not a directory, get the parent directory
if start_is_dir:
start_dir = parsed_start.path
else:
# Ensure the directory has a trailing slash so urljoin works properly
start_dir = parsed_start.path.rsplit("/", 1)[0] + "/"

# Join the start directory to the relative path and find the absolute path
abs_path = urljoin(start_dir, parsed_source.path)
abs_path = abs_path.replace("\\", "/")

return urlunparse(
(
parsed_start.scheme,
parsed_start.netloc,
abs_path,
parsed_source.params,
parsed_source.query,
parsed_source.fragment,
)
)


def _make_absolute_href_path(
parsed_source: URLParseResult,
parsed_start: URLParseResult,
start_is_dir: bool = False,
) -> str:

# If the source is already absolute, just return it
if _pathlib.isabs(parsed_source.path):
return urlunparse(parsed_source)

# If the start path is not a directory, get the parent directory
start_dir = (
parsed_start.path if start_is_dir else _pathlib.dirname(parsed_start.path)
)

# Join the start directory to the relative path and find the absolute path
abs_path = _pathlib.abspath(_pathlib.join(start_dir, parsed_source.path))

# Account for the normalization of abspath for
# things like /vsitar// prefixes by replacing the
# original start_dir text when abspath modifies the start_dir.
if not start_dir == _pathlib.abspath(start_dir):
abs_path = abs_path.replace(_pathlib.abspath(start_dir), start_dir)

return abs_path


class StringEnum(str, Enum):
"""Base :class:`enum.Enum` class for string enums that will serialize as the string
value."""

def __str__(self) -> str:
return cast(str, self.value)


class JoinType(StringEnum):
"""Allowed join types for :func:`~pystac.utils.join_path_or_url`."""

@staticmethod
def from_parsed_uri(parsed_uri: URLParseResult) -> "JoinType":
"""Determines the appropriate join type based on the scheme of the parsed
result.
Args:
parsed_uri (urllib.parse.ParseResult) : A named tuple representing the
parsed URI.
Returns:
JoinType : The join type for the URI.
"""
if parsed_uri.scheme == "":
return JoinType.PATH
else:
return JoinType.URL

PATH = "path"
URL = "url"


def make_absolute_href(
source_href: str, start_href: Optional[str] = None, start_is_dir: bool = False
) -> str:
"""Returns a new string that represents ``source_href`` as an absolute path. If
``source_href`` is already absolute it is returned unchanged. If ``source_href``
is relative, the absolute HREF is constructed by joining ``source_href`` to
``start_href``.
May be used on either local file paths or URLs.
Args:
source_href : The HREF to make absolute.
start_href : The HREF that will be used as the basis for resolving relative
paths, if ``source_href`` is a relative path. Defaults to the current
working directory.
start_is_dir : If ``True``, ``start_href`` is treated as a directory.
Otherwise, ``start_href`` is considered to be a path to a file. Defaults to
``False``.
Returns:
str: The absolute HREF.
"""
if start_href is None:
start_href = os.getcwd()
start_is_dir = True

parsed_start = safe_urlparse(start_href)
parsed_source = safe_urlparse(source_href)

if (
JoinType.from_parsed_uri(parsed_source) == JoinType.URL
or JoinType.from_parsed_uri(parsed_start) == JoinType.URL
):
return _make_absolute_href_url(parsed_source, parsed_start, start_is_dir)
else:
return _make_absolute_href_path(parsed_source, parsed_start, start_is_dir)
8 changes: 1 addition & 7 deletions stac_validator/validate.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
get_stac_type,
is_valid_url,
link_request,
make_absolute_href,
set_schema_addr,
)

Expand Down Expand Up @@ -168,17 +167,12 @@ def custom_validator(self):
resolver = RefResolver(custom_uri, self.custom)
jsonschema.validate(self.stac_content, schema, resolver=resolver)
# deal with a relative path in the schema
elif self.custom.startswith("."):
else:
file_directory = os.path.dirname(os.path.abspath(self.stac_file))
self.custom = os.path.join(file_directory, self.custom)
self.custom = os.path.abspath(os.path.realpath(self.custom))
schema = fetch_and_parse_schema(self.custom)
jsonschema.validate(self.stac_content, schema)
# use pystac code
else:
href = make_absolute_href(self.custom, self.stac_file)
schema = fetch_and_parse_schema(href)
jsonschema.validate(self.stac_content, schema)

def core_validator(self, stac_type: str):
stac_type = stac_type.lower()
Expand Down

0 comments on commit 1756c6a

Please sign in to comment.