adopt enactment_index module from Legislice

authorityspoke/facts.py

@@ -106,7 +106,7 @@ def __init__(
     def wrapped_string(self):
         text = super().wrapped_string
         if self.standard_of_proof:
-            text += "\n" + indented("by the STANDARD {self.standard_of_proof}")
+            text += "\n" + indented(f"by the STANDARD {self.standard_of_proof}")
         return text
 
     def __str__(self):

authorityspoke/io/enactment_index.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from collections import OrderedDict
+from copy import deepcopy
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
+
+from legislice.enactments import RawEnactment
+
+RawPredicate = Dict[str, Union[str, bool]]
+RawFactor = Dict[str, Union[RawPredicate, Sequence[Any], str, bool]]
+
+
+class EnactmentIndex(OrderedDict):
+    """Index of cross-referenced objects, keyed to phrases that reference them."""
+
+    def insert_by_name(self, obj: Dict) -> None:
+        """Add record to dict, using value of record's "name" field as the dict key."""
+        self[obj["name"]] = obj.copy()
+        self[obj["name"]].pop("name")
+        return None
+
+    def get_by_name(self, name: str) -> Dict:
+        """
+        Convert retrieved record so name is a field rather than the key for the whole record.
+        :param name:
+            the name of the key where the record can be found in the Mentioned dict.
+        :returns:
+            the value stored at the key "name", plus a name field.
+        """
+        value = {"name": name}
+        value.update(self[name])
+        return value
+
+    def __repr__(self):
+        return f"EnactmentIndex({repr(dict(self))})"
+
+    def enactment_has_anchor(
+        self, enactment_name: str, anchor: Dict[str, Union[str, int]]
+    ) -> bool:
+        anchors_for_selected_element = self[enactment_name].get("anchors") or []
+        return any(
+            existing_anchor == anchor
+            for existing_anchor in anchors_for_selected_element
+        )
+
+    def add_anchor_for_enactment(
+        self, enactment_name: str, anchor: Dict[str, Union[str, int]]
+    ) -> None:
+        anchors_for_selected_element = self[enactment_name].get("anchors") or []
+        if not self.enactment_has_anchor(enactment_name, anchor):
+            anchors_for_selected_element.append(anchor)
+        self[enactment_name]["anchors"] = anchors_for_selected_element
+
+    def __add__(self, other: EnactmentIndex) -> EnactmentIndex:
+        new_index = deepcopy(self)
+        for key in other.keys():
+            other_dict = other.get_by_name(key)
+            new_index.index_enactment(other_dict)
+        return new_index
+
+    def index_enactment(self, obj: RawEnactment) -> Union[str, RawEnactment]:
+        r"""
+        Update index of mentioned Factors with 'obj', if obj is named.
+        If there is already an entry in the mentioned index with the same name
+        as obj, the old entry won't be replaced. But if any additional text
+        anchors are present in the new obj, the anchors will be added.
+        If obj has a name, it will be collapsed to a name reference.
+        :param obj:
+            data from JSON to be loaded as a :class:`.Enactment`
+        """
+        if obj.get("name"):
+            if obj["name"] in self:
+                if obj.get("anchors"):
+                    for anchor in obj["anchors"]:
+                        self.add_anchor_for_enactment(
+                            enactment_name=obj["name"], anchor=anchor
+                        )
+            else:
+                self.insert_by_name(obj)
+            obj = obj["name"]
+        return obj
+
+
+def create_name_for_enactment(obj: RawEnactment) -> str:
+    name: str = obj["node"]
+    if obj.get("start_date"):
+        name += f'@{obj["start_date"]}'
+
+    for field_name in ["start", "end", "prefix", "exact", "suffix"]:
+        if obj.get(field_name):
+            name += f':{field_name}="{obj[field_name]}"'
+    return name
+
+
+def ensure_enactment_has_name(obj: RawEnactment) -> RawEnactment:
+
+    if not obj.get("name"):
+        new_name = create_name_for_enactment(obj)
+        if new_name:
+            obj["name"] = new_name
+    return obj
+
+
+def collect_enactments(
+    obj: Union[RawFactor, List[Union[RawFactor, str]]],
+    mentioned: Optional[EnactmentIndex] = None,
+    keys_to_ignore: Sequence[str] = ("predicate", "anchors", "children"),
+) -> Tuple[RawFactor, EnactmentIndex]:
+    """
+    Make a dict of all nested objects labeled by name, creating names if needed.
+    To be used during loading to expand name references to full objects.
+    """
+    mentioned = mentioned or EnactmentIndex()
+    if isinstance(obj, List):
+        new_list = []
+        for item in obj:
+            new_item, new_mentioned = collect_enactments(item, mentioned)
+            mentioned.update(new_mentioned)
+            new_list.append(new_item)
+        obj = new_list
+    if isinstance(obj, Dict):
+        new_dict = {}
+        for key, value in obj.items():
+            if key not in keys_to_ignore and isinstance(value, (Dict, List)):
+                new_value, new_mentioned = collect_enactments(value, mentioned)
+                mentioned.update(new_mentioned)
+                new_dict[key] = new_value
+            else:
+                new_dict[key] = value
+
+        if new_dict.get("node") or (new_dict.get("name") in mentioned.keys()):
+            new_dict = ensure_enactment_has_name(new_dict)
+            new_dict = mentioned.index_enactment(new_dict)
+        obj = new_dict
+    return obj, mentioned

authorityspoke/io/loaders.py

@@ -9,7 +9,7 @@
 from typing import Any, Dict, List, Iterator, Optional, Tuple, Union
 
 from legislice.download import Client
-from legislice.name_index import EnactmentIndex
+
 
 from authorityspoke.decisions import Decision
 from authorityspoke.holdings import Holding

authorityspoke/io/readers.py

@@ -10,15 +10,12 @@
 from typing import Dict, List, Optional, Tuple, Type, Union
 
 from anchorpoint.textselectors import TextQuoteSelector
-from legislice import Enactment
 from legislice.download import Client
-from legislice.name_index import EnactmentIndex, collect_enactments
 from nettlesome.entities import Entity
-from nettlesome.predicates import Predicate
+from nettlesome.factors import Factor
 
 from authorityspoke.decisions import Decision
 from authorityspoke.evidence import Exhibit, Evidence
-from nettlesome.factors import Factor
 from authorityspoke.facts import Fact
 from authorityspoke.holdings import Holding
 from authorityspoke.opinions import AnchoredHoldings
@@ -36,6 +33,7 @@
     RawSelector,
 )
 from authorityspoke.io.name_index import index_names, Mentioned
