Refactors import statements across multiple modules for improved organization and consistency. Updates argument and predicate filtering functions to follow naming conventions. Enhances test files by ensuring compatibility with the original PredPatt implementation and improving readability. Additionally, minor formatting adjustments and code cleanups are applied throughout the codebase.

Author: aaronstevenwhite

decomp/__init__.py

@@ -1,14 +1,12 @@
-import os
 import importlib.resources
+import os
+from logging import DEBUG, basicConfig
 
-from logging import basicConfig, DEBUG
 
 # get the data directory using importlib.resources
 DATA_DIR = str(importlib.resources.files('decomp') / 'data')
 basicConfig(filename=os.path.join(DATA_DIR, 'build.log'),
  filemode='w',
  level=DEBUG)
 
-from .semantics.uds import UDSCorpus
-from .semantics.uds import NormalizedUDSAnnotation
-from .semantics.uds import RawUDSAnnotation
+from .semantics.uds import NormalizedUDSAnnotation, RawUDSAnnotation, UDSCorpus

decomp/corpus/corpus.py

@@ -1,10 +1,11 @@
 """Module for defining abstract graph corpus readers"""
 
 from abc import ABCMeta, abstractmethod
-
-from random import sample
+from collections.abc import Hashable, Iterator
 from logging import warning
-from typing import Hashable, TypeVar, Iterator, Generic, TypeAlias
+from random import sample
+from typing import Generic, TypeAlias, TypeVar
+
 
 InGraph = TypeVar('InGraph') # the input graph type
 OutGraph = TypeVar('OutGraph') # the output graph type
