Initial version of CMIP7 submission code

cdds/cdds/deprecated/general_config/CMIP7/general/CMIP7.cfg

@@ -0,0 +1,10 @@
+# (C) British Crown Copyright 2019-2025, Met Office.
+# Please see LICENSE.md for license details.
+#
+# This general config file lists settings used throughout CDDS
+[transfer_facetmaps]
+valid = drs_specs|mip|date|experiment_id|grid|institution_id|mip_era|variant_label|model_id||variable|stream|output|region|branding|frequency
+atomic = drs_specs|mip|date|experiment_id|grid|institution_id|mip_era|variant_label|model_id||variable|region|branding|frequency
+name = variable|branding|frequency|region|model_id|experiment_id|variant_label|grid|[date]
+dataset_id = drs_specs|mip_era|mip|institution_id|model_id|experiment_id|variant_label|region|frequency|variable|branding|grid
+mass = drs_specs|mip_era|mip|institution_id|model_id|experiment_id|variant_label|region|frequency|variable|branding|grid

cdds/cdds/deprecated/transfer/constants.py

@@ -1,6 +1,9 @@
 # (C) British Crown Copyright 2024-2025, Met Office.
 # Please see LICENSE.md for license details.
 
-KNOWN_RABBITMQ_QUEUES = ['CMIP6_available', 'CMIP6_withdrawn', 'CMIP6Plus_available', 'CMIP6Plus_withdrawn']
+KNOWN_RABBITMQ_QUEUES = [
+    'CMIP7_available', 'CMIP7_withdrawn',
+    'CMIP6Plus_available', 'CMIP6Plus_withdrawn'
+]
 
 OPTIONAL_FACETS = ['sub_experiment_id']

cdds/cdds/deprecated/transfer/dds.py

@@ -28,8 +28,6 @@ class DataTransfer(object):
     find_mass_facets -- search MASS directories for matching facets
     inform -- inform BADC of significant MASS state changes
     rerun_change_mass_state -- complete a move that failed part-way through
-    rerun_send_to_mass -- complete a send that failed part-way through
-    send_to_mass -- copy facets from local directory to MASS
     serialise_facets -- serialise facets to a form that can be saved
     """
 
@@ -53,88 +51,6 @@ def __init__(self, config, project, simulation=False):
         self._simulation = simulation
         self._stream = {}
 
-    def send_to_mass(self, local_top, filesets, state):
-        """Send facet(s) to MASS in the specified state.
-
-        Locates filesets on local disk, deduces their path on MASS and
-        copies them across. If the specified state is one that BADC
-        are informed about, messages will be sent.
-
-        Parameters
-        ----------
-        local_top: str
-            path to top of local directory
-        filesets: drs.AtomicDatasetCollection
-            fileset(s) to send
-        state: state.State
-            state the filesets should be placed in
-        """
-        if not state.can_be_put():
-            raise ValueError("Cannot send files to MASS in state \"{}\""
-                             "".format(state))
-        for fileset in filesets:
-            self._run_put(local_top, fileset, state, moo_cmd.put)
-        return
-
-    def rerun_send_to_mass(self, local_top, filesets, state, timestamp):
-        """Re-run a MASS send that failed part way through.
-
-        Locates facets on local disk, deduces their path on MASS and
-        copies them across if necessary. If the specified state is one
-        that BADC are informed about, messages will be sent.
-
-        You need to specify the date that the original "send_to_mass"
-        was run so that the code can identify the directories that
-        should exist on MASS for the supplied facets.
-
-        Parameters
-        ----------
-        local_top: str
-            path to top of local directory
-        filesets: drs.AtomicDatasetCollection
-            fileset(s) to send
-        state: state.State
-            state the filesets should be placed in
-        timestamp: str
-            date the initial "send" method was run
-        """
-        if not state.can_be_put():
-            raise ValueError("Cannot send files to MASS in state \"{}\""
-                             "".format(state))
-        (last_id, last_var) = self._find_last_successful(
-            filesets, state, timestamp)
-        if last_id is None and last_var is None:
-            # We didn't successfully run anything, so we can just run
-            # a normal send.
-            self.send_to_mass(local_top, filesets, state)
-            return
-        # Last successful data set may have died partway through.
-        self._run_put(
-            local_top, filesets.get_drs_facet_builder(last_id, last_var),
-            state, moo_cmd.put_safe_overwrite, timestamp=timestamp)
-        # Finish off any remaining vars in the last successful id...
-        drs_vars = filesets.drs_variables(last_id)
-        if last_var != drs_vars[-1]:
-            last_loc = drs_vars.index(last_var)
-            for drs_var in drs_vars[last_loc + 1:]:
-                self._run_put(
-                    local_top,
-                    filesets.get_drs_facet_builder(last_id, drs_var),
-                    state, moo_cmd.put, timestamp=timestamp)
-        dataset_ids = filesets.dataset_ids()
-        if last_id == dataset_ids[-1]:
-            # We have nothing else left to put.
-            return
-        # and then send any remaining data sets.
-        last_loc = dataset_ids.index(last_id)
-        for dataset_id in dataset_ids[last_loc + 1:]:
-            for drs_var in filesets.drs_variables(dataset_id):
-                self._run_put(
-                    local_top,
-                    filesets.get_drs_facet_builder(dataset_id, drs_var),
-                    state, moo_cmd.put, timestamp=timestamp)
-        return
-
     def change_mass_state(
             self, filesets, old_state, new_state, timestamp=None):
         """Change the state of facet(s) in MASS.
