[Feature] Management Command to Compute and Store Reaction <-> Rule Associations (#453)

Co-authored-by: Tim Lorsbach <tim@lorsba.ch>
Reviewed-on: enviPath/enviPy#453

epdb/management/commands/localize_urls.py

@@ -44,15 +44,19 @@ class Command(BaseCommand):
             "EPModel",
             "ApplicabilityDomain",
             "EnzymeLink",
+            "AdditionalInformation",
         ]
         for model in MODELS:
             obj_cls = apps.get_model("epdb", model)
-            obj_cls.objects.update(
+
-                url=Replace(F("url"), Value(options["old"]), Value(options["new"]))
+            update_fields = {"url": Replace(F("url"), Value(options["old"]), Value(options["new"]))}
+            if hasattr(obj_cls, "description"):
+                update_fields["description"] = Replace(
+                    F("description"), Value(options["old"]), Value(options["new"])
                 )
+
             if issubclass(obj_cls, EnviPathModel):
-                obj_cls.objects.update(
+                update_fields["kv"] = Cast(
-                    kv=Cast(
                     Replace(
                         Cast(F("kv"), output_field=TextField()),
                         Value(options["old"]),
@@ -60,4 +64,5 @@ class Command(BaseCommand):
                     ),
                     output_field=JSONField(),
                 )
-                )
+
+            obj_cls.objects.update(**update_fields)

epdb/management/commands/reaction_rule_mapping.py

@@ -0,0 +1,97 @@
+import logging
+
+from django.core.management.base import BaseCommand
+from django.db import transaction
+from uuid import uuid4
+from epdb.models import Package, ReactionExplanation
+from utilities.chem import FormatConverter
+from django.utils import timezone
+
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--rule-package",
+            action="append",
+            default=["32de3cf4-e3e6-4168-956e-32fa5ddb0ce1"],
+            type=str,
+            help="UUID to process. Can be specified multiple times.",
+        )
+
+        parser.add_argument(
+            "--reaction-package",
+            action="append",
+            default=[
+                "32de3cf4-e3e6-4168-956e-32fa5ddb0ce1",  # BBD
+                "f05e38d8-e9b4-4c3e-b0d8-9ab29966eccf",  # Sediment
+                "521c547a-fd2a-491c-ad5b-7eaa1577fb65",  # Sludge
+                "5882df9c-dae1-4d80-a40e-db4724271456",  # Soil
+                "87a49584-d937-482c-9c33-25928dcb02a8",  # PFAS
+            ],
+            type=str,
+            help="UUID to process. Can be specified multiple times.",
+        )
+
+        parser.add_argument(
+            "--dry-run",
+            default=False,
+            action="store_true",
+            help="Perform dry run",
+        )
+
+    @transaction.atomic
+    def handle(self, *args, **options):
+        RUN_UUID = uuid4()
+        RUN_START = timezone.now()
+
+        rule_packages = Package.objects.filter(uuid__in=options["rule_package"])
+        reaction_packages = Package.objects.filter(uuid__in=options["reaction_package"])
+
+        rules = []
+        for rule_package in rule_packages:
+            rules.extend(rule_package.get_applicable_rules())
+
+        reactions = []
+        for reaction_package in reaction_packages:
+            reactions.extend(reaction_package.reactions)
+
+        logger.debug(f"Collected {len(rules)} rules and {len(reactions)} reactions.")
+
+        for i, reaction in enumerate(reactions):
+            logger.debug(f"Reaction {i} / {len(reactions)}")
+            for j, rule in enumerate(rules):
+                reactants, products = reaction.smirks().split(">>")
+
+                if len(reactants.split(".")) > 1:
+                    logger.debug(f"Skipping reaction {reaction.uuid} as it has multiple reactants.")
+                    break
+
+                products = products.split(".")
+
+                # Run reaction with rule
+                rule_products = rule.apply(reactants)
+
+                # Check if products match (in both directions if extras are not allowed)
+                for product_set in rule_products:
+                    covered, exact = FormatConverter.smiles_covered_by(
+                        products,
+                        product_set.product_set,
+                        standardize=True,
+                        canonicalize_tautomers=True,
+                        return_exact_match=True,
+                    )
+
+                    if covered and not options["dry-run"]:
+                        logger.debug(f"Reaction {reaction.uuid} explained by rule {rule.uuid}")
+                        re = ReactionExplanation()
+                        re.run_uuid = RUN_UUID
+                        re.run_start = RUN_START
+                        re.reaction = reaction
+                        re.rule = rule
+                        re.exact = exact
+                        re.save()
+                        # Its explained, if there are more sets skip them
+                        break

epdb/migrations/0029_reactionexplanation_reaction_explained_by.py

@@ -0,0 +1,63 @@
+# Generated by Django 6.0.3 on 2026-08-13 09:58
+
+import django.db.models.deletion
+import django.utils.timezone
+import model_utils.fields
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("epdb", "0028_auto_20260812_0902"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="ReactionExplanation",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
+                    ),
+                ),
+                (
+                    "created",
+                    model_utils.fields.AutoCreatedField(
+                        default=django.utils.timezone.now, editable=False, verbose_name="created"
+                    ),
+                ),
+                (
+                    "modified",
+                    model_utils.fields.AutoLastModifiedField(
+                        default=django.utils.timezone.now, editable=False, verbose_name="modified"
+                    ),
+                ),
+                ("run_uuid", models.UUIDField()),
+                ("run_start", models.DateTimeField()),
+                ("exact", models.BooleanField(default=False)),
+                (
+                    "reaction",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="epdb.reaction"
+                    ),
+                ),
+                (
+                    "rule",
+                    models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to="epdb.rule"),
+                ),
+            ],
+            options={
+                "abstract": False,
+            },
+        ),
+        migrations.AddField(
+            model_name="reaction",
+            name="explained_by",
+            field=models.ManyToManyField(
+                related_name="explained_reactions",
+                through="epdb.ReactionExplanation",
+                to="epdb.rule",
+            ),
+        ),
+    ]