@@ -60,19 +61,17 @@ def _graphbuilder(self,
 
  @property
  def graphs(self) -> dict[Hashable, OutGraph]:
- """the graphs in corpus"""
+ """The graphs in corpus"""
  return self._graphs
 
  @property
  def graphids(self) -> list[Hashable]:
  """The graph ids in corpus"""
-
  return list(self._graphs)
 
  @property
  def ngraphs(self) -> int:
  """Number of graphs in corpus"""
-
  return len(self._graphs)
 
  def sample(self, k: int) -> dict[Hashable, OutGraph]:
@@ -83,6 +82,5 @@ def sample(self, k: int) -> dict[Hashable, OutGraph]:
  k
  the number of graphs to sample
  """
- 
  sampled_keys = sample(list(self._graphs.keys()), k=k)
  return {tid: self._graphs[tid] for tid in sampled_keys}

decomp/graph/__init__.py

@@ -1,6 +1,7 @@
 """Module for converting between NetworkX and RDFLib graphs"""
 
-from .rdf import RDFConverter
 from .nx import NXConverter
+from .rdf import RDFConverter
+
 
-__all__ = ['RDFConverter', 'NXConverter']
+__all__ = ['NXConverter', 'RDFConverter']

decomp/graph/nx.py

@@ -1,7 +1,7 @@
 """Module for converting from networkx to RDF"""
 
-from networkx import DiGraph, to_dict_of_dicts
-from rdflib import Graph, URIRef, Literal
+from networkx import DiGraph
+from rdflib import Graph
 
 
 class NXConverter:
@@ -26,7 +26,6 @@ def rdf_to_networkx(cls, rdfgraph: Graph) -> DiGraph:
  rdfgraph
  the RDFLib graph to convert
  """
-
  converter = cls(rdfgraph)
 
  raise NotImplementedError

decomp/graph/rdf.py

@@ -1,8 +1,9 @@
 """Module for converting from networkx to RDF"""
 
 from typing import Any
+
 from networkx import DiGraph, to_dict_of_dicts
-from rdflib import Graph, URIRef, Literal
+from rdflib import Graph, Literal, URIRef
 
 
 class RDFConverter:
@@ -35,7 +36,6 @@ def networkx_to_rdf(cls, nxgraph: DiGraph) -> Graph:
  nxgraph
  the NetworkX graph to convert
  """
-
  converter = cls(nxgraph)
 
  nxdict = to_dict_of_dicts(nxgraph)
@@ -50,44 +50,44 @@ def networkx_to_rdf(cls, nxgraph: DiGraph) -> Graph:
 
  def _add_node_attributes(self, nodeid: str) -> None:
  self._construct_node(nodeid)
- 
+
  self._add_attributes(nodeid,
  list(self.nxgraph.nodes[nodeid].items()))
 
- 
+
  def _add_edge_attributes(self, nodeid1: str, nodeid2: str) -> None:
  edgeid = self._construct_edge(nodeid1, nodeid2)
  edgetup = (nodeid1, nodeid2)
- 
+
  self._add_attributes(edgeid,
  list(self.nxgraph.edges[edgetup].items()))
- 
+
 
  def _add_attributes(self, nid: str, attributes: list[tuple[str, Any]]) -> None:
  triples = []
- 
+
  for attrid1, attrs1 in attributes:
  if not isinstance(attrs1, dict):
  if isinstance(attrs1, list) or isinstance(attrs1, tuple):
  errmsg = 'Cannot convert list- or tuple-valued' +\
  ' attributes to RDF'
  raise ValueError(errmsg)
- 
+
  triples += self._construct_property(nid,
  attrid1,
  attrs1)
 
- else: 
+ else:
  for attrid2, attrs2 in attrs1.items():
  triples += self._construct_property(nid,
  attrid2,
  attrs2,
  attrid1)
 
  for t in triples:
- self.rdfgraph.add(t) 
- 
- def _construct_node(self, nodeid: str) -> None: 
+ self.rdfgraph.add(t)
+
+ def _construct_node(self, nodeid: str) -> None:
  if nodeid not in self.nodes:
  self.nodes[nodeid] = URIRef(nodeid)
 
@@ -102,7 +102,7 @@ def _construct_edge(self, nodeid1: str, nodeid2: str) -> str:
  triple = (node1, self.nodes[edgeid], node2)
 
  self.rdfgraph.add(triple)
- 
+
  return edgeid
 
  else:
@@ -113,15 +113,15 @@ def _construct_property(self, nodeid: str, propid: str, val: Any,
 
  c = self.__class__
  triples: list[tuple[URIRef, URIRef, URIRef | Literal]]
- 
+
  if isinstance(val, dict) and subspaceid is not None:
  # We currently do not support querying on raw UDS
  # annotations, all of which have dict-valued 'value'
  # and 'confidence' fields.
  if isinstance(val['value'], dict) or isinstance(val['confidence'], dict):
  raise TypeError('Attempted query of graph with raw properties. Querying '\
  'graphs with raw properties is prohibited.')
- triples = c._construct_subspace(subspaceid, propid) 
+ triples = c._construct_subspace(subspaceid, propid)
  triples += [(self.nodes[nodeid],
  c.PROPERTIES[propid],
  Literal(val['value'])),
@@ -132,26 +132,26 @@ def _construct_property(self, nodeid: str, propid: str, val: Any,
  elif propid in ['domain', 'type']:
  if val not in c.VALUES:
  c.VALUES[val] = URIRef(val)
- 
+
  triples = [(self.nodes[nodeid],
  c.PROPERTIES[propid],
  c.VALUES[val])]
 
  else:
  if propid not in c.PROPERTIES:
  c.PROPERTIES[propid] = URIRef(propid)
- 
+
  triples = [(self.nodes[nodeid],
  c.PROPERTIES[propid],
- Literal(val))] 
- 
+ Literal(val))]
+
  return triples
 
  @classmethod
  def _construct_subspace(cls, subspaceid: str, propid: str) -> list[tuple[URIRef, URIRef, URIRef | Literal]]:
  if subspaceid not in cls.SUBSPACES:
  cls.SUBSPACES[subspaceid] = URIRef(subspaceid)
- 
+
  if propid not in cls.PROPERTIES:
  cls.PROPERTIES[propid] = URIRef(propid)
  cls.PROPERTIES[propid+'-confidence'] = URIRef(propid+'-confidence')
@@ -161,7 +161,7 @@ def _construct_subspace(cls, subspaceid: str, propid: str) -> list[tuple[URIRef,
  cls.SUBSPACES[subspaceid]),
  (cls.PROPERTIES[propid+'-confidence'],
  cls.PROPERTIES['subspace'],
- cls.SUBSPACES[subspaceid]), 
+ cls.SUBSPACES[subspaceid]),
  (cls.PROPERTIES[propid],
  cls.PROPERTIES['confidence'],
  cls.PROPERTIES[propid+'-confidence'])]

decomp/semantics/predpatt/core/argument.py

@@ -57,11 +57,7 @@ def __init__(
  self,
  root: Token,
  ud: Any = dep_v1,
- rules: list[Any] = [], # NOTE: Mutable default to match original
- # WARNING: This mutable default is INTENTIONAL and REQUIRED
- # for exact compatibility with original PredPatt.
- # Instances share the same list when rules is not provided.
- # DO NOT CHANGE to None - this would break compatibility!
+ rules: list[Any] | None = None,
  share: bool = False
  ) -> None:
  """Initialize an Argument.
@@ -80,7 +76,7 @@ def __init__(
  """
  # maintain exact initialization order as original
  self.root = root
- self.rules = rules  # intentionally using mutable default
+ self.rules = rules if rules is not None else []
  self.position = root.position
  self.ud = ud
  self.tokens: list[Token] = []

decomp/semantics/predpatt/core/predicate.py

@@ -260,14 +260,14 @@ def is_broken(self) -> bool | None:
  return True
  return None
 
- def _format_predicate(self, name: dict[Any, str], C: Any = no_color) -> str:
+ def _format_predicate(self, name: dict[Any, str], c: Any = no_color) -> str:
  """Format predicate with argument placeholders.
 
  Parameters
  ----------
  name : dict[Any, str]
  Mapping from arguments to their names.
- C : callable, optional
+ c : callable, optional
  Color function for formatting.
 
  Returns
@@ -276,7 +276,7 @@ def _format_predicate(self, name: dict[Any, str], C: Any = no_color) -> str:
  Formatted predicate string.
  """
  # collect tokens and arguments
- X = sort_by_position(self.tokens + self.arguments)
+ x = sort_by_position(self.tokens + self.arguments)
 
  if self.type == POSS:
  # possessive format: "?a 's ?b"
@@ -295,7 +295,7 @@ def _format_predicate(self, name: dict[Any, str], C: Any = no_color) -> str:
  if gov_arg:
  # format: gov_arg is/are other_tokens_and_args
  rest = []
- for item in X:
+ for item in x:
  if item == gov_arg:
  continue
  if item in self.arguments:
@@ -306,7 +306,7 @@ def _format_predicate(self, name: dict[Any, str], C: Any = no_color) -> str:
  return f'{name[gov_arg]} is/are {rest_str}'
  else:
  # fallback if no governor argument found
- return ' '.join(name[item] if item in self.arguments else item.text for item in X)
+ return ' '.join(name[item] if item in self.arguments else item.text for item in x)
 
  else:
  # normal predicate or xcomp special case
@@ -317,7 +317,7 @@ def _format_predicate(self, name: dict[Any, str], C: Any = no_color) -> str:
  self.root.tag not in {postag.VERB, postag.ADJ}):
  # add is/are after first argument
  first_arg_added = False
- for item in X:
+ for item in x:
  if item in self.arguments:
  result.append(name[item])
  if not first_arg_added:
@@ -327,7 +327,7 @@ def _format_predicate(self, name: dict[Any, str], C: Any = no_color) -> str:
  result.append(item.text)
  else:
  # normal formatting
- for item in X:
+ for item in x:
  if item in self.arguments:
  result.append(name[item])
  else:
@@ -338,7 +338,7 @@ def _format_predicate(self, name: dict[Any, str], C: Any = no_color) -> str:
  def format(
  self,
  track_rule: bool = False,
- C: Any = no_color,
+ c: Any = no_color,
  indent: str = '\t'
  ) -> str:
  """Format predicate with arguments for display.
@@ -347,7 +347,7 @@ def format(
  ----------
  track_rule : bool, optional
  Whether to include rule tracking information.
- C : callable, optional
+ c : callable, optional
  Color function for formatting.
  indent : str, optional
  Indentation string to use.
@@ -362,24 +362,24 @@ def format(
  verbose = ''
  if track_rule:
  rules_str = ','.join(sorted(map(str, self.rules)))
- verbose = ' ' + C(f'[{self.root.text}-{self.root.gov_rel},{rules_str}]', 'magenta')
+ verbose = ' ' + c(f'[{self.root.text}-{self.root.gov_rel},{rules_str}]', 'magenta')
 
- pred_str = self._format_predicate(argument_names(self.arguments), C)
+ pred_str = self._format_predicate(argument_names(self.arguments), c)
  lines.append(f'{indent}{pred_str}{verbose}')
 
  # format arguments
  name = argument_names(self.arguments)
  for arg in self.arguments:
  if (arg.isclausal() and arg.root.gov in self.tokens and
  self.type == NORMAL):
- s = C('SOMETHING', 'yellow') + ' := ' + arg.phrase()
+ s = c('SOMETHING', 'yellow') + ' := ' + arg.phrase()
  else:
- s = C(arg.phrase(), 'green')
+ s = c(arg.phrase(), 'green')
  rule = ''
  if track_rule:
  rules_str = ','.join(sorted(map(str, arg.rules)))
  rule = f',{rules_str}'
- verbose = C(f' [{arg.root.text}-{arg.root.gov_rel}{rule}]',
+ verbose = c(f' [{arg.root.text}-{arg.root.gov_rel}{rule}]',
  'magenta')
  else:
  verbose = ''