diff --git a/.dockerignore b/.dockerignore
index 1cdb253059fe20ed6b6229d896d19f7074680fee..63410b57cc6cdbaa2fce4f58ae2a0a05bc1c8503 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -1,8 +1,10 @@
.coverage
.DS_Store
+.eggs
.git
+.mypy_cache
.pytest_cache
coverage.xml
-node_modules
-test
+build
+dist
*.pyc
diff --git a/.gitignore b/.gitignore
index 12c925f9cade217df72225a9e39a8d97c9b6dc67..574fe32b04533cd4854491fea23fa54a18abd6d3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,10 +2,10 @@
cov.xml
.DS_Store
.eggs
-node_modules
*.pyc
coverage.xml
.ipynb_checkpoints*
+.mypy_cache
.pytest_cache
htmlcov
.vscode
diff --git a/Dockerfile b/Dockerfile
index 17ee426a6147fb4c9ed9901abee4889c411398fe..8bcc278acc50baec3852ba31e89ae8de083d8f49 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,8 +1,57 @@
-ARG FROM_IMAGE=usgs/obspy:3.8
+ARG FROM_IMAGE=usgs/centos:7
+
+
+################################################################################
+# base python image
+
+FROM ${FROM_IMAGE} as python
+ARG PYTHON_VERSION=38
-FROM ${FROM_IMAGE}
LABEL maintainer="Jeremy Fee <jmfee@usgs.gov>"
+# put rh-python38 at start of path
+ENV PATH="/opt/rh/rh-python38/root/usr/bin:/opt/rh/rh-python38/root/usr/local/bin:${PATH}"
+# configure ssl intercept
+ENV PIP_CERT="${SSL_CERT_FILE}"
+ENV REQUESTS_CA_BUNDLE="${SSL_CERT_FILE}"
+
+RUN yum install -y \
+ centos-release-scl \
+ glibc-langpack-en \
+ which \
+ && yum install -y rh-python${PYTHON_VERSION} \
+ && python -m pip install -U \
+ pip \
+ poetry \
+ wheel \
+ && yum clean all
+
+
+################################################################################
+# python builder image
+
+FROM python as obspy
+ARG PYTHON_VERSION=38
+
+# build with compilers in separate stage
+RUN yum groupinstall -y "Development Tools"
+RUN yum install -y rh-python${PYTHON_VERSION}-python-devel
+ENV LD_LIBRARY_PATH="/opt/rh/rh-python${PYTHON_VERSION}/root/usr/lib64"
+ENV PKG_CONFIG_PATH="/opt/rh/rh-python${PYTHON_VERSION}/root/usr/lib64/pkgconfig"
+
+# obspy
+RUN python -m pip wheel obspy --wheel-dir /wheels
+# pycurl
+RUN yum install -y libcurl-devel openssl-devel
+RUN export PYCURL_SSL_LIBRARY=nss \
+ && python -m pip wheel pycurl --wheel-dir /wheels
+
+
+################################################################################
+# geomag-algorithms image
+
+FROM python
+
ARG GIT_BRANCH_NAME=none
ARG GIT_COMMIT_SHA=none
ARG WEBSERVICE="false"
@@ -12,24 +61,32 @@ ENV GIT_BRANCH_NAME=${GIT_BRANCH_NAME} \
GIT_COMMIT_SHA=${GIT_COMMIT_SHA} \
WEBSERVICE=${WEBSERVICE}
+# install obspy and pycurl using wheels
+COPY --from=obspy /wheels /wheels
+RUN python -m pip install --find-links file:///wheels obspy pycurl
-# install packages into system python, when Pipfile changes
-COPY Pipfile Pipfile.lock /geomag-algorithms/
+# install packages when dependencies change
+COPY pyproject.toml poetry.lock /geomag-algorithms/
RUN cd /geomag-algorithms \
- && pipenv install --dev --pre --system
+ # install into system python
+ && poetry config virtualenvs.create false \
+ # only install dependencies, not project
+ && poetry install --no-root
# install rest of library as editable
COPY . /geomag-algorithms
RUN cd /geomag-algorithms \
- && pip install -e . \
+ # now install project to install scripts
+ && poetry install \
# add data directory owned by usgs-user
&& mkdir -p /data \
&& chown -R usgs-user:usgs-user /data
+# configure python path, so project can be volume mounted
+ENV PYTHONPATH="/geomag-algorithms"
+# run as usgs-user
USER usgs-user
WORKDIR /data
-
-
# entrypoint needs double quotes
ENTRYPOINT [ "/geomag-algorithms/docker-entrypoint.sh" ]
EXPOSE 8000
diff --git a/bin/geomag_webservice.py b/bin/geomag_webservice.py
deleted file mode 100755
index 1f027d3e2f1bf1091c34e61382173c5cd2c0f38e..0000000000000000000000000000000000000000
--- a/bin/geomag_webservice.py
+++ /dev/null
@@ -1,37 +0,0 @@
-#! /usr/bin/env python
-
-from __future__ import absolute_import, print_function
-
-import os
-import sys
-from wsgiref.simple_server import make_server
-
-# ensure geomag is on the path before importing
-try:
- import geomagio # noqa (tells linter to ignore this line.)
-except ImportError:
- path = os.path
- script_dir = path.dirname(path.abspath(__file__))
- sys.path.append(path.normpath(path.join(script_dir, "..")))
- import geomagio
-
-
-if __name__ == "__main__":
- # read configuration from environment
- edge_host = os.getenv("EDGE_HOST", "cwbpub.cr.usgs.gov")
- edge_port = int(os.getenv("EDGE_PORT", "2060"))
- factory_type = os.getenv("GEOMAG_FACTORY_TYPE", "edge")
- webservice_host = os.getenv("GEOMAG_WEBSERVICE_HOST", "")
- webservice_port = int(os.getenv("GEOMAG_WEBSERVICE_PORT", "7981"))
- version = os.getenv("GEOMAG_VERSION", None)
-
- # configure factory
- if factory_type == "edge":
- factory = geomagio.edge.EdgeFactory(host=edge_host, port=edge_port)
- else:
- raise "Unknown factory type '%s'" % factory_type
-
- print("Starting webservice on %s:%d" % (webservice_host, webservice_port))
- app = geomagio.WebService(factory, version)
- httpd = make_server(webservice_host, webservice_port, app)
- httpd.serve_forever()
diff --git a/geomagio/Controller.py b/geomagio/Controller.py
index 4342bd856db72ec76959f369520faa6d1058ea01..432f63f9f4d54d3451b7495d0c88e7b5625bde6c 100644
--- a/geomagio/Controller.py
+++ b/geomagio/Controller.py
@@ -639,7 +639,7 @@ def get_realtime_interval(interval_seconds: int) -> Tuple[UTCDateTime, UTCDateTi
return starttime, endtime
-def main(args):
+def main(args: Optional[List[str]] = None):
"""command line factory for geomag algorithms
Inputs
@@ -651,6 +651,9 @@ def main(args):
parses command line options using argparse, then calls the controller
with instantiated I/O factories, and algorithm(s)
"""
+ # parse command line arguments by default
+ if args is None:
+ args = parse_args(sys.argv[1:])
# only try to parse deprecated arguments if they've been enabled
if args.enable_deprecated_arguments:
diff --git a/geomagio/WebService.py b/geomagio/WebService.py
deleted file mode 100644
index 324029279f7a1300ca71be3f0b596e44228c0043..0000000000000000000000000000000000000000
--- a/geomagio/WebService.py
+++ /dev/null
@@ -1,417 +0,0 @@
-"""WSGI implementation of Intermagnet Web Service
-"""
-
-from __future__ import print_function
-from html import escape
-from urllib.parse import parse_qs
-
-from collections import OrderedDict
-from datetime import datetime
-from json import dumps
-import sys
-
-from geomagio.edge import EdgeFactory
-from geomagio.iaga2002 import IAGA2002Writer
-from geomagio.imfjson import IMFJSONWriter
-from geomagio.ObservatoryMetadata import ObservatoryMetadata
-from geomagio.WebServiceUsage import WebServiceUsage
-from obspy.core import UTCDateTime
-
-
-DEFAULT_DATA_TYPE = "variation"
-DEFAULT_ELEMENTS = ("X", "Y", "Z", "F")
-DEFAULT_OUTPUT_FORMAT = "iaga2002"
-DEFAULT_SAMPLING_PERIOD = "60"
-ERROR_CODE_MESSAGES = {
- 204: "No Data",
- 400: "Bad Request",
- 404: "Not Found",
- 409: "Conflict",
- 500: "Internal Server Error",
- 501: "Not Implemented",
- 503: "Service Unavailable",
-}
-VALID_DATA_TYPES = ["variation", "adjusted", "quasi-definitive", "definitive"]
-VALID_OUTPUT_FORMATS = ["iaga2002", "json"]
-VALID_SAMPLING_PERIODS = ["1", "60"]
-
-
-def _get_param(params, key, required=False):
- """Get parameter from dictionary.
-
- Parameters
- ----------
- params : dict
- parameters dictionary.
- key : str
- parameter name.
- required : bool
- if required parameter.
-
- Returns
- -------
- value : str
- value from dictionary.
-
- Raises
- ------
- WebServiceException
- if the parameter is specified more than once
- or if required paramenter is not specified.
- """
- value = params.get(key)
- if isinstance(value, (list, tuple)):
- if len(value) > 1:
- raise WebServiceException('"' + key + '" may only be specified once.')
- value = escape(value[0])
- if value is None:
- if required:
- raise WebServiceException('"' + key + '" is a required parameter.')
- return value
-
-
-class WebService(object):
- def __init__(
- self,
- factory=None,
- version=None,
- metadata=None,
- usage_documentation=None,
- error_stream=sys.stderr,
- ):
- self.error_stream = error_stream
- self.factory = factory or EdgeFactory()
- self.metadata = metadata or ObservatoryMetadata().metadata
- self.version = version
- self.usage_documentation = usage_documentation or WebServiceUsage()
-
- def __call__(self, environ, start_response):
- """Implement WSGI interface"""
- if environ["QUERY_STRING"] == "":
- return self.usage_documentation.__call__(environ, start_response)
- try:
- # parse params
- query = self.parse(parse_qs(environ["QUERY_STRING"]))
- query._verify_parameters()
- self.output_format = query.output_format
- except Exception as e:
- message = str(e)
- ftype = parse_qs(environ["QUERY_STRING"]).get("format", [""])[0]
- if ftype == "json":
- self.output_format = "json"
- else:
- self.output_format = "iaga2002"
- error_body = self.error(400, message, environ, start_response)
- return [error_body]
- try:
- # fetch timeseries
- timeseries = self.fetch(query)
- # format timeseries
- timeseries_string = self.format_data(
- query, timeseries, start_response, environ
- )
- if isinstance(timeseries_string, str):
- timeseries_string = timeseries_string.encode("utf8")
- except Exception as e:
- if self.error_stream:
- print("Error processing request: %s" % str(e), file=self.error_stream)
- message = "Server error."
- error_body = self.error(500, message, environ, start_response)
- return [error_body]
- return [timeseries_string]
-
- def error(self, code, message, environ, start_response):
- """Assign error_body value based on error format."""
- error_body = self.http_error(code, message, environ)
- status = str(code) + " " + ERROR_CODE_MESSAGES[code]
- start_response(status, [("Content-Type", "text/plain")])
- if isinstance(error_body, str):
- error_body = error_body.encode("utf8")
- return error_body
-
- def fetch(self, query):
- """Get requested timeseries.
-
- Parameters
- ----------
- query : dict
- parsed query parameters
-
- Returns
- -------
- obspy.core.Stream
- timeseries object with requested data.
- """
- if query.sampling_period == "1":
- sampling_period = "second"
- if query.sampling_period == "60":
- sampling_period = "minute"
- timeseries = self.factory.get_timeseries(
- observatory=query.observatory_id,
- channels=query.elements,
- starttime=query.starttime,
- endtime=query.endtime,
- type=query.data_type,
- interval=sampling_period,
- )
- return timeseries
-
- def format_data(self, query, timeseries, start_response, environ):
- """Format requested timeseries.
-
- Parameters
- ----------
- query : dictionary of parsed query parameters
- timeseries : obspy.core.Stream
- timeseries object with data to be written
-
- Returns
- -------
- unicode
- IAGA2002 formatted string.
- """
- url = environ["HTTP_HOST"] + environ["PATH_INFO"] + environ["QUERY_STRING"]
- if query.output_format == "json":
- timeseries_string = IMFJSONWriter.format(timeseries, query.elements, url)
- else:
- timeseries_string = IAGA2002Writer.format(timeseries, query.elements)
- start_response("200 OK", [("Content-Type", "text/plain")])
- return timeseries_string
-
- def http_error(self, code, message, environ):
- """Format http error message.
-
- Returns
- -------
- http_error_body : str
- body of http error message.
- """
- query_string = environ["QUERY_STRING"]
- path_info = environ["PATH_INFO"]
- host = environ["HTTP_HOST"]
- if self.output_format == "json":
- http_error_body = self.json_error(
- code, message, path_info, query_string, host
- )
- else:
- http_error_body = self.iaga2002_error(
- code, message, path_info, query_string
- )
- return http_error_body
-
- def iaga2002_error(self, code, message, path_info, query_string):
- """Format iaga2002 error message.
-
- Returns
- -------
- error_body : str
- body of iaga2002 error message.
- """
- status_message = ERROR_CODE_MESSAGES[code]
- error_body = (
- "Error "
- + str(code)
- + ": "
- + status_message
- + "\n\n"
- + message
- + "\n\n"
- + "Usage details are available from "
- + "http://geomag.usgs.gov/ws/edge/ \n\n"
- + "Request:\n"
- + path_info
- + "?"
- + query_string
- + "\n\n"
- + "Request Submitted:\n"
- + datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
- + "\n\n"
- )
- # Check if there is version information available
- if self.version is not None:
- error_body += "Service version:\n" + str(self.version)
- return error_body
-
- def json_error(self, code, message, path_info, query_string, host):
- """Format json error message.
-
- Returns
- -------
- error_body : str
- body of json error message.
- """
- error_dict = OrderedDict()
- error_dict["type"] = "Error"
- error_dict["metadata"] = OrderedDict()
- error_dict["metadata"]["status"] = 400
- date = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ")
- error_dict["metadata"]["generated"] = date
- error_dict["metadata"]["url"] = host + path_info + "?" + query_string
- status_message = ERROR_CODE_MESSAGES[code]
- error_dict["metadata"]["title"] = status_message
- error_dict["metadata"]["api"] = str(self.version)
- error_dict["metadata"]["error"] = message
- error_body = dumps(error_dict, ensure_ascii=True).encode("utf8")
- error_body = str(error_body)
- return error_body
-
- def parse(self, params):
- """Parse query string parameters and set defaults.
-
- Parameters
- ----------
- params : dictionary
- parameters dictionary.
-
- Returns
- -------
- WebServiceQuery
- parsed query object.
-
- Raises
- ------
- WebServiceException
- if any parameters are not supported.
- """
- # Get values
- observatory_id = _get_param(params, "id", required=True)
- starttime = _get_param(params, "starttime")
- endtime = _get_param(params, "endtime")
- elements = _get_param(params, "elements")
- sampling_period = _get_param(params, "sampling_period")
- data_type = _get_param(params, "type")
- output_format = _get_param(params, "format")
- # Assign values or defaults
- if not output_format:
- output_format = DEFAULT_OUTPUT_FORMAT
- else:
- output_format = output_format.lower()
- observatory_id = observatory_id.upper()
- if observatory_id not in list(self.metadata.keys()):
- raise WebServiceException(
- 'Bad id value "%s".'
- " Valid values are: %s" % (observatory_id, list(self.metadata.keys()))
- )
- if not starttime:
- now = datetime.now()
- today = UTCDateTime(year=now.year, month=now.month, day=now.day, hour=0)
- starttime = today
- else:
- try:
- starttime = UTCDateTime(starttime)
- except Exception:
- raise WebServiceException(
- 'Bad starttime value "%s".'
- " Valid values are ISO-8601 timestamps." % starttime
- )
- if not endtime:
- endtime = starttime + (24 * 60 * 60 - 1)
- else:
- try:
- endtime = UTCDateTime(endtime)
- except Exception:
- raise WebServiceException(
- 'Bad endtime value "%s".'
- " Valid values are ISO-8601 timestamps." % endtime
- )
- if not elements:
- elements = DEFAULT_ELEMENTS
- else:
- elements = [e.strip().upper() for e in elements.replace(",", "")]
- if not sampling_period:
- sampling_period = DEFAULT_SAMPLING_PERIOD
- else:
- sampling_period = sampling_period
- if not data_type:
- data_type = DEFAULT_DATA_TYPE
- else:
- data_type = data_type.lower()
- # Create WebServiceQuery object and set properties
- query = WebServiceQuery()
- query.observatory_id = observatory_id
- query.starttime = starttime
- query.endtime = endtime
- query.elements = elements
- query.sampling_period = sampling_period
- query.data_type = data_type
- query.output_format = output_format
- return query
-
-
-class WebServiceQuery(object):
- """Query parameters for a web service request.
-
- Parameters
- ----------
- observatory_id : str
- observatory
- starttime : obspy.core.UTCDateTime
- time of first requested sample
- endtime : obspy.core.UTCDateTime
- time of last requested sample
- elements : array_like
- list of requested elements
- sampling_period : int
- period between samples in seconds
- default 60.
- data_type : {'variation', 'adjusted', 'quasi-definitive', 'definitive'}
- data type
- default 'variation'.
- output_format : {'iaga2002', 'json'}
- output format.
- default 'iaga2002'.
- """
-
- def __init__(
- self,
- observatory_id=None,
- starttime=None,
- endtime=None,
- elements=None,
- sampling_period=60,
- data_type="variation",
- output_format="iaga2002",
- ):
- self.observatory_id = observatory_id
- self.starttime = starttime
- self.endtime = endtime
- self.elements = elements
- self.sampling_period = sampling_period
- self.data_type = data_type
- self.output_format = output_format
-
- def _verify_parameters(self):
- """Verify that parameters are valid.
-
- Raises
- ------
- WebServiceException
- if any parameters are not supported.
- """
- if len(self.elements) > 4 and self.output_format == "iaga2002":
- raise WebServiceException(
- "No more than four elements allowed for iaga2002 format."
- )
- if self.starttime > self.endtime:
- raise WebServiceException("Starttime must be before endtime.")
- if self.data_type not in VALID_DATA_TYPES:
- raise WebServiceException(
- 'Bad type value "%s".'
- " Valid values are: %s" % (self.data_type, VALID_DATA_TYPES)
- )
- if self.sampling_period not in VALID_SAMPLING_PERIODS:
- raise WebServiceException(
- 'Bad sampling_period value "%s".'
- " Valid values are: %s" % (self.sampling_period, VALID_SAMPLING_PERIODS)
- )
- if self.output_format not in VALID_OUTPUT_FORMATS:
- raise WebServiceException(
- 'Bad format value "%s".'
- " Valid values are: %s" % (self.output_format, VALID_OUTPUT_FORMATS)
- )
-
-
-class WebServiceException(Exception):
- """Base class for exceptions thrown by web services."""
-
- pass
diff --git a/geomagio/WebServiceUsage.py b/geomagio/WebServiceUsage.py
deleted file mode 100644
index 1530c9248f3191b04f398f916d6c75fc874036fb..0000000000000000000000000000000000000000
--- a/geomagio/WebServiceUsage.py
+++ /dev/null
@@ -1,191 +0,0 @@
-"""Factory that loads html for Web Service Usage Documentation"""
-from datetime import datetime
-from geomagio.ObservatoryMetadata import ObservatoryMetadata
-
-
-class WebServiceUsage(object):
- def __init__(self, metadata=None, mount_path=None, host_prefix=None):
- metadata = metadata or list(ObservatoryMetadata().metadata.keys())
- self.date = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
- self.metadata = ", ".join(sorted(metadata))
- self.mount_path = mount_path
- self.host_prefix = host_prefix
-
- def __call__(self, environ, start_response):
- """Implement documentation page"""
- start_response("200 OK", [("Content-Type", "text/html")])
- if self.mount_path is None:
- self.mount_path = "/ws/edge"
- if self.host_prefix is None:
- self.host_prefix = environ["HTTP_HOST"]
- usage_page = self.set_usage_page()
- return [usage_page]
-
- def set_usage_page(self):
- """Set body of Web Service Usage Documentation Page"""
- stylesheet = "https://geomag.usgs.gov/theme/site/geomag/index.css"
- ids = ""
- observatories = self.metadata.split(", ")
- for idx, obs_id in enumerate(observatories):
- ids += "<code>" + obs_id + "</code>"
- if idx != len(observatories) - 1:
- ids += ", "
- if idx % 9 == 0 and idx != 0:
- ids += "<br/>"
- usage_body = """
- <!doctype html>
- <html>
- <head>
- <title>Geomag Web Service Usage</title>
- <base href={host_prefix}>
- <meta charset="utf-8"/>
- <meta name="viewport" content="width=device-width,
- initial-scale=1"/>
- <link rel="stylesheet" href={stylesheet} type="text/css">
- <style>
- code,
- pre {{
- background: #f8f8f8;
- border-radius: 3px;
- color: #555;
- font-family: monospace;
- }}
- </style>
- </head>
-
- <body style="font-size:135%">
- <main role="main" class="page" aria-labelledby="page-header">
- <header class="page-header" id="page-header">
- <h1>Geomag Web Service Usage</h1>
- </header>
-
-
- <h2>Example Requests</h3>
- <dl>
- <dt>BOU observatory data for current UTC day in IAGA2002
- format</dt>
- <dd>
- <a href="{link1}">
- {link1}</a>
- </dd>
- <dt>BOU observatory data for current UTC day in JSON
- format</dt>
- <dd>
- <a href="{link2}">
- {link2}</a>
- </dd>
- <dt>BOU electric field data for current UTC day in
- IAGA2002 format</dt>
- <dd>
- <a href="{link3}">
- {link3}</a>
- </dd>
-
- <h2>Parameters</h2>
- <dl>
- <dt>id</dt>
- <dd>
- Observatory code.
- Required.<br/>
- Valid values:<br/>
- {metadata}
- </dd>
-
- <dt>starttime</dt>
- <dd>
- Time of first requested data.<br/>
- Default: start of current UTC day<br/>
- Format: ISO8601
- (<code>YYYY-MM-DDTHH:MM:SSZ</code>)<br/>
- Example: <code>{date}</code>
- </dd>
-
- <dt>endtime</dt>
- <dd>
- Time of last requested data.<br/>
- Default: starttime + 24 hours<br/>
- Format: ISO8601
- (<code>YYYY-MM-DDTHH:MM:SSZ</code>)<br/>
- Example: <code>{date}</code>
- </dd>
-
- <dt>elements</dt>
- <dd>
- Comma separated list of requested elements.<br/>
- Default: <code>X</code>,<code>Y</code>,<code>Z</code>,
- <code>F</code><br/>
- Valid values: <code>D</code>, <code>DIST</code>,
- <code>DST</code>, <code>E</code>,
- <code>E-E</code>, <code>E-N</code>,
- <code>F</code>, <code>G</code>,
- <code>H</code>, <code>SQ</code>,
- <code>SV</code>, <code>UK1</code>,
- <code>UK2</code>, <code>UK3</code>,
- <code>UK4</code>, <code>X</code>,
- <code>Y</code>, <code>Z</code>
- <br/>
- </dd>
- <dt>sampling_period</dt>
- <dd>
- Interval in seconds between values.<br/>
- Default: <code>60</code><br/>
- Valid values:
- <code>1</code>,
- <code>60</code>
- </dd>
-
- <dt>type</dt>
- <dd>
- Type of data.<br/>
- Default: <code>variation</code><br/>
- Valid values:
- <code>variation</code>,
- <code>adjusted</code>,
- <code>quasi-definitive</code>,
- <code>definitive</code><br/>
- <small>
- NOTE: the USGS web service also supports specific
- EDGE location codes.
- For example:
- <code>R0</code> is "internet variation",
- <code>R1</code> is "satellite variation".
- </small>
- </dd>
-
- <dt>format</dt>
- <dd>
- Output format.<br/>
- Default: <code>iaga2002</code><br/>
- Valid values:
- <code>iaga2002</code>.
- </dd>
- </dl>
- </main>
-
-
- <nav class="site-footer">
- <p> Not what you were looking for?<br/>
- Search usa.gov: </p>
- <form class="site-search" role="search"
- action="//search.usa.gov/search" method="get"
- accept-charset="UTF-8">
- <input name="utf8" type="hidden" value="x"/>
- <input name="affiliate" type="hidden" value="usgs"/>
- <input name="sitelimit" type="hidden" />
- <input id="query" name="query" type="search"
- placeholder="Search usa.gov..." title="Search"/>
- <button type="submit">Search</button>
- </form>
- </nav>
- </body>
- </html>
- """.format(
- metadata=ids,
- date=self.date,
- host_prefix=self.host_prefix,
- stylesheet=stylesheet,
- link1=self.host_prefix + self.mount_path + "/?id=BOU",
- link2=self.host_prefix + self.mount_path + "/?id=BOU&format=json",
- link3=self.host_prefix + self.mount_path + "/?id=BOU&elements=E-N,E-E",
- )
- return usage_body
diff --git a/geomagio/__init__.py b/geomagio/__init__.py
index 5c391414edd5b3cb148be27ed8c6ea6cc56bc6e8..05fcd276c5d8542e00e34daaa4e6f71b9a3567e1 100644
--- a/geomagio/__init__.py
+++ b/geomagio/__init__.py
@@ -13,7 +13,6 @@ from .ObservatoryMetadata import ObservatoryMetadata
from .PlotTimeseriesFactory import PlotTimeseriesFactory
from .TimeseriesFactory import TimeseriesFactory
from .TimeseriesFactoryException import TimeseriesFactoryException
-from .WebService import WebService
__all__ = [
"ChannelConverter",
@@ -26,5 +25,4 @@ __all__ = [
"TimeseriesFactoryException",
"TimeseriesUtility",
"Util",
- "WebService",
]
diff --git a/setup.py b/setup.py
deleted file mode 100644
index 0d7801818eb84eaf21ae6c77c8b308409cf748ed..0000000000000000000000000000000000000000
--- a/setup.py
+++ /dev/null
@@ -1,34 +0,0 @@
-import os
-import setuptools
-import setuptools.ssl_support
-
-# configure ssl certifiate bundle from environment, if set
-ssl_cert_file = os.environ.get("SSL_CERT_FILE") or os.environ.get("PIP_CERT")
-if ssl_cert_file:
- setuptools.ssl_support.cert_paths = [ssl_cert_file]
-
-setuptools.setup(
- name="geomag-algorithms",
- version="1.3.5",
- description="USGS Geomag Algorithms Library",
- url="https://github.com/usgs/geomag-algorithms",
- packages=setuptools.find_packages(exclude=["test*"]),
- project_urls={
- "Bug Reports": "https://github.com/usgs/geomag-algorithms/issues",
- "Source": "https://github.com/usgs/geomag-algorithms",
- },
- python_requires=">=3.6, <4",
- scripts=["bin/geomag.py", "bin/geomag_webservice.py", "bin/make_cal.py"],
- setup_requires=[
- "setuptools-pipfile",
- ],
- use_pipfile=True,
- entry_points={
- "console_scripts": [
- "generate-matrix=geomagio.processing.affine_matrix:main",
- "geomag-metadata=geomagio.metadata.main:main",
- "magproc-prepfiles=geomagio.processing.magproc:main",
- "obsrio-filter=geomagio.processing.obsrio:main",
- ],
- },
-)
diff --git a/test/WebService_test.py b/test/WebService_test.py
deleted file mode 100644
index 99c5e0ad50c517a27d92ffbad46de19ae5ec0092..0000000000000000000000000000000000000000
--- a/test/WebService_test.py
+++ /dev/null
@@ -1,165 +0,0 @@
-"""Unit Tests for WebService"""
-from urllib.parse import parse_qs
-from datetime import datetime
-from numpy.testing import assert_equal, assert_raises
-import numpy
-import webtest
-
-from geomagio.WebService import _get_param
-from geomagio.WebService import WebService
-import obspy.core
-from obspy.core.stream import Stream
-from obspy.core.utcdatetime import UTCDateTime
-
-
-class TestFactory(object):
- "Factory to test for 200 and 400 response statuses."
-
- @staticmethod
- def get_timeseries(
- observatory=None,
- channels=None,
- starttime=None,
- endtime=None,
- type=None,
- interval=None,
- ):
- stream = obspy.core.Stream()
- for channel in channels:
- stats = obspy.core.Stats()
- stats.channel = channel
- stats.starttime = starttime
- stats.network = "Test"
- stats.station = observatory
- stats.location = observatory
- if interval == "second":
- stats.sampling_rate = 1.0
- elif interval == "minute":
- stats.sampling_rate = 1.0 / 60.0
- elif interval == "hourly":
- stats.sampling_rate = 1.0 / 3600.0
- elif interval == "daily":
- stats.sampling_rate = 1.0 / 86400.0
- length = int((endtime - starttime) * stats.sampling_rate)
- stats.npts = length + 1
- data = numpy.full(length, numpy.nan, dtype=numpy.float64)
- trace = obspy.core.Trace(data, stats)
- stream.append(trace)
- return stream
-
-
-class ErrorFactory(object):
- "Factory to test for 500 response status."
-
- @staticmethod
- def get_timeseries(
- observatory=None,
- channels=None,
- starttime=None,
- endtime=None,
- type=None,
- interval=None,
- ):
- pass
-
-
-def test__get_param():
- """WebService_test.test__get_param()
-
- Call function _get_param to make certain it gets back
- the appropriate values and raises exceptions for invalid values.
- """
- params = {
- "id": None,
- "elements": "H,E,Z,F",
- "sampling_period": ["1", "60"],
- }
- assert_raises(Exception, _get_param, params, "id", required=True)
- elements = _get_param(params, "elements")
- assert_equal(elements, "H,E,Z,F")
- assert_raises(Exception, _get_param, params, "sampling_period")
-
-
-def test_fetch():
- """WebService_test.test_fetch())
-
- Call function WebService.fetch to confirm tht it returns an
- obspy.core.stream object.
- """
- service = WebService(TestFactory())
- query = service.parse(
- parse_qs(
- "id=BOU&starttime=2016-06-06"
- "&endtime=2016-06-07&elements=H,E,Z,F&sampling_period=60"
- "&format=iaga2002&type=variation"
- )
- )
- timeseries = service.fetch(query)
- assert_equal(isinstance(timeseries, Stream), True)
-
-
-def test_parse():
- """WebService_test.test_parse()
-
- Create WebService instance and call parse to confirm that query
- string values are applied to the correct class attribute. Also
- confirm that default values are applied correctly.
- """
- service = WebService(TestFactory())
- query = service.parse(
- parse_qs(
- "id=BOU&starttime=2016-06-06"
- "&endtime=2016-06-07&elements=H,E,Z,F&sampling_period=60"
- "&format=iaga2002&type=variation"
- )
- )
- assert_equal(query.observatory_id, "BOU")
- assert_equal(query.starttime, UTCDateTime(2016, 6, 6, 0))
- assert_equal(query.endtime, UTCDateTime(2016, 6, 7, 0))
- assert_equal(query.elements, ["H", "E", "Z", "F"])
- assert_equal(query.sampling_period, "60")
- assert_equal(query.output_format, "iaga2002")
- assert_equal(query.data_type, "variation")
- # Test that defaults are set for unspecified values
- now = datetime.now()
- today = UTCDateTime(year=now.year, month=now.month, day=now.day, hour=0)
- tomorrow = today + (24 * 60 * 60 - 1)
- query = service.parse(parse_qs("id=BOU"))
- assert_equal(query.observatory_id, "BOU")
- assert_equal(query.starttime, today)
- assert_equal(query.endtime, tomorrow)
- assert_equal(query.elements, ("X", "Y", "Z", "F"))
- assert_equal(query.sampling_period, "60")
- assert_equal(query.output_format, "iaga2002")
- assert_equal(query.data_type, "variation")
- assert_raises(Exception, service.parse, parse_qs("/?id=bad"))
-
-
-def test_requests():
- """WebService_test.test_requests()
-
- Use TestApp to confirm correct response status, status int,
- and content-type.
- """
- app = webtest.TestApp(WebService(TestFactory()))
- # Check invalid request (bad values)
- response = app.get("/?id=bad", expect_errors=True)
- assert_equal(response.status_int, 400)
- assert_equal(response.status, "400 Bad Request")
- assert_equal(response.content_type, "text/plain")
- # Check invalid request (duplicates)
- response = app.get("/?id=BOU&id=BOU", expect_errors=True)
- assert_equal(response.status_int, 400)
- assert_equal(response.status, "400 Bad Request")
- assert_equal(response.content_type, "text/plain")
- # Check valid request (upper and lower case)
- response = app.get("/?id=BOU")
- assert_equal(response.status_int, 200)
- assert_equal(response.status, "200 OK")
- assert_equal(response.content_type, "text/plain")
- # Test internal server error (use fake factory)
- app = webtest.TestApp(WebService(ErrorFactory(), error_stream=None))
- response = app.get("/?id=BOU", expect_errors=True)
- assert_equal(response.status_int, 500)
- assert_equal(response.status, "500 Internal Server Error")
- assert_equal(response.content_type, "text/plain")