Skip to content

Commit

Permalink
[pre-commit.ci] auto fixes from pre-commit.com hooks
Browse files Browse the repository at this point in the history
for more information, see https://pre-commit.ci
  • Loading branch information
pre-commit-ci[bot] committed Oct 3, 2024
1 parent a80885a commit e097575
Show file tree
Hide file tree
Showing 3 changed files with 88 additions and 61 deletions.
114 changes: 68 additions & 46 deletions romancal/associations/skycell_asn.py
Original file line number Diff line number Diff line change
@@ -1,22 +1,23 @@
"""Create an association based on skycells"""
import argparse
import sys

import argparse
import json
import logging
import numpy as np
import sys

from romancal.associations import asn_from_list
import romancal.patch_match.patch_match as pm
import numpy as np
import roman_datamodels as rdm

import romancal.patch_match.patch_match as pm
from romancal.associations import asn_from_list

__all__ = ['skycell_asn']
__all__ = ["skycell_asn"]

# Configure logging
logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())
logger.setLevel('INFO')
logger.setLevel("INFO")


def skycell_asn(self):
"""Create the associaton from the list"""
Expand All @@ -33,31 +34,46 @@ def skycell_asn(self):
unique_patches = np.unique(np.concatenate(all_patches))
for item in unique_patches:
member_list = []
patch_name = pm.PATCH_TABLE[item]['name']
patch_name = pm.PATCH_TABLE[item]["name"]
for a in file_list:
if np.isin(item, a[1]):
member_list.append(a[0])

# grab all the wcs parameters needed for generate_tan_wcs
projcell_info = dict([('pixel_scale', float(pm.PATCH_TABLE[item]['pixel_scale'])),
('ra_cent' ,float(pm.PATCH_TABLE[item]['ra_projection_center'])),
('dec_cent' ,float(pm.PATCH_TABLE[item]['dec_projection_center'])),
('shiftx' , float(pm.PATCH_TABLE[item]['x0_projection'])),
('shifty' , float(pm.PATCH_TABLE[item]['y0_projection'])),
('nx', int(pm.PATCH_TABLE[item]['nx'])),
('ny', int(pm.PATCH_TABLE[item]['ny'])),
])
projcell_info = dict(
[
("pixel_scale", float(pm.PATCH_TABLE[item]["pixel_scale"])),
("ra_cent", float(pm.PATCH_TABLE[item]["ra_projection_center"])),
("dec_cent", float(pm.PATCH_TABLE[item]["dec_projection_center"])),
("shiftx", float(pm.PATCH_TABLE[item]["x0_projection"])),
("shifty", float(pm.PATCH_TABLE[item]["y0_projection"])),
("nx", int(pm.PATCH_TABLE[item]["nx"])),
("ny", int(pm.PATCH_TABLE[item]["ny"])),
]
)
program_id = member_list[0][1:6]
root_asn_name = self.parsed.output_file_root
product_type = self.parsed.product_type
product_release = self.parsed.release_product
suffix = "i2d"
sep = "_"
asn_file_name = root_asn_name + sep + patch_name + sep + product_type + \
sep + filter_id + sep + product_release + sep + suffix
with open(asn_file_name+"_asn.json", "w") as outfile:
prompt_product_asn = asn_from_list.asn_from_list(member_list,
product_name = asn_file_name )
asn_file_name = (
root_asn_name
+ sep
+ patch_name
+ sep
+ product_type
+ sep
+ filter_id
+ sep
+ product_release
+ sep
+ suffix
)
with open(asn_file_name + "_asn.json", "w") as outfile:
prompt_product_asn = asn_from_list.asn_from_list(
member_list, product_name=asn_file_name
)
prompt_product_asn["asn_type"] = "image"
prompt_product_asn["program"] = program_id
prompt_product_asn["target"] = patch_name
Expand All @@ -66,7 +82,7 @@ def skycell_asn(self):
outfile.write(serialized)


class Main():
class Main:
"""Command-line interface for list_to_asn
Parameters
Expand All @@ -77,74 +93,80 @@ class Main():
- `[str, ...]`: A list of strings which create the command line
with the similar structure as `sys.argv`
"""

def __init__(self, args=None):
if args is None:
args = sys.argv[1:]
if isinstance(args, str):
args = args.split(" ")

parser = argparse.ArgumentParser(
description='Create an association from a list of files',
usage='skycell_asn --product-type visit --release-product prompt *_cal.asdf -o r512',
description="Create an association from a list of files",
usage="skycell_asn --product-type visit --release-product prompt *_cal.asdf -o r512",
)

parser.add_argument(
'-o', '--output-file-root',
"-o",
"--output-file-root",
type=str,
required=True,
help='Root string for file to write association to'
help="Root string for file to write association to",
)

