Rcal 910 Add target option to the asn_from_list command

CHANGES.rst

@@ -1,6 +1,12 @@
 0.16.3 (2024-08-29)
 ===================
 
+associations
+------------
+
+- Add target option for asn_from_list command [#1411]
+
+
 mosaic_pipeline
 ---------------
 

changes/1411.associations.rst

@@ -0,0 +1 @@
+add target to asn_from_list command

docs/roman/associations/asn_from_list.rst

@@ -9,13 +9,12 @@ Create an association using either the command line tool
 :func:`romancal.associations.asn_from_list.asn_from_list`
 
 
-Level2 Associations
-^^^^^^^^^^^^^^^^^^^
+Associations
+^^^^^^^^^^^^
 
-Refer to TBD for a full description of Level2
-associations.
+Refer to TBD for a full description of associations.
 
-To create a Level2 association, use the following command:
+To create an association, use the following command:
 
 .. code-block:: python
 
@@ -43,6 +42,12 @@ To create a association with all the detectors for a given exposure from the com
 
 .. code-block:: python
 
-		asn_from_list -o detector_asn.json --product-name r0000101001001001001_01101_0001_WFI_cal.asdf data/*_cal.asdf
+		asn_from_list -o detector_asn.json --product-name r0000101001001001001_01101_0001_WFI data/*_cal.asdf
 
 where the individual calibrated detector files are in a data subdirectory.
+
+In addition you can also specify a target for the association file,
+
+.. code-block:: python
+
+		asn_from_list -o level3_mosaic_asn.json --product-name level3_mosaic --target r274dp63x32y80  data/*_cal.asdf

romancal/associations/asn_from_list.py

@@ -37,6 +37,9 @@ def asn_from_list(items, rule=DMS_ELPP_Base, **kwargs):
     """
     asn = rule()
     asn._add_items(items, **kwargs)
+    if "target" in kwargs.keys():
+        target = kwargs["target"]
+        asn["target"] = target
     return asn
 
 
@@ -110,6 +113,14 @@ def __init__(self, args=None):
             help='The association candidate id to use. Default: "%(default)s"',
             dest="acid",
         )
+        parser.add_argument(
+            "-t",
+            "--target",
+            type=str,
+            default="None",
+            help='The target name for the association. Default: "%(default)s"',
+            dest="target",
+        )
 
         parser.add_argument(
             "filelist",
@@ -130,6 +141,7 @@ def __init__(self, args=None):
                 rule=rule,
                 product_name=parsed.product_name,
                 acid=parsed.acid,
+                target=parsed.target,
             )
             _, serialized = asn.dump(format=parsed.format)
             outfile.write(serialized)

romancal/associations/association.py

@@ -87,19 +87,21 @@
     def __init__(
         self,
         version_id=None,
+        target=None,
     ):
-        self.data = dict()
+        self.data = {}
         self.run_init_hook = True
         self.meta = {}
 
         self.version_id = version_id
-
+        self.target = target
         self.data.update(
             {
                 "asn_type": "None",
                 "asn_rule": self.asn_rule,
                 "version_id": self.version_id,
                 "code_version": __version__,
+                "target": self.target,
             }
         )
 
@@ -135,6 +137,7 @@
                 - [ProcessList[, ...]]: List of items to process again.
         """
         asn = cls(version_id=version_id)
+
         matches, reprocess = asn.add(item)
         if not matches:
             return None, reprocess
@@ -227,8 +230,8 @@
         """
         if self.is_valid:
             return self.ioregistry[format].dump(self, **kwargs)
-        else:
-            raise AssociationNotValidError(f"Association {self} is not valid")
+
+        raise AssociationNotValidError(f"Association {self} is not valid")
 
     @classmethod
     def load(cls, serialized, format=None, validate=True, **kwargs):
@@ -430,8 +433,8 @@
         """
         if self.is_valid:
             return [self]
-        else:
-            return None
+
+        return None
 
     def is_item_member(self, item):
         """Check if item is already a member of this association
@@ -534,6 +537,7 @@
 
 
 def make_timestamp():
+    """Generate a timestamp based on runtime"""
     timestamp = datetime.utcnow().strftime(_TIMESTAMP_TEMPLATE)
     return timestamp
 

romancal/associations/lib/asn_schema_jw_level3.json

@@ -23,7 +23,14 @@
         },
         "target": {
             "description": "Canonical name of the astronomical object being observed.",
-            "type": "string"
+            "anyOf":[
+                {
+                    "type": "string"
+                },
+                {
+                    "type": "null"
+                }
+            ]
         },
         "asn_pool": {
             "description": "Name of the Association Pool from which this association was generated.",

romancal/associations/lib/dms_base.py

@@ -189,6 +189,7 @@ def asn_name(self):
         version_id = self.version_id
         asn_type = self.data["asn_type"]
         sequence = self.sequence
+        target = self.target
 
         if version_id:
             name = _ASN_NAME_TEMPLATE_STAMP.format(
@@ -197,13 +198,15 @@ def asn_name(self):
                 stamp=version_id,
                 type=asn_type,
                 sequence=sequence,
+                target=target,
             )
         else:
             name = _ASN_NAME_TEMPLATE.format(
                 program=program,
                 acid=self.acid.id,
                 type=asn_type,
                 sequence=sequence,
+                target=target,
             )
         return name.lower()
 
@@ -402,16 +405,19 @@ def update_degraded_status(self):
                             break
 
     def update_validity(self, entry):
+        """Check/Update the validity of the association"""
         for test in self.validity.values():
             if not test["validated"]:
                 test["validated"] = test["check"](entry)
 
     @classmethod
     def reset_sequence(cls):
+        """Reset the sequence counter to 1"""
         cls._sequence = Counter(start=1)
 
     @classmethod
     def validate(cls, asn):
+        """Validate the association generated"""
         super().validate(asn)
 
         if isinstance(asn, DMSBaseMixin):

romancal/associations/lib/rules_elpp_base.py

@@ -356,8 +356,8 @@
             if asn.is_valid:
                 results.append(asn)
                 return results
-            else:
-                return None
+
+            return None
 
     def _init_hook(self, item):
         """Post-check and pre-add initialization"""
@@ -394,6 +394,11 @@
         # Update meta info
         self.update_asn(item=item, member=member)
 
+    def update_target(self, target):
+        """Update association target field"""
+        if self.data["target"] is None:
+            self.data["target"] = target
+
     def _add_items(self, items, product_name=None, with_exptype=False, **kwargs):
         """Force adding items to the association
 
@@ -439,7 +444,7 @@
     def __repr__(self):
         # flake8:  noqa: F821
         try:
-            file_name, json_repr = self.ioregistry["json"].dump(self)
+            _, json_repr = self.ioregistry["json"].dump(self)
         except KeyValueRegistryError:
             return str(self.__class__)
         return json_repr
@@ -1031,8 +1036,8 @@
         """
         if self.is_valid:
             return self.make_fov_asn()
-        else:
-            return None
+
+        return None
 
 
 class AsnMixin_Lv2Image:

romancal/associations/tests/test_asn_from_list.py

@@ -27,6 +27,25 @@ def test_base_roundtrip():
     assert asn["members"] == reloaded["members"]
 
 
+def test_association_target():
+    """Create the simple associations with target set"""
+    items = ["a", "b", "c"]
+    target_name = "r274dp63x32y80"
+    product_name = "l3_target"
+    rule_name = "Association"
+    asn = asn_from_list(
+        items,
+        rule=Association,
+        product_name=product_name,
+        target=target_name,
+        version_id="c55",
+    )
+    assert asn["asn_rule"] == rule_name
+    assert asn["asn_type"] == "None"
+    assert asn["members"] == items
+    assert asn["target"] == target_name
+
+
 def test_default_simple():
     """Default ELPP association"""
     product_name = "test_product"