Merge branch 'master' into new-branch

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/ambv/black
-    rev: 22.3.0
+    rev: 23.9.1
     hooks:
       - id: black
         args: [--safe]
@@ -9,11 +9,11 @@ repos:
     hooks:
     -   id: seed-isort-config
   - repo: https://github.com/timothycrosley/isort
-    rev: 5.7.0
+    rev: 5.12.0
     hooks:
       - id: isort
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v3.4.0
+    rev: v4.4.0
     hooks:
       - id: check-ast
       - id: check-docstring-first
@@ -26,7 +26,7 @@ repos:
       - id: debug-statements
       - id: end-of-file-fixer
       - id: trailing-whitespace
-  - repo: https://github.com/PyCQA/flake8
-    rev: 3.8.4
+  - repo: https://github.com/pycqa/flake8
+    rev: 6.1.0
     hooks:
       - id: flake8

pycid/analyze/reasoning_patterns.py

@@ -41,7 +41,7 @@ def _path_is_effective(mb: MACIDBase, path: List[str], effective_set: Set[str])
 
 
 def _directed_effective_path_not_through_set_y(
-    mb: MACIDBase, start: str, finish: str, effective_set: Set[str], y: Set[str] = None
+    mb: MACIDBase, start: str, finish: str, effective_set: Set[str], y: Optional[Set[str]] = None
 ) -> bool:
     """Check whether a directed effective path exists that doesn't pass through any of the nodes in the set y."""
     if y is None:
@@ -125,7 +125,6 @@ def manipulation(macid: MACID, decision: str, effective_set: Set[str]) -> bool:
 
         for u in agent_utils:
             if _effective_dir_path_exists(macid, decision_b, u, effective_set):
-
                 for u_b in agent_b_utils:
                     if _directed_effective_path_not_through_set_y(macid, decision, u_b, effective_set, {decision_b}):
                         return True

pycid/core/causal_bayesian_network.py

@@ -36,7 +36,6 @@ def __init__(self, cbn: CausalBayesianNetwork, *args: Any, **kwargs: Any) -> Non
             self.domain: Dict[str, List[Outcome]] = {}
 
         def __setitem__(self, variable: str, relationship: Relationship) -> None:
-
             # Update the keys
             if variable in self.keys():
                 self.__delitem__(variable)
@@ -81,7 +80,7 @@ def to_tabular_cpd(self, variable: str, relationship: Relationship) -> TabularCP
             elif isinstance(relationship, Mapping):
                 return ConstantCPD(variable, relationship, self.cbn)
 
-    def __init__(self, edges: Iterable[Tuple[str, str]] = None, **kwargs: Any):
+    def __init__(self, edges: Optional[Iterable[Tuple[str, str]]] = None, **kwargs: Any):
         """Initialize a Causal Bayesian Network
 
         Parameters
@@ -134,7 +133,7 @@ def is_structural_causal_model(self) -> bool:
         return True
 
     def query(
-        self, query: Iterable[str], context: Dict[str, Outcome], intervention: Dict[str, Outcome] = None
+        self, query: Iterable[str], context: Dict[str, Outcome], intervention: Optional[Dict[str, Outcome]] = None
     ) -> BeliefPropagation:
         """Return P(query|context, do(intervention))*P(context | do(intervention)).
 
@@ -199,7 +198,7 @@ def expected_value(
         self,
         variables: Iterable[str],
         context: Dict[str, Outcome],
-        intervention: Dict[str, Outcome] = None,
+        intervention: Optional[Dict[str, Outcome]] = None,
     ) -> List[float]:
         """Compute the expected value of a real-valued variable for a given context,
         under an optional intervention
@@ -271,9 +270,9 @@ def _get_label(self, node: str) -> str:
 
     def draw(
         self,
-        node_color: Callable[[str], Union[str, np.ndarray]] = None,
-        node_shape: Callable[[str], str] = None,
-        node_label: Callable[[str], str] = None,
+        node_color: Optional[Callable[[str], Union[str, np.ndarray]]] = None,
+        node_shape: Optional[Callable[[str], str]] = None,
+        node_label: Optional[Callable[[str], str]] = None,
         layout: Optional[Callable[[Any], Dict[Any, Any]]] = None,
     ) -> None:
         """