@@ -574,10 +490,20 @@ def _prepare_message(self, drs_facet_builder, mass_dir, state):
         :class:`cdds.deprecated.transfer.msg.MooseMessage`
             Message to be sent.
         """
+        logger = logging.getLogger(__name__)
+        # the following is a somewhat unpleasant requirement to identify the version number
+        # without major disruption to the code
+
+        dataset_version_number = os.path.basename(mass_dir)
+        logger.debug(f'Inserting dataset version number "{dataset_version_number}" into message')
+        drs_facet_builder.facets['directoryDateDD'] = dataset_version_number
+        dataset_id = '.'.join([drs_facet_builder.dataset_id(), dataset_version_number])
+
         msg_content = {
-            "mass_dir": mass_dir, "state": state.name(),
+            "mass_dir": mass_dir,
+            "state": state.name(),
             "facets": drs_facet_builder.facets,
-            "dataset_id": drs_facet_builder.dataset_id()
+            "dataset_id": dataset_id
         }
         # The following is not ideal, but a signficant amount of effort
         # is needed to force all tests to use the facet "mip_era" rather than
@@ -587,6 +513,8 @@ def _prepare_message(self, drs_facet_builder, mass_dir, state):
             msg_content["mip_era"] = drs_facet_builder.facets["mip_era"]
         else:
             msg_content["mip_era"] = drs_facet_builder.facets["project"]
+
+        logger.debug('Message content: ' + json.dumps(msg_content))
         message = msg.MooseMessage(content=msg_content)
         return message
 

cdds/cdds/deprecated/transfer/drs.py

@@ -948,8 +948,18 @@ def filter_filesets(atomic_dataset_collection, variables_to_operate_on):
     """
     logger = logging.getLogger(__name__)
     for drs_facet_builder in atomic_dataset_collection:
-        key = (drs_facet_builder.facets['table_id'],
-               drs_facet_builder.facets['variable'])
+        if 'CMIP6' in drs_facet_builder.facets['mip_era']:
+            key = (
+                drs_facet_builder.facets['table_id'],
+                drs_facet_builder.facets['variable'])
+        elif 'CMIP7' in drs_facet_builder.facets['mip_era']:
+            # need to make this more intelligent to filter on grid label too
+            key = (
+                drs_facet_builder.facets['frequency'],
+                '{}_{}'.format(
+                    drs_facet_builder.facets['variable'],
+                    drs_facet_builder.facets['branding']))
+
         if key not in variables_to_operate_on:
             logger.info('Dataset "{}/{}" not included in variables list. '
                         'Skipping.'.format(*key))

