Rcal 910 Add target option to the asn_from_list command

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"