parser.add_argument(
'-f', '--format',
"-f",
"--format",
type=str,
default='json',
help='Format of the association files. Default: "%(default)s"'
default="json",
help='Format of the association files. Default: "%(default)s"',
)

parser.add_argument(
'--product-type',
"--product-type",
type=str,
default='visit',
help='The product type when creating the association'
default="visit",
help="The product type when creating the association",
)

parser.add_argument(
'--release-product',
"--release-product",
type=str,
default='prompt',
help='The release product when creating the association'
default="prompt",
help="The release product when creating the association",
)

parser.add_argument(
'-r', '--rule',
"-r",
"--rule",
type=str,
default='DMS_ELPP_Base',
default="DMS_ELPP_Base",
help=(
'The rule to base the association structure on.'
"The rule to base the association structure on."
' Default: "%(default)s"'
)
),
)
parser.add_argument(
'-i', '--id',
"-i",
"--id",
type=str,
default='o999',
default="o999",
help='The association candidate id to use. Default: "%(default)s"',
dest='acid'
dest="acid",
)

parser.add_argument(
'filelist',
"filelist",
type=str,
nargs='+',
help='File list to include in the association'
nargs="+",
help="File list to include in the association",
)

self.parsed = parser.parse_args(args=args)
logger.info("Command-line arguments: %s", self.parsed)

skycell_asn(self)

if __name__ == '__main__':

if __name__ == "__main__":

Main()
5 changes: 2 additions & 3 deletions romancal/associations/tests/test_skycell_asn.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,10 @@

import pytest

#from romancal.associations import Association, AssociationRegistry, load_asn
# from romancal.associations import Association, AssociationRegistry, load_asn
from romancal.associations.skycell_asn import Main


def test_cmdline_fails():
"""Exercise the command line interface"""

Expand All @@ -15,5 +16,3 @@ def test_cmdline_fails():
# Only the association file argument
with pytest.raises(SystemExit):
Main(["-o", "test_asn.json"])


30 changes: 18 additions & 12 deletions romancal/regtest/test_skycell_generation.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

from romancal.associations import skycell_asn


def passfail(bool_expr):
"""set pass fail"""
if bool_expr:
Expand All @@ -18,24 +19,29 @@ def test_skycell_asn_generation(rtdata, ignore_asdf_paths):
"""Test for the generation of associations based on skycells"""

# This test should generate seven json files
args = ['r0000101001001001001_01101_0002_WFI01_cal.asdf', \
'r0000101001001001001_01101_0002_WFI10_cal.asdf', '-o', 'r512']
#rtdata.get_asn("WFI/image/L3_regtest_asn.json")
args = [
"r0000101001001001001_01101_0002_WFI01_cal.asdf",
"r0000101001001001001_01101_0002_WFI10_cal.asdf",
"-o",
"r512",
]
# rtdata.get_asn("WFI/image/L3_regtest_asn.json")
rtdata.get_data("WFI/image/r0000101001001001001_01101_0002_WFI01_cal.asdf")
rtdata.get_data("WFI/image/r0000101001001001001_01101_0002_WFI10_cal.asdf")

skycell_asn.Main(args)

output_files = ['r512_r274dp63x31y80_visit_F158_prompt_i2d_asn.json',
'r512_r274dp63x31y81_visit_F158_prompt_i2d_asn.json',
'r512_r274dp63x31y82_visit_F158_prompt_i2d_asn.json',
'r512_r274dp63x32y80_visit_F158_prompt_i2d_asn.json',
'r512_r274dp63x32y81_visit_F158_prompt_i2d_asn.json',
'r512_r274dp63x33y80_visit_F158_prompt_i2d_asn.json',
'r512_r274dp63x33y81_visit_F158_prompt_i2d_asn.json']
output_files = [
"r512_r274dp63x31y80_visit_F158_prompt_i2d_asn.json",
"r512_r274dp63x31y81_visit_F158_prompt_i2d_asn.json",
"r512_r274dp63x31y82_visit_F158_prompt_i2d_asn.json",
"r512_r274dp63x32y80_visit_F158_prompt_i2d_asn.json",
"r512_r274dp63x32y81_visit_F158_prompt_i2d_asn.json",
"r512_r274dp63x33y80_visit_F158_prompt_i2d_asn.json",
"r512_r274dp63x33y81_visit_F158_prompt_i2d_asn.json",
]
# Test that the json files exist
for file in output_files:
skycell_asn.logger.info(
"Check that the json file exists "
+ passfail(os.path.isfile(file))
"Check that the json file exists " + passfail(os.path.isfile(file))
)

0 comments on commit e097575

Please sign in to comment.