Add obs data cloning job (#324)

* add job to copy obs data

* include job file

* preserve user flagged, add note about cron considerations

* testing

* extra coverage

* coverage

* last bit of diff coverage

* update comments and correct test

* consistent arg ordering, update comment

Author: lboeman

sfa_api/admincli.py

@@ -457,3 +457,29 @@ def reference_persistence_job(
         'reference_persistence', name, user_id, cron_string, timeout,
         base_url=base_url)
     click.echo(f'Job created with id {id_}')
+
+
+@create_jobs.command('trial-data-copy')
+@with_default_options
+@base_url
+@timeout
+@name_arg
+@user_id_arg
+@cron_string
+@click.argument('copy_from', type=click.UUID)
+@click.argument('copy_to', type=click.UUID)
+def trial_data_copy_job(
+        name, user_id, cron_string, timeout, copy_from, copy_to, base_url,
+        **kwargs):
+    """
+    Create a job to copy latest data from one observation to another.
+    For short intervals, cron_string should schedule execution every
+    x minutes where x is the observation interval length. For long
+    intervals, cron string should should be set to execute shortly
+    after data is posted to avoid introducing high latency.
+    """
+    from sfa_api.jobs import create_job
+    id_ = create_job(
+        'trial_data_copy', name, user_id, cron_string, timeout,
+        copy_to=str(copy_to), copy_from=str(copy_from), base_url=base_url)
+    click.echo(f'Job created with id {id_}')

sfa_api/jobs.py

@@ -19,6 +19,7 @@
 from sfa_api.utils.auth0_info import exchange_refresh_token
 from sfa_api.utils.queuing import get_queue
 import sfa_api.utils.storage_interface as storage
+from sfa_api.utils.trial_jobs import copy_observation_data
 
 
 logger = logging.getLogger(__name__)
@@ -101,6 +102,8 @@ def create_job(job_type, name, user_id, cron_string, timeout=None, **kwargs):
     elif job_type in ('reference_persistence',
                       'reference_probabilistic_persistence'):
         keys = ()
+    elif job_type in ('trial_data_copy'):
+        keys = ('copy_from', 'copy_to')
     else:
         raise ValueError(f'Job type {job_type} is not supported')
     params = {}
@@ -166,6 +169,9 @@ def execute_job(name, job_type, user_id, **kwargs):
         max_run_time = utcnow()
         return make_latest_probabilistic_persistence_forecasts(
             token, max_run_time, base_url=base_url)
+    elif job_type == 'trial_data_copy':
+        return copy_observation_data(
+            token, kwargs['copy_from'], kwargs['copy_to'], base_url=base_url)
     else:
         raise ValueError(f'Job type {job_type} is not supported')
 

sfa_api/tests/test_admincli.py

@@ -490,3 +490,14 @@ def test_reference_persistence_job(app_cli_runner, mocker, user_id):
         + auth_args)
     assert result.exit_code == 0
     assert result.output == 'Job created with id jobid\n'
+
+
+def test_trial_data_job(app_cli_runner, mocker, user_id, observation_id):
+    mocker.patch('sfa_api.jobs.create_job', return_value='jobid',
+                 autospec=True)
+    result = app_cli_runner.invoke(
+        admincli.trial_data_copy_job,
+        ['Val Job', user_id, '* * * * *', '--base-url', 'http://test',
+         observation_id, observation_id] + auth_args)
+    assert result.exit_code == 0
+    assert result.output == 'Job created with id jobid\n'

sfa_api/tests/test_jobs.py

@@ -180,6 +180,11 @@ def test_convert_sql_job_to_rq_job_not_cron(sql_job, mocker):
     ('reference_probabilistic_persistence',
      {},
      'sfa_api.jobs.make_latest_probabilistic_persistence_forecasts'),