cdds/cdds/deprecated/transfer/list_queue.py

@@ -24,6 +24,4 @@ def print_queue(queue_name, full=False):
     for i, message in enumerate(comm.get_all_messages(queue)):
         print(i, message.dataset_id)
         if full:
-            message_data = vars(message)
-            message_data['body'] = message_data['body'].decode('utf-8')
-            print(json.dumps(message_data, indent=2, sort_keys=True).replace('\\"', '"'))
+            print(json.dumps(message.content, indent=2, sort_keys=True).replace('\\"', '"'))

cdds/cdds/deprecated/transfer/moo.py

@@ -112,7 +112,7 @@ def run_moo_cmd(sub_cmd, args, simulation=False, logger=None):
         process = subprocess.Popen(cmd_to_run, stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE)
         (cmd_out, cmd_err) = process.communicate()
-        command_id_search = re.search(rb"(command-id=\d+)", cmd_out)
+        command_id_search = re.search(rb"(command-id=[0-9a-fA-F-]+)", cmd_out)
         if command_id_search:
             command_id = command_id_search.group(0)
         else:

cdds/cdds/deprecated/transfer/moo_cmd.py

@@ -50,7 +50,7 @@ def rmdir(moose_dir, simulation=False):
     simulation: bool
         if true simulate moo command.
     """
-    moo.run_moo_cmd("rmdir", [moose_dir], simulation=simulation)
+    moo.run_moo_cmd("rmdir", ["--force", moose_dir], simulation=simulation)
     return
 
 
@@ -184,9 +184,9 @@ def ls_tree(moose_dir, simulation=False):
         if true simulate moo command.
     """
     if simulation == LS_ONLY:
-        result = moo.run_moo_cmd("ls", ["-xR", "-p1-1000:25000", moose_dir])
+        result = moo.run_moo_cmd("ls", ["-xR", "-p", "1-1000:25000", moose_dir])
     else:
-        result = moo.run_moo_cmd("ls", ["-xR", "-p1-1000:25000", moose_dir],
+        result = moo.run_moo_cmd("ls", ["-xR", "-p", "1-1000:25000", moose_dir],
                                  simulation=simulation)
     return result
 

cdds/cdds/deprecated/transfer/msg.py

@@ -93,28 +93,18 @@ def sortable(message):
                 "Message must have been published to be sortable")
         return key
 
-    def __init__(self, content=None, body=None):
+    def __init__(self, content=None):
         """Create a new Message object.
 
-        Messages can be created either using a content dict, or using
-        a body (str in JSON format). You must supply one of the
-        arguments, and can't supply both.
+        Messages can be created using a content dict
 
         Parameters
         ----------
         content: dict
             message content
-        body: str
-            message content in JSON format
         """
-        if content and body:
-            raise ValueError("Need only one of content or body")
-        if not content and not body:
-            raise ValueError("Need content or body")
         if content:
             self._initialise_from_content(content)
-        if body:
-            self._initialise_from_body(body)
         self.delivery_tag = None
 
     def queue_prefix(self):
@@ -134,7 +124,6 @@ def add_published_ts(self):
         UTC.
         """
         self.content["published"] = Message._utc_now_timestamp()
-        self.body = self._body_from_content(self.content)
 
     def timeless_content(self):
         """Return a copy of my message with publication date removed."""
@@ -166,22 +155,8 @@ def _utc_now_timestamp():
         now = datetime.utcnow()
         return now.strftime(Message.TS_FMT)
 
-    @staticmethod
-    def _body_from_content(content):
-        return json.dumps(content)
-
-    @staticmethod
-    def _content_from_body(body):
-        return json.loads(body)
-
     def _initialise_from_content(self, content):
         self.content = content
-        self.body = self._body_from_content(content)
-        return
-
-    def _initialise_from_body(self, body):
-        self.body = body
-        self.content = self._content_from_body(body)
         return
 
     def _get_from_content(self, attr_name):
@@ -222,7 +197,7 @@ class MooseMessage(Message):
 
     TYPE = "moose"
 
-    def __init__(self, content=None, body=None):
+    def __init__(self, content=None):
         """Create a MOOSE message object.
 
         You must supply one (and only one) of either content or body