pycid/core/cpd.py

@@ -49,7 +49,7 @@ def __init__(
         cbn: CausalBayesianNetwork,
         domain: Optional[Sequence[Outcome]] = None,
         state_names: Optional[Mapping[str, List]] = None,
-        label: str = None,
+        label: Optional[str] = None,
     ) -> None:
         """Initialize StochasticFunctionCPD with a variable name and a stochastic function.
 
@@ -130,7 +130,7 @@ def stochastic_function(self, **pv: Outcome) -> Mapping[Outcome, float]:
         else:
             return {ret: 1}
 
-    def compute_label(self, function: Callable = None) -> str:
+    def compute_label(self, function: Optional[Callable] = None) -> str:
         """Try to generate a string that succinctly describes the relationship"""
         function = function if function is not None else self.func
         if hasattr(function, "__name__") and function.__name__ != "<lambda>":
@@ -224,7 +224,7 @@ def __init__(
         variable: str,
         dictionary: Mapping,
         cbn: CausalBayesianNetwork,
-        domain: Sequence[Outcome] = None,
+        domain: Optional[Sequence[Outcome]] = None,
         label: Optional[str] = None,
     ):
         super().__init__(variable, lambda **pv: dictionary, cbn, domain=domain, label=label or str(dictionary))
@@ -267,7 +267,7 @@ def discrete_uniform(domain: List[Outcome]) -> Dict[Outcome, float]:
 
 
 def noisy_copy(
-    value: Outcome, probability: float = 0.9, domain: List[Outcome] = None
+    value: Outcome, probability: float = 0.9, domain: Optional[List[Outcome]] = None
 ) -> Dict[Outcome, Optional[float]]:
     """specify a variable's CPD as copying the value of some other variable with a certain probability."""
     dist = dict.fromkeys(domain) if domain else {}

pycid/core/get_paths.py

@@ -1,4 +1,4 @@
-from typing import Callable, Iterable, Iterator, List, Sequence, Set, Tuple
+from typing import Callable, Iterable, Iterator, List, Optional, Sequence, Set, Tuple
 
 import networkx as nx
 