+from authorityspoke.io.enactment_index import collect_enactments
 
 FACTOR_SUBCLASSES = {
     class_obj.__name__: class_obj

authorityspoke/io/schemas.py

@@ -11,7 +11,8 @@
 from anchorpoint.textselectors import TextQuoteSelector, TextPositionSelector
 from anchorpoint.schemas import SelectorSchema
 from legislice import Enactment
-from legislice.schemas import EnactmentSchema
+from legislice.schemas import EnactmentSchema as LegisliceSchema
+from legislice.schemas import enactment_needs_api_update
 from nettlesome.entities import Entity
 from nettlesome.predicates import Predicate
 from nettlesome.quantities import Comparison, QuantityRange, Quantity
@@ -21,6 +22,7 @@
 from nettlesome.factors import Factor
 from authorityspoke.facts import Fact
 from authorityspoke.holdings import Holding
+from authorityspoke.io.enactment_index import EnactmentIndex
 from authorityspoke.io.name_index import Mentioned
 from authorityspoke.io.name_index import RawFactor, RawPredicate
 from authorityspoke.io.nesting import nest_fields
@@ -164,6 +166,26 @@ def format_data_to_load(self, data: RawDecision, **kwargs) -> RawDecision:
         return data
 
 
+class EnactmentSchema(LegisliceSchema):
+    def get_indexed_enactment(self, data, **kwargs):
+        """Replace data to load with any object with same name in "enactment_index"."""
+        if isinstance(data, str):
+            name_to_retrieve = data
+        elif data.get("name") and enactment_needs_api_update(data):
+            name_to_retrieve = data["name"]
+        else:
+            return data
+
+        mentioned = self.context.get("enactment_index") or EnactmentIndex()
+        return deepcopy(mentioned.get_by_name(name_to_retrieve))
+
+    @pre_load
+    def format_data_to_load(self, data, **kwargs):
+        """Prepare Enactment to load."""
+        data = self.get_indexed_enactment(data)
+        return super().format_data_to_load(data)
+
+
 def dump_quantity(obj: Predicate) -> Optional[Union[date, float, int, str]]:
     """Convert quantity to string if it's a pint ureg.Quantity object."""
     if not hasattr(obj, "quantity"):

authorityspoke/rules.py

@@ -419,7 +419,9 @@ def explanations_implication(
                     other.procedure, context
                 )
 
-    def implies(self, other, context: Optional[ContextRegister] = None) -> bool:
+    def implies(
+        self, other: Comparable, context: Optional[ContextRegister] = None
+    ) -> bool:
         r"""
         Test if ``self`` implies ``other`` if posited in valid and decided :class:`.Holding`\s.
 
@@ -443,11 +445,9 @@ def implies(self, other, context: Optional[ContextRegister] = None) -> bool:
                 f'"implies" test not supported between class {self.__class__} and class {other.__class__}.'
             )
         if not isinstance(other, self.__class__):
-            if hasattr(other, "implied_by"):
-                if context:
-                    context = context.reversed()
-                return other.implied_by(self, context=context)
-            return False
+            if context:
+                context = context.reversed()
+            return other.implied_by(self, context=context)
         return any(
             explanation is not None
             for explanation in self.explanations_implication(other, context)
@@ -555,7 +555,7 @@ def union(
             return self._union_with_rule(other, context=context)
         elif hasattr(other, "union") and hasattr(other, "rule"):
             return other.union(self, context=context.reversed())
-        raise TypeError
+        raise TypeError(f"Union operation not possible between Rule and {type(other)}.")
 
     def __or__(self, other: Rule) -> Optional[Rule]:
         r"""

requirements.txt

@@ -1,7 +1,7 @@
 anchorpoint>=0.4.4
 apispec[validation]~=4.3.0
 marshmallow>=3.10
-git+git://github.com/mscarey/legislice.git@12e342bfcf522188b7cdb9fd71608f9aa065ecec#egg=legislice
+git+git://github.com/mscarey/legislice.git@1033eb44ad2662e5eff71417adb3cf6b5e2048ca#egg=legislice
 git+git://github.com/mscarey/nettlesome.git@70162a82b1bbb3fbade1b7ccc1c50c52d2701d31#egg=nettlesome
 pint>=0.15
 python-dotenv