[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

romancal/associations/load_as_asn.py

@@ -3,27 +3,22 @@
 from functools import partial
 from os import path as os_path
 
-from ..associations import (
-    Association,
-    AssociationRegistry,
-    libpath,
-    load_asn
-)
+from ..associations import Association, AssociationRegistry, libpath, load_asn
 from ..associations.asn_from_list import asn_from_list
 from ..associations.lib.rules_elpp_base import DMS_ELPP_Base
 from ..associations.lib.rules_level2 import Asn_Lv2Image
 
 __all__ = [
-    'LoadAsAssociation',
-    'LoadAsLevel2Asn',
+    "LoadAsAssociation",
+    "LoadAsLevel2Asn",
 ]
 
 
-DEFAULT_NAME = 'singleton'
+DEFAULT_NAME = "singleton"
 DEFAULT_ASN_META = {
-    'program': DEFAULT_NAME,
-    'target': DEFAULT_NAME,
-    'asn_pool': DEFAULT_NAME
+    "program": DEFAULT_NAME,
+    "target": DEFAULT_NAME,
+    "asn_pool": DEFAULT_NAME,
 }
 
 
@@ -45,12 +40,15 @@ class LoadAsAssociation(dict):
     """
 
     @classmethod
-    def load(cls, obj,
-             meta=DEFAULT_ASN_META,
-             registry=AssociationRegistry,
-             rule=Association,
-             product_name_func=None):
-        """ Load object and return an association of it
+    def load(
+        cls,
+        obj,
+        meta=DEFAULT_ASN_META,
+        registry=AssociationRegistry,
+        rule=Association,
+        product_name_func=None,
+    ):
+        """Load object and return an association of it
 
         Parameters
         ----------
@@ -93,7 +91,7 @@ def load(cls, obj,
                 obj,
                 rule=rule,
                 meta=DEFAULT_ASN_META,
-                product_name_func=product_name_func
+                product_name_func=product_name_func,
             )
             asn.filename = DEFAULT_NAME
         else:
@@ -105,12 +103,11 @@ def load(cls, obj,
 
 
 class LoadAsLevel2Asn(LoadAsAssociation):
-    """Read in or create a Level2 association
-    """
+    """Read in or create a Level2 association"""
 
     @classmethod
     def load(cls, obj, basename=None):
-        """ Open object and return a Level2 association of it
+        """Open object and return a Level2 association of it
 
         Parameters
         ----------
@@ -142,21 +139,24 @@ def load(cls, obj, basename=None):
         if isinstance(obj, str):
             file_name, file_ext = os_path.splitext(obj)
 
-            if file_ext == '.fits':
-                items = [(obj, 'science')]
-                asn = asn_from_list(items, product_name=file_name,
-                                    rule=DMS_ELPP_Base, with_exptype=True,
-                                    meta={"asn_pool":"singleton"})
+            if file_ext == ".fits":
+                items = [(obj, "science")]
+                asn = asn_from_list(
+                    items,
+                    product_name=file_name,
+                    rule=DMS_ELPP_Base,
+                    with_exptype=True,
+                    meta={"asn_pool": "singleton"},
+                )
                 return asn
 
-        asn = super(LoadAsLevel2Asn, cls).load(
+        asn = super().load(
             obj,
             registry=AssociationRegistry(
-                definition_files=[libpath('rules_level2.py')],
-                include_default=False
+                definition_files=[libpath("rules_level2.py")], include_default=False
             ),
             rule=Asn_Lv2Image,
-            product_name_func=product_name_func
+            product_name_func=product_name_func,
         )
         return asn
 
@@ -203,5 +203,5 @@ def name_with_index(basename, obj, idx, *args, **kwargs):
         """
         basename, extension = os_path.splitext(os_path.basename(basename))
         if idx > 1:
-            basename = basename + '_' + str(idx)
+            basename = basename + "_" + str(idx)
         return basename

romancal/datamodels/tests/data/fake.asdf

@@ -1 +1 @@
-not actually an ASDF file
+not actually an ASDF file

romancal/pipeline/exposure_pipeline.py

@@ -4,12 +4,13 @@
 
 import numpy as np
 from roman_datamodels import datamodels as rdm
+
 import romancal.datamodels.filetype as filetype
-from romancal.associations.load_as_asn import LoadAsLevel2Asn
-from romancal.associations.exceptions import AssociationNotValidError
 
 # step imports
 from romancal.assign_wcs import AssignWcsStep
+from romancal.associations.exceptions import AssociationNotValidError
+from romancal.associations.load_as_asn import LoadAsLevel2Asn
 from romancal.dark_current import DarkCurrentStep
 from romancal.dq_init import dq_init_step
 from romancal.flatfield import FlatFieldStep
@@ -76,29 +77,28 @@ def process(self, input):
         # open the input file
         file_type = filetype.check(input)
         asn = None
-        if file_type == 'asdf':
+        if file_type == "asdf":
             try:
                 input = rdm.open(input)
             except TypeError:
                 log.debug("Error opening file:")
                 return
-
-        if file_type == 'asn':
-            save_results = True
+
+        if file_type == "asn":
             try:
                 asn = LoadAsLevel2Asn.load(input, basename=self.output_file)
             except AssociationNotValidError:
                 log.debug("Error opening file:")
                 return
-            
+
         # Build a list of observations to process
         expos_file = []
-        if file_type == 'asdf':
+        if file_type == "asdf":
             expos_file = [input]
-        elif file_type == 'asn':
-            for product in asn['products']:
-                for member in product['members']:
-                    expos_file.append(member['expname'])
+        elif file_type == "asn":
+            for product in asn["products"]:
+                for member in product["members"]:
+                    expos_file.append(member["expname"])
 
         results = []
         for in_file in expos_file:
@@ -107,8 +107,8 @@ def process(self, input):
                 log.info(f"Input file name: {input_filename}")
             else:
                 input_filename = None
-                
-            #Open the file
+
+            # Open the file
             input = rdm.open(in_file)
             log.info(f"Processing a WFI exposure {in_file}")
 
@@ -155,7 +155,7 @@ def process(self, input):
             else:
                 log.info("Flat Field step is being SKIPPED")
                 result.meta.cal_step.flat_field = "SKIPPED"
-                
+
             if result.meta.exposure.type == "WFI_IMAGE":
                 result = self.photom(result)
                 result = self.source_detection(result)
@@ -164,8 +164,6 @@ def process(self, input):
                 result.meta.cal_step.photom = "SKIPPED"
                 result.meta.cal_step.source_detection = "SKIPPED"
 
-
-
             # setup output_file for saving
             self.setup_output(result)
             log.info("Roman exposure calibration pipeline ending...")