@@ -236,7 +211,7 @@ def __init__(self, content=None, body=None):
             message content in JSON format
         """
         # self.type = MooseMessage.TYPE
-        super(MooseMessage, self).__init__(content, body)
+        super(MooseMessage, self).__init__(content)
         self.dataset_id = self.content.get('dataset_id', None)
 
     def queue_prefix(self):
@@ -326,21 +301,18 @@ def queue_prefix(self):
         """Return queue prefix (str) for admin messages."""
         return AdminMessage.TYPE
 
-    def __init__(self, content=None, body=None):
+    def __init__(self, content):
         """Create a new admin message object.
 
-        You must supply one (and only one) of content or body to
-        create the message.
+        You must supply content to create the message.
 
         Parameters
         ----------
         content: dict
             message content
-        body: str
-            message content in JSON format
         """
         # self.type = AdminMessage.TYPE
-        super(AdminMessage, self).__init__(content, body)
+        super(AdminMessage, self).__init__(content)
 
     def queue_suffix(self):
         """Return the queue suffix (str) for this message."""
@@ -561,7 +533,7 @@ def publish_message(self, message, queue=None):
                          ''.format(queue.queue_name,
                                    message.queue().queue_name))
         published = False
-        channel_callable = rabbit.PersistentPublish(queue, message.body)
+        channel_callable = rabbit.PersistentPublish(queue, message.content)
         if self._rabbit_mgr.call(channel_callable):
             published = True
         if not published:
@@ -590,8 +562,8 @@ def get_first_matching_message(self, queue):
         channel_callable = rabbit.GetFirst(queue)
         result = self._rabbit_mgr.call(channel_callable)
         if result:
-            (method_frame, body) = result
-            message = self._make_message(queue, method_frame, body)
+            (method_frame, content) = result
+            message = self._make_message(queue, method_frame, content)
         return message
 
     def get_all_messages(self, queue):
@@ -616,8 +588,8 @@ def get_all_messages(self, queue):
         result = self._rabbit_mgr.call(channel_callable)
         if result:
             for msg in result:
-                (method_frame, body) = msg
-                message = self._make_message(queue, method_frame, body)
+                (method_frame, content) = msg
+                message = self._make_message(queue, method_frame, content)
                 messages.append(message)
         return messages
 
@@ -652,10 +624,10 @@ def store_message(self, message):
         message_store.store_message(message)
         return
 
-    def _make_message(self, queue, method_frame, body):
+    def _make_message(self, queue, method_frame, content):
         msg_type = queue.message_class()
         if msg_type:
-            message = msg_type(body=body)
+            message = msg_type(content=json.loads(content))
             message.delivery_tag = method_frame.delivery_tag
         else:
             message = None
@@ -764,18 +736,18 @@ def _save_message(self, msg_base, content):
 
         logger.debug('Writing message to file "{}"'.format(msg_file))
 
-        fh = open(msg_file, "w")
-        json.dump(content, fh)
-        fh.close()
+        with open(msg_file, "w") as fh:
+            json.dump(content, fh, indent=2, sort_keys=True)
+
         return
 
     def _read_message(self, msg_base):
         msg_file = self._msg_full_path(msg_base)
         if not os.path.exists(msg_file):
             raise IOError("No message file %s" % msg_file)
-        fh = open(msg_file)
-        content = json.load(fh)
-        fh.close()
+        with open(msg_file) as fh:
+            content = json.load(fh)
+
         return content
 
     def _msg_full_path(self, msg_base):