"""GFINDER CLI `search` command."""
import click
from .docstring import set_docstring
from gfinder.config import Config, DEFAULT_N_STEPS, DEFAULT_TIME_STEP
from gfinder.datastore import DataStore
from gfinder.opportunity import Opportunity, TimeInputs, OpportunityDefinition
@click.command(name='search', context_settings=dict(
ignore_unknown_options=True,
allow_extra_args=True,
))
@click.argument('odf_file')
@click.option('--opportunity-id', type=click.STRING, help='Identifier of an opportunity to used as input time window input (cascading search).')
@click.option('--start-time', type=click.STRING, help='Start time of the input time window (UTC), or relative start time string if --event-id is used.')
@click.option('--stop-time', type=click.STRING, help='Stop time of the input time window (UTC), or relative stop time string if --event-id is used.')
@click.option('--event-id', type=click.STRING, help='Identifier of the mission event (phase/segment) to be used as input time window (based on Mission Event file).')
@click.option('--n-steps', type=click.INT, help=f'Number of time steps used to derive time step (default: {DEFAULT_N_STEPS}).', default=DEFAULT_N_STEPS)
@click.option('--time-step', type=click.FLOAT, help=f'Time step, in seconds, to be used to bracket roots (default: {DEFAULT_TIME_STEP} sec.).', default=DEFAULT_TIME_STEP)
@click.option('--mission-scenario-id', type=click.STRING, help=f'Mission scenario ID (default: {Config().default_mission_scenario_id})', default=Config().default_mission_scenario_id)
@click.option('--target', type=click.STRING, help='Target name ODF overwrite.')
@click.option('--compute/--no-compute', help='Run observation geometry data computation for all opportunity windows.', default='False')
@click.option('--overwrite/--no-overwrite', help='Overwrite existing opportunity data files.', default='False')
@click.option('--suffix', type=click.STRING, help='Add suffix string to the searched opportunity ID.')
@click.pass_context
def search_cmd(ctx, odf_file, opportunity_id, start_time, stop_time, event_id, n_steps, time_step, mission_scenario_id, target, compute, overwrite, suffix):
"""Search observation opportunities.
This command takes as input an Opportunity Definition File (ODF) file, containing the geometry condition(s) and
geometry parameters corresponding to a user-defined observation opportunity.
Search time window can be provided as timing inputs (using start_time, stop_time, event_name options), or as an
'opportunity' resulting from a previous search.
It produces a set of output geometry event (GeoEvent) files in JSON format bundled in an 'opportunity' directory.
It also produces companion GeoJSON files when applicable. The identifier of an opportunity is automatically
assigned.
"""
# Retrieve extra arguments to be passed to the command function.
kwargs = []
for arg in ctx.args:
pair = arg[2:].split('=')
key = pair[0].replace('-', '_')
try:
val = float(pair[1])
except:
val = pair[1]
kwargs.append((key,val))
kwargs = dict(kwargs)
search(odf_file, opportunity_id=opportunity_id, start_time=start_time, stop_time=stop_time, event_id=event_id,
n_steps=n_steps, time_step=time_step, mission_scenario_id=mission_scenario_id, target=target, compute=compute,
overwrite=overwrite, suffix=suffix, **kwargs)
[docs]@set_docstring(search_cmd, {'odf_file': 'ODF file output_dir relative to ODF data store directory.'})
def search(odf_file, opportunity_id=None, start_time=None, stop_time=None, event_id=None, n_steps=DEFAULT_N_STEPS,
time_step=DEFAULT_TIME_STEP, mission_scenario_id=Config().default_mission_scenario_id, target=None, compute=False,
overwrite=False, suffix=None, silent=False, no_save=False, **kwargs):
# Init data store
datastore = DataStore()
# Load SPICE kernels for the selected mission scenario
if not mission_scenario_id:
mission_scenario_id = Config().default_mission_scenario_id
mission_scenario = datastore.getMissionScenario(mission_scenario_id)
if not mission_scenario:
raise AttributeError(f'`{mission_scenario_id}` mission scenario does not exist, or is not correctly defined in scenario/mission-scenarios.json file.')
mission_scenario.loadKernels()
# Load input Opportunity Definition File (ODF) (geometry event definition)
odf_file_abspath = datastore.get_ODF_filename(odf_file)
if odf_file_abspath:
opportunity_definition = OpportunityDefinition(odf_file_abspath, target=target, datastore=datastore, **kwargs)
else:
raise FileNotFoundError(f'Input `{odf_file}` ODF file was not found (neither in your user data nor the app data directories).')
if not opportunity_definition.valid:
raise AttributeError(f'Invalid input `{odf_file}` ODF file.')
# Set time inputs as object/time_inputs_dict
time_inputs_dict = {
'opportunity_id': opportunity_id,
'event_id': event_id,
'start_time': start_time,
'stop_time': stop_time,
'n_steps': n_steps,
'time_step': time_step
}
in_opportunity = None
if opportunity_id:
print('Loading and retrieving <{}> opportunity sequence time intervals to be used as search input window...'.format(opportunity_id))
in_opportunity = datastore.loadOpportunity(opportunity_id)
time_inputs_dict['event_id'] = in_opportunity.time_inputs.event_id
print('Done.')
time_inputs = TimeInputs(time_inputs_dict, opportunity=in_opportunity, mission_scenario=mission_scenario)
if not time_inputs.valid:
print('[ERROR] Invalid time inputs.')
return
opportunity_id = ''
# Instantiate opportunity (calculation)
opportunity = Opportunity(mission_scenario, opportunity_definition, time_inputs)
# Check if such an opportunity has been computed before (available in the data store), expect if --overwrite option is used.
if not overwrite and not no_save: # default (--no-overwrite)
exist, opportunity_id = datastore.opportunityExists(opportunity, suffix=suffix)
if exist:
print()
print('Opportunity exists already: ' + opportunity_id)
print('Use `gfinder opportunity {}` command to get more information.'.format(opportunity_id))
return
# Search opportunities of observations for the input search window
opportunity.search()
print(f'{opportunity.sequence.n_sub_events} opportunity window(s) found.')
# Compute opportunity geometry data
if opportunity.sequence.n_sub_events > 0:
if compute:
print('Computing all geometry data for each found opportunity window...')
opportunity.compute()
else:
print('Computing searchable-only geometry data for each found opportunity window...')
opportunity.compute(searchable_only=True)
# Report on search results
if not silent:
print()
opportunity.summary()
if no_save is False:
# Write opportunity data files (opportunity.json and GeoEvent JSON files).
opportunity_id, opportunity_path = datastore.writeOpportunity(opportunity, overwrite=overwrite, suffix=suffix)
print('Opportunity ID : '+ opportunity_id)
print('All results in {} directory.'.format(opportunity_path))
return opportunity