epdb/migrations/0030_auto_20260814_0741.py

@@ -0,0 +1,37 @@
+# Generated by Django 6.0.3 on 2026-08-14 07:41
+
+from django.db import migrations
+
+
+def forward_func(apps, schema_editor):
+    ContentType = apps.get_model("contenttypes", "ContentType")
+    AdditionalInformation = apps.get_model("epdb", "AdditionalInformation")
+
+    models = {}
+
+    for c in ContentType.objects.all():
+        try:
+            models[(c.app_label, c.model)] = apps.get_model(c.app_label, c.model)
+        except Exception:
+            pass
+
+    for ai in AdditionalInformation.objects.all():
+        if ai.url is None:
+            if ai.content_type is None:
+                ai.url = "{}/additional-information/{}".format(ai.scenario.url, ai.uuid)
+            else:
+                model = models[(ai.content_type.app_label, ai.content_type.model)]
+                obj = model.objects.get(pk=ai.object_id)
+                ai.url = "{}/additional-information/{}".format(obj.url, ai.uuid)
+
+            ai.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("epdb", "0029_reactionexplanation_reaction_explained_by"),
+    ]
+
+    operations = [
+        migrations.RunPython(forward_func, reverse_code=migrations.RunPython.noop),
+    ]

epdb/models.py

@@ -1730,6 +1730,14 @@ class SequentialRuleOrdering(models.Model):
     order_index = models.IntegerField(null=False, blank=False)
 
 
+class ReactionExplanation(TimeStampedModel):
+    run_uuid = models.UUIDField(null=False, blank=False)
+    run_start = models.DateTimeField(null=False, blank=False)
+    reaction = models.ForeignKey("epdb.Reaction", on_delete=models.CASCADE)
+    rule = models.ForeignKey("epdb.Rule", on_delete=models.CASCADE)
+    exact = models.BooleanField(default=False)
+
+
 class Reaction(
     EnviPathModel, AliasMixin, ScenarioMixin, ReactionIdentifierMixin, AdditionalInformationMixin
 ):
@@ -1755,6 +1763,12 @@ class Reaction(
 
     external_identifiers = GenericRelation("ExternalIdentifier")
 
+    explained_by = models.ManyToManyField(
+        "epdb.Rule",
+        through="ReactionExplanation",
+        related_name="explained_reactions",
+    )
+
     def _url(self):
         return "{}/reaction/{}".format(self.package.url, self.uuid)
 

utilities/chem.py

@@ -448,6 +448,7 @@ class FormatConverter(object):
         r_smiles: List[str],
         standardize: bool = True,
         canonicalize_tautomers: bool = True,
+        return_exact_match: bool = False,
     ) -> bool:
         """
         Check if all SMILES in the left list are covered by (contained in) the right list.
@@ -517,8 +518,11 @@ class FormatConverter(object):
                 standardized_r_smiles.append(smi)
         else:
             standardized_r_smiles = r_smiles
-
+        if not return_exact_match:
             return len(set(standardized_l_smiles).difference(set(standardized_r_smiles))) == 0
+        return len(set(standardized_l_smiles).difference(set(standardized_r_smiles))) == 0, set(
+            standardized_l_smiles
+        ) == set(standardized_r_smiles)
 
 
 class Standardizer(ABC):