@@ -184,7 +184,7 @@ def _get_path_edges(cbn: CausalBayesianNetwork, path: Sequence[str]) -> List[Tup
     return structure
 
 
-def is_active_path(cbn: CausalBayesianNetwork, path: Sequence[str], observed: Set[str] = None) -> bool:
+def is_active_path(cbn: CausalBayesianNetwork, path: Sequence[str], observed: Optional[Set[str]] = None) -> bool:
     """
     Check if a specifc path remains active given the 'observed' set of variables.
     """
@@ -213,7 +213,7 @@ def is_active_path(cbn: CausalBayesianNetwork, path: Sequence[str], observed: Se
 
 
 def is_active_indirect_frontdoor_trail(
-    cbn: CausalBayesianNetwork, start_node: str, end_node: str, observed: Set[str] = None
+    cbn: CausalBayesianNetwork, start_node: str, end_node: str, observed: Optional[Set[str]] = None
 ) -> bool:
     """
     checks whether an active indirect frontdoor path exists given the 'observed' set of variables.
@@ -240,7 +240,7 @@ def is_active_indirect_frontdoor_trail(
 
 
 def is_active_backdoor_trail(
-    cbn: CausalBayesianNetwork, start_node: str, end_node: str, observed: Set[str] = None
+    cbn: CausalBayesianNetwork, start_node: str, end_node: str, observed: Optional[Set[str]] = None
 ) -> bool:
     """
     Returns true if there is a backdoor path that's active given the 'observed' set of nodes.

pycid/core/macid_base.py

@@ -67,9 +67,9 @@ def to_tabular_cpd(self, variable: str, relationship: Union[Relationship, Sequen
 
     def __init__(
         self,
-        edges: Iterable[Tuple[str, str]] = None,
-        agent_decisions: Mapping[AgentLabel, List[str]] = None,
-        agent_utilities: Mapping[AgentLabel, List[str]] = None,
+        edges: Optional[Iterable[Tuple[str, str]]] = None,
+        agent_decisions: Optional[Mapping[AgentLabel, List[str]]] = None,
+        agent_utilities: Optional[Mapping[AgentLabel, List[str]]] = None,
         **kwargs: Any,
     ):
         """Initialize a new MACIDBase instance.
@@ -142,7 +142,7 @@ def add_cpds(self, *cpds: TabularCPD, **relationships: Union[Relationship, List[
         super().add_cpds(*cpds, **relationships)
 
     def query(
-        self, query: Iterable[str], context: Dict[str, Outcome], intervention: Dict[str, Outcome] = None
+        self, query: Iterable[str], context: Dict[str, Outcome], intervention: Optional[Dict[str, Outcome]] = None
     ) -> BeliefPropagation:
         """Return P(query|context, do(intervention))*P(context | do(intervention)).
 
@@ -184,7 +184,7 @@ def query(
         return super().query(query, context, intervention)
 
     def expected_utility(
-        self, context: Dict[str, Outcome], intervention: Dict[str, Outcome] = None, agent: AgentLabel = 0
+        self, context: Dict[str, Outcome], intervention: Optional[Dict[str, Outcome]] = None, agent: AgentLabel = 0
     ) -> float:
         """Compute the expected utility of an agent for a given context and optional intervention
 
@@ -286,15 +286,15 @@ def pure_decision_rules(self, decision: str) -> Iterator[StochasticFunctionCPD]:
 
         # We begin by representing each possible decision rule as a tuple of outcomes, with
         # one element for each possible decision context
-        number_of_decision_contexts = int(np.product(parent_cardinalities))
+        number_of_decision_contexts = int(np.prod(parent_cardinalities))
         functions_as_tuples = itertools.product(domain, repeat=number_of_decision_contexts)
 
         def arg2idx(pv: Dict[str, Outcome]) -> int:
             """Convert a decision context into an index for the function list"""
             idx = 0
             for i, parent in enumerate(parents):
                 name_to_no: Dict[Outcome, int] = self.get_cpds(parent).name_to_no[parent]
-                idx += name_to_no[pv[parent]] * int(np.product(parent_cardinalities[:i]))
+                idx += name_to_no[pv[parent]] * int(np.prod(parent_cardinalities[:i]))
             assert 0 <= idx <= number_of_decision_contexts
             return idx
 
@@ -327,8 +327,8 @@ def optimal_pure_policies(
         decisions = set(decisions)
         try:
             (agent,) = {self.decision_agent[d] for d in decisions}
-        except ValueError:
-            raise ValueError("Decisions not all from the same agent")
+        except ValueError as e:
+            raise ValueError("Decisions not all from the same agent") from e
 
         macid = self.copy()
         for d in macid.decisions:
@@ -361,8 +361,8 @@ def impute_random_decision(self, d: str) -> None:
         """Impute a random policy to the given decision node"""
         try:
             domain = self.model.domain[d]
-        except KeyError:
-            raise ValueError(f"can't figure out domain for {d}, did you forget to specify DecisionDomain?")
+        except KeyError as e:
+            raise ValueError(f"can't figure out domain for {d}, did you forget to specify " f"DecisionDomain?") from e
         else:
             self.model[d] = StochasticFunctionCPD(
                 d, lambda **pv: {outcome: 1 / len(domain) for outcome in domain}, self, domain, label="random_decision"
@@ -436,7 +436,7 @@ def _get_color(self, node: str) -> Union[np.ndarray, str]:
         Assign a unique colour to each new agent's decision and utility nodes
         """
         agents = list(self.agents)
-        colors = cm.rainbow(np.linspace(0, 1, len(agents)))
+        colors = cm.rainbow(np.linspace(0, 1, len(agents)))  # type: ignore
         try:
             agent = self.decision_agent[node]
         except KeyError:

pycid/core/relevance_graph.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import itertools
-from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Sequence
+from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Sequence
 
 import matplotlib.cm as cm
 import matplotlib.pyplot as plt
@@ -21,7 +21,7 @@ class RelevanceGraph(nx.DiGraph):
     - an edge D -> D' exists iff D' is r-reachable from D (ie D strategically or probabilistically relies on D')
     """
 
-    def __init__(self, cid: MACIDBase, decisions: Iterable[str] = None):
+    def __init__(self, cid: MACIDBase, decisions: Optional[Iterable[str]] = None):
         super().__init__()
         if decisions is None:
             decisions = cid.decisions
@@ -47,7 +47,7 @@ def get_sccs(self) -> List[set]:
 
     def _set_color_scc(self, node: str, sccs: Sequence[Any]) -> np.ndarray:
         "Assign a unique color to the set of nodes in each SCC."
-        colors = cm.rainbow(np.linspace(0, 1, len(sccs)))
+        colors = cm.rainbow(np.linspace(0, 1, len(sccs)))  # type: ignore
         scc_index = 0
         for idx, scc in enumerate(sccs):
             if node in scc:

pycid/examples/story_cids.py

@@ -3,7 +3,6 @@
 
 
 def get_introduced_bias() -> CID:
-
     cid = CID(
         [
             ("A", "X"),  # defining the graph's nodes and edges

pycid/examples/story_macids.py

@@ -238,8 +238,8 @@ def robot_warehouse() -> MACID:
             ("B", "R"),
             ("B", "D2"),
             ("D2", "R"),
-            ("D2", "O"),
-            ("O", "U1"),
+            ("D2", "Ob"),
+            ("Ob", "U1"),
             ("R", "U2"),
         ],
         agent_decisions={
@@ -258,8 +258,8 @@ def robot_warehouse() -> MACID:
         Q=lambda D1: noisy_copy(D1, domain=[0, 1]),
         B=lambda D1: noisy_copy(D1, probability=0.3, domain=[0, 1]),
         R=lambda B, D2: int(not B or D2),
-        O=lambda D2: noisy_copy(D2, probability=0.6, domain=[0, 1]),
-        U1=lambda Q, B, O: int(Q and not O) - int(B),
+        Ob=lambda D2: noisy_copy(D2, probability=0.6, domain=[0, 1]),
+        U1=lambda Q, B, Ob: int(Q and not Ob) - int(B),
         U2=lambda R: R,
     )
     return macid

pycid/export/gambit.py

@@ -177,7 +177,7 @@ def behavior_to_cpd(
     macid: MACIDBase,
     parents_to_infoset: Mapping[Tuple[Hashable, Tuple[Tuple[str, Any], ...]], pygambit.Infoset],
     behavior: pygambit.lib.libgambit.MixedStrategyProfile,
-    decisions_in_sg: Union[KeysView[str], Set[str]] = None,
+    decisions_in_sg: Optional[Union[KeysView[str], Set[str]]] = None,
 ) -> List[StochasticFunctionCPD]:
     """Convert a pygambit behavior strategy to list of CPDs for each decision node.
     Args:

pycid/random/random_cpd.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import contextlib
-from typing import Iterator, Mapping, Sequence
+from typing import Iterator, Mapping, Optional, Sequence
 
 import numpy as np
 
@@ -23,7 +23,9 @@ class RandomCPD:
     Sample a random CPD, with outcomes in the given domain
     """
 
-    def __init__(self, domain: Sequence[Outcome] = None, smoothness: float = 1.0, seed: int = None) -> None:
+    def __init__(
+        self, domain: Optional[Sequence[Outcome]] = None, smoothness: float = 1.0, seed: Optional[int] = None
+    ) -> None:
         """
         Parameters
         ----------

pycid/random/random_macidbase.py

@@ -41,7 +41,6 @@ def random_macidbase(
         )
 
     for _ in range(max_resampling_attempts):
-
         dag = random_dag(number_of_nodes=number_of_nodes, edge_density=edge_density, max_in_degree=max_in_degree)
 
         # assign utility nodes to each agent based on the barren nodes in the random dag