+    ('trial_data_copy',
+     {'base_url': 'https://',
+      'copy_from': 'id1',
+      'copy_to': 'id2'},
+     'sfa_api.jobs.copy_observation_data')
 ])
 def test_execute_job(jtype, params, func, mocker, userid):
     mocker.patch('sfa_api.jobs.exchange_token',
@@ -261,6 +266,11 @@ def my_err(job, *exc_info):
     ('periodic_report', {'report_id': 'id'}),
     ('reference_persistence', {'base_url': 'https://'}),
     ('reference_probabilistic_persistence', {'base_url': 'https://'}),
+    ('trial_data_copy', {
+        'base_url': 'https://',
+        'copy_from': 'id1',
+        'copy_to': 'id2'
+    }),
     pytest.param('badtype', {}, marks=pytest.mark.xfail(
         strict=True, raises=ValueError)),
     pytest.param('daily_observation_validation', {}, marks=pytest.mark.xfail(

sfa_api/utils/tests/test_trial_jobs.py

@@ -0,0 +1,139 @@
+import pytest
+import pandas as pd
+from requests.exceptions import HTTPError
+
+
+from sfa_api.utils import trial_jobs
+
+
+source_index = pd.date_range(
+    '2019-04-14T07:00Z',
+    '2019-04-14T07:09Z',
+    freq='1T'
+)
+
+source_data = pd.DataFrame(
+    [(x, x) for x in range(0, 10)],
+    index=source_index,
+    columns=['value', 'quality_flag']
+)
+
+target_data = pd.DataFrame(
+    [(0, 0)],
+    index=source_index[:0],
+    columns=['value', 'quality_flag']
+)
+
+
+@pytest.fixture()
+def mock_session(mocker):
+    mock_sess = mocker.MagicMock()
+    mock_sess.get_observation_time_range = mocker.MagicMock(
+        return_value=(
+            pd.Timestamp('2019-04-14T07:00Z'),
+            pd.Timestamp('2019-04-14T07:00Z')
+        )
+    )
+    mock_sess.get_observation_values = mocker.MagicMock(
+        return_value=source_data[1:]
+    )
+    mock_sess.post_observation_values = mocker.MagicMock(return_value=None)
+
+    mocker.patch('sfa_api.utils.trial_jobs.APISession', return_value=mock_sess)
+    return mock_sess
+
+
+@pytest.fixture()
+def mock_logging(mocker):
+    mock_logger = mocker.MagicMock()
+    mocker.patch(
+        'sfa_api.utils.trial_jobs.logging.getLogger',
+        return_value=mock_logger
+    )
+    return mock_logger
+
+
+def test_copy_data(mocker, mock_session, observation_id, mock_logging):
+    trial_jobs.copy_observation_data("token", observation_id, observation_id)
+    mock_logging.info.assert_called_with(
+        "Copied %s points from obs %s to %s.", 9,
+        observation_id, observation_id
+    )
+    post_args = mock_session.post_observation_values.call_args_list
+    post_args_df = post_args[0][0][1]
+    qfs = post_args_df['quality_flag'].values
+    assert (qfs == [1, 0, 1, 0, 1, 0, 1, 0, 1]).all()
+
+
+def test_copy_data_no_copy_from_read(
+        mocker, mock_session, observation_id):
+    resp = mocker.MagicMock()
+    resp.status_code = 404
+    mock_session.get_observation_values = mocker.MagicMock(
+        side_effect=HTTPError(response=resp)
+    )
+    with pytest.raises(ValueError) as e:
+        trial_jobs.copy_observation_data(
+            "token", observation_id, observation_id
+        )
+    assert "Read values failure for copy_from " in str(e.value)
+
+
+def test_copy_data_no_copy_to_read(
+        mocker, mock_session, observation_id):
+    resp = mocker.MagicMock()
+    resp.status_code = 404
+    mock_session.get_observation_time_range = mocker.MagicMock(
+        side_effect=HTTPError(response=resp)
+    )
+    with pytest.raises(ValueError) as e:
+        trial_jobs.copy_observation_data(
+            "token", observation_id, observation_id
+        )
+    assert "Read values failure for copy_to " in str(e.value)
+
+
+def test_copy_data_no_copy_to_data(
+        mocker, mock_session, mock_logging, observation_id):
+    resp = mocker.MagicMock()
+    resp.status_code = 404
+    mock_session.get_observation_time_range = mocker.MagicMock(
+        side_effect=[
+            (pd.NaT, pd.NaT),
+            (source_index[0], source_index[-1])
+        ]
+    )
+    trial_jobs.copy_observation_data(
+        "token", observation_id, observation_id
+    )
+    # Mocked observation_values get will return full test data frame,
+    # so check that get_observation_values was called with the expected
+    # timestamp.
+    mock_session.get_observation_values.assert_called_with(
+        observation_id, source_index[-1], mocker.ANY
+    )
+
+
+def test_copy_data_no_copy_to_write(
+        mocker, mock_session, observation_id):
+    resp = mocker.MagicMock()
+    resp.status_code = 404
+    mock_session.post_observation_values = mocker.MagicMock(
+        side_effect=HTTPError(response=resp)
+    )
+    with pytest.raises(ValueError) as e:
+        trial_jobs.copy_observation_data(
+            "token", observation_id, observation_id
+        )
+    assert "Write values failure for copy_to " in str(e.value)
+
+
+def test_copy_data_nothing_to_copy(
+        mocker, mock_session, observation_id, mock_logging):
+    mock_session.get_observation_values = mocker.MagicMock(
+        return_value=pd.DataFrame()
+    )
+    trial_jobs.copy_observation_data("token", observation_id, observation_id)
+    mock_logging.info.assert_called_with(
+        "No points to copy from %s to %s.", observation_id, observation_id
+    )

sfa_api/utils/trial_jobs.py

@@ -0,0 +1,87 @@
+import logging
+from requests.exceptions import HTTPError
+
+import pandas as pd
+from solarforecastarbiter.io.api import APISession
+from solarforecastarbiter.validation.quality_mapping import (
+    BITMASK_DESCRIPTION_DICT
+)
+
+USER_FLAGGED = BITMASK_DESCRIPTION_DICT[1]["USER FLAGGED"]
+
+
+def copy_observation_data(token, copy_from, copy_to, base_url=None):
+    """Helper job to copy observation data from one observation to
+    another.
+
+    Parameters
+    ----------
+    token: str
+        API token to use.
+    copy_from: str
+        UUID of the observation to copy data from. Must have read_values
+        permission on this observation.
+    copy_to: str
+        UUID of the observation to copy data to. Must have write_values
+        permission on this observation.
+    base_url: str
+        URL of the api instance to connect to.
+
+    Raises
+    ------
+    ValueError
+        If reading values from either observation fails, or writing to
+        copy_to fails.
+    """
+    logger = logging.getLogger(__name__)
+    sess = APISession(token, base_url=base_url)
+    # Get latest timestamps for copy_to
+    try:
+        latest_copy_to = sess.get_observation_time_range(copy_to)[1]
+    except HTTPError as e:
+        raise ValueError(
+            "Copy observation failed. Read values failure for copy_to "
+            f"observation with error code: {e.response.status_code}"
+        )
+
+    try:
+        # If no data in copy_to, start from latest copy_from value.
+        # Otherwise adjust latest_copy_to forward one minute to avoid
+        # overwriting data.
+        if pd.isna(latest_copy_to):
+            latest_copy_to = sess.get_observation_time_range(copy_from)[1]
+        else:
+            latest_copy_to += pd.Timedelta('1T')
+
+        # Get values for copy_from from latest copy_to to now.
+        values_to_copy = sess.get_observation_values(
+            copy_from,
+            latest_copy_to,
+            pd.Timestamp.utcnow()
+        )
+    except HTTPError as e:
+        raise ValueError(
+            "Copy observation failed. Read values failure for copy_from "
+            f"observation with error code: {e.response.status_code}"
+        )
+
+    if not values_to_copy.empty:
+        # Need to reset quality_flag and retained user flagged data, so
+        # take the & of current flags and USER_FLAGGED
+        flags = values_to_copy['quality_flag'] & USER_FLAGGED
+        values_to_copy['quality_flag'] = flags
+
+        try:
+            sess.post_observation_values(copy_to, values_to_copy)
+        except HTTPError as e:
+            raise ValueError(
+                "Copy observation failed. Write values failure for copy_to "
+                f"observation with error code: {e.response.status_code}"
+            )
+        else:
+            logger.info(
+                "Copied %s points from obs %s to %s.",
+                len(values_to_copy), copy_from, copy_to
+            )
+    else:
+        logger.info("No points to copy from %s to %s.", copy_from, copy_to)