silver: add Relation schema and RelationBuilder; map SIGNOR interactions as relations

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: jonsch1

pypath/inputs_v2/signor.py

@@ -28,6 +28,7 @@
     Member,
     MembersFromList,
     MembershipBuilder,
+    RelationBuilder,
 )
 
 
@@ -395,8 +396,58 @@ def parsed_annotation_values(column_name: str, *, default_term: str | None = Non
     ),
 )
 
-interactions_schema = EntityBuilder(
-    entity_type=EntityTypeCv.INTERACTION,
+def signor_predicate(row: dict[str, object]) -> str:
+    raw_causal = str(row.get('Causal statement') or '').strip()
+    if raw_causal and raw_causal != '-':
+        mi_match = re.search(_MI_REGEX, raw_causal)
+        if mi_match:
+            return mi_match.group(1)
+    raw_type = str(row.get('Interaction type(s)') or '').strip()
+    if raw_type and raw_type != '-':
+        mi_match = re.search(_MI_REGEX, raw_type)
+        if mi_match:
+            return mi_match.group(1)
+    return 'interacts_with'
+
+
+interactor_a_builder = EntityBuilder(
+    entity_type=interactor_entity_type('A'),
+    identifiers=IdentifiersBuilder(
+        general_identifier_cv('\ufeff#ID(s) interactor A'),
+        general_identifier_cv('Alt. ID(s) interactor A'),
+        interactor_tax_cv('A'),
+    ),
+    annotations=AnnotationsBuilder(
+        mi_term_cv('Biological role(s) interactor A'),
+        mi_term_cv('Experimental role(s) interactor A'),
+        CV(
+            term=parsed_annotation_terms('Feature(s) interactor A'),
+            value=parsed_annotation_values('Feature(s) interactor A'),
+        ),
+    ),
+)
+
+interactor_b_builder = EntityBuilder(
+    entity_type=interactor_entity_type('B'),
+    identifiers=IdentifiersBuilder(
+        general_identifier_cv('ID(s) interactor B'),
+        general_identifier_cv('Alt. ID(s) interactor B'),
+        interactor_tax_cv('B'),
+    ),
+    annotations=AnnotationsBuilder(
+        mi_term_cv('Biological role(s) interactor B'),
+        mi_term_cv('Experimental role(s) interactor B'),
+        CV(
+            term=parsed_annotation_terms('Feature(s) interactor B'),
+            value=parsed_annotation_values('Feature(s) interactor B'),
+        ),
+    ),
+)
+
+interactions_schema = RelationBuilder(
+    subject=interactor_a_builder,
+    predicate=signor_predicate,
+    object=interactor_b_builder,
     identifiers=IdentifiersBuilder(
         interaction_identifier_cv(),
     ),
@@ -417,44 +468,6 @@ def parsed_annotation_values(column_name: str, *, default_term: str | None = Non
             ),
         ),
     ),
-    membership=MembershipBuilder(
-        Member(
-            entity=EntityBuilder(
-                entity_type=interactor_entity_type('A'),
-                identifiers=IdentifiersBuilder(
-                    general_identifier_cv('\ufeff#ID(s) interactor A'),
-                    general_identifier_cv('Alt. ID(s) interactor A'),
-                ),
-                annotations=AnnotationsBuilder(interactor_tax_cv('A')),
-            ),
-            annotations=AnnotationsBuilder(
-                mi_term_cv('Biological role(s) interactor A'),
-                mi_term_cv('Experimental role(s) interactor A'),
-                CV(
-                    term=parsed_annotation_terms('Feature(s) interactor A'),
-                    value=parsed_annotation_values('Feature(s) interactor A'),
-                ),
-            ),
-        ),
-        Member(
-            entity=EntityBuilder(
-                entity_type=interactor_entity_type('B'),
-                identifiers=IdentifiersBuilder(
-                    general_identifier_cv('ID(s) interactor B'),
-                    general_identifier_cv('Alt. ID(s) interactor B'),
-                ),
-                annotations=AnnotationsBuilder(interactor_tax_cv('B')),
-            ),
-            annotations=AnnotationsBuilder(
-                mi_term_cv('Biological role(s) interactor B'),
-                mi_term_cv('Experimental role(s) interactor B'),
-                CV(
-                    term=parsed_annotation_terms('Feature(s) interactor B'),
-                    value=parsed_annotation_values('Feature(s) interactor B'),
-                ),
-            ),
-        ),
-    ),
 )
 
 resource = Resource(

pypath/internals/silver_schema.py

@@ -23,7 +23,9 @@
     'EntityRef',
     'OntologyRelation',
     'Entity',
+    'Relation',
     'ENTITY_SCHEMA',
+    'RELATION_SCHEMA',
 ]
 
 class Identifier(NamedTuple):
@@ -115,7 +117,7 @@ class OntologyRelation(NamedTuple):
 class Entity(NamedTuple):
     """ Entity record matching entities schema."""
 
-    type: EntityTypeCv # e.g. EntityTypeCv.INTERACTION, EntityTypeCv.CV_TERM
+    type: EntityTypeCv | str # e.g. EntityTypeCv.INTERACTION, EntityTypeCv.CV_TERM
     identifiers: List[Identifier]  # e.g. IdentifierNamespaceCv.NAME and .SYNONYM)
     annotations: List[Annotation] | None = None
 
@@ -195,6 +197,65 @@ def pretty(self, indent: int = 0) -> str:
 
         return "\n".join(lines)
 
+
+class Relation(NamedTuple):
+    """Relation record representing a directed edge/interaction between entities."""
+
+    subject: EntityRef | Entity
+    predicate: str
+    object: EntityRef | Entity
+    category: str | None = None
+    interaction_class: str | None = None
+    identifiers: List[Identifier] | None = None
+    annotations: List[Annotation] | None = None
+
+    def __repr__(self) -> str:
+        """Compact representation of relation."""
+        sub_str = repr(self.subject)
+        obj_str = repr(self.object)
+        pred_str = cv_term_label_accession(self.predicate) if hasattr(self.predicate, 'value') else str(self.predicate)
+        extra = []
+        if self.category:
+            extra.append(f"category={self.category}")
+        if self.interaction_class:
+            extra.append(f"interaction_class={self.interaction_class}")
+        if self.identifiers:
+            extra.append(f"identifiers={self.identifiers!r}")
+        if self.annotations:
+            extra.append(f"annotations={self.annotations!r}")
+        extra_str = f" ({', '.join(extra)})" if extra else ""
+        return f"Relation({sub_str} -[{pred_str}]-> {obj_str}{extra_str})"
+
+    def __str__(self) -> str:
+        return repr(self)
+
+    def pretty(self, indent: int = 0) -> str:
+        """Tree-like representation with indentation."""
+        prefix = "  " * indent
+        pred_str = cv_term_label_accession(self.predicate) if hasattr(self.predicate, 'value') else str(self.predicate)
+        lines = [f"{prefix}Relation: {pred_str}"]
+        lines.append(f"{prefix}├─ subject:")
+        sub_lines = (self.subject.pretty(0) if hasattr(self.subject, 'pretty') else repr(self.subject)).split("\n")
+        for line in sub_lines:
+            lines.append(f"{prefix}│  {line}")
+        lines.append(f"{prefix}├─ object:")
+        obj_lines = (self.object.pretty(0) if hasattr(self.object, 'pretty') else repr(self.object)).split("\n")
+        for line in obj_lines:
+            lines.append(f"{prefix}│  {line}")
+        if self.category:
+            lines.append(f"{prefix}├─ category: {self.category}")
+        if self.interaction_class:
+            lines.append(f"{prefix}├─ interaction_class: {self.interaction_class}")
+        if self.identifiers:
+            lines.append(f"{prefix}├─ identifiers: ({len(self.identifiers)})")
+            for id_ in self.identifiers:
+                lines.append(f"{prefix}│  ├─ {id_!r}")
+        if self.annotations:
+            lines.append(f"{prefix}└─ annotations: ({len(self.annotations)})")
+            for ann in self.annotations:
+                lines.append(f"{prefix}   └─ {ann!r}")
+        return "\n".join(lines)
+
 #### PyArrow Schemas (needed for parquet files) ####
 
 # Reusable field definitions
@@ -273,3 +334,17 @@ def pretty(self, indent: int = 0) -> str:
 ]
 
 ENTITY_SCHEMA = pa.schema(ENTITY_FIELDS)
+
+# Full relation schema
+RELATION_FIELDS = [
+    pa.field('subject', pa.struct(NESTED_ENTITY_FIELDS), nullable=False),
+    pa.field('predicate', pa.string(), nullable=False),
+    pa.field('object', pa.struct(NESTED_ENTITY_FIELDS), nullable=False),
+    pa.field('category', pa.string()),
+    pa.field('interaction_class', pa.string()),
+    pa.field('identifiers', pa.list_(pa.struct(IDENTIFIER_FIELDS))),
+    pa.field('annotations', pa.list_(pa.struct(ANNOTATION_FIELDS))),
+]
+
+RELATION_SCHEMA = pa.schema(RELATION_FIELDS)
+

pypath/internals/tabular_builder.py

@@ -23,6 +23,7 @@
     Identifier as SilverIdentifier,
     Membership as SilverMembership,
     OntologyRelation as SilverOntologyRelation,
+    Relation as SilverRelation,
 )
 from pypath.internals.cv_terms import (
     EntityTypeCv,
@@ -1318,6 +1319,117 @@ def _allows_empty_identifiers(self, entity_type: Any) -> bool:
         }
 
 
+class RelationBuilder:
+    """Declarative spec that produces `Relation` records from rows."""
+
+    def __init__(
+        self,
+        *,
+        subject: EntityBuilder | Callable[[Any], SilverEntity | SilverEntityRef | None],
+        predicate: str | Column | Callable[[Any], str | None],
+        object: EntityBuilder | Callable[[Any], SilverEntity | SilverEntityRef | None],
+        category: str | Column | Callable[[Any], str | None] | None = None,
+        interaction_class: str | Column | Callable[[Any], str | None] | None = None,
+        identifiers: IdentifiersBuilder | None = None,
+        annotations: AnnotationsBuilder | None = None,
+    ) -> None:
+        self.subject = subject
+        self.predicate = predicate
+        self.object = object
+        self.category = category
+        self.interaction_class = interaction_class
+        self.identifiers = identifiers
+        self.annotations = annotations
+
+    def __call__(self, row: Any) -> SilverRelation | None:
+        return self.build(row)
+
+    def build(self, row: Any) -> SilverRelation | None:
+        cache = ColumnCache()
+
+        # Build subject entity
+        if isinstance(self.subject, EntityBuilder):
+            subject_entity = self.subject.build(row)
+        elif callable(self.subject):
+            try:
+                subject_entity = self.subject(row)
+            except Exception as exc:  # pragma: no cover - defensive
+                logger.debug("Relation subject callable failed: %s", exc)
+                return None
+        else:
+            subject_entity = self.subject
+
+        if subject_entity is None:
+            return None
+
+        # Build object entity
+        if isinstance(self.object, EntityBuilder):
+            object_entity = self.object.build(row)
+        elif callable(self.object):
+            try:
+                object_entity = self.object(row)
+            except Exception as exc:  # pragma: no cover - defensive
+                logger.debug("Relation object callable failed: %s", exc)
+                return None
+        else:
+            object_entity = self.object
+
+        if object_entity is None:
+            return None
+
+        # Resolve predicate
+        if isinstance(self.predicate, Column):
+            pred_vals = cache.values(self.predicate, row)
+            predicate_val = str(pred_vals[0]) if pred_vals else 'interacts_with'
+        elif callable(self.predicate):
+            try:
+                predicate_val = str(self.predicate(row) or 'interacts_with')
+            except Exception as exc:  # pragma: no cover - defensive
+                logger.debug("Relation predicate callable failed: %s", exc)
+                predicate_val = 'interacts_with'
+        else:
+            predicate_val = str(self.predicate)
+
+        # Resolve optional category
+        category_val = None
+        if isinstance(self.category, Column):
+            cat_vals = cache.values(self.category, row)
+            category_val = str(cat_vals[0]) if cat_vals else None
+        elif callable(self.category):
+            try:
+                category_val = self.category(row)
+            except Exception:  # pragma: no cover
+                pass
+        elif self.category is not None:
+            category_val = str(self.category)
+
+        # Resolve optional interaction_class
+        class_val = None
+        if isinstance(self.interaction_class, Column):
+            cls_vals = cache.values(self.interaction_class, row)
+            class_val = str(cls_vals[0]) if cls_vals else None
+        elif callable(self.interaction_class):
+            try:
+                class_val = self.interaction_class(row)
+            except Exception:  # pragma: no cover
+                pass
+        elif self.interaction_class is not None:
+            class_val = str(self.interaction_class)
+
+        identifiers = self.identifiers.build(row, cache) if self.identifiers else None
+        annotations = self.annotations.build(row, cache) if self.annotations else None
+
+        return SilverRelation(
+            subject=subject_entity,
+            predicate=predicate_val,
+            object=object_entity,
+            category=category_val,
+            interaction_class=class_val,
+            identifiers=identifiers if identifiers else None,
+            annotations=annotations if annotations else None,
+        )
+
+
 __all__ = [
     "AssociationBuilder",
     "AssociationsBuilder",
@@ -1331,4 +1443,5 @@ def _allows_empty_identifiers(self, entity_type: Any) -> bool:
     "Member",
     "MembershipBuilder",
     "MembersFromList",
+    "RelationBuilder",
 ]