Merge pull request #34 from seldon-code/micromamba

Micromamba

.github/workflows/ci_micromamba.yml

@@ -0,0 +1,38 @@
+name: CI-Micromamba
+
+on: [push, pull_request]
+
+jobs:
+  test:
+    runs-on: ${{ matrix.os }}
+    name: test (${{ matrix.os }})
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, macos-latest]
+        python-version: ['3.10', '3.11']
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Micromamba
+      uses: mamba-org/setup-micromamba@v1
+      with:
+        micromamba-version: latest
+        environment-file: environment.yml
+
+    - name: Create or update environment with Micromamba
+      run: |
+        micromamba create -n hariplotterenv -f environment.yml --yes
+
+    - name: Install the package in editable mode
+      run: |
+        micromamba run -n hariplotterenv pip install -e .
+
+    - name: Install pytest in the environment
+      run: |
+        micromamba run -n hariplotterenv micromamba install pytest --yes
+
+    - name: Run tests with pytest
+      run: |
+        micromamba run -n hariplotterenv pytest

environment.yml

@@ -1,22 +1,9 @@
-# $ micromamba create -f environment.yml
-# $ micromamba activate hariplotterenv
-#
-
 name: hariplotterenv
 channels:
   - conda-forge
-# requirements:
-#   build:
-#     - {{ compiler('c') }}
-#     - {{ compiler('c++') }}
 dependencies:
+  - python>=3.10
   - meson
-  - numpy
-  - pytest
-  - coverage
-  - matplotlib
-  - networkx
+  - pip
   - pip:
-    - rich
-    - rich-pixels
-    - Pillow
+      - -r requirements.txt

hari_plotter/dynamics.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
+import math
+from typing import Any, Union
+
+import matplotlib.pyplot as plt
+import numpy as np
 
 from .graph import Graph
 from .lazy_graph import LazyGraph
-import numpy as np
-import matplotlib.pyplot as plt
-import math
 
 
 class Dynamics:
@@ -17,8 +18,8 @@ class Dynamics:
     networks and opinions, and provides a unified interface to access and manipulate each LazyHariGraph.
 
     Attributes:
-        lazy_hari_graphs (List[LazyHariGraph]): A list of LazyHariGraph objects.
-        groups (List[List[int]]): A list where each element is a list of indices representing a group.
+        lazy_hari_graphs (list[LazyHariGraph]): A list of LazyHariGraph objects.
+        groups (list[list[int]]): A list where each element is a list of indices representing a group.
 
     Methods:
         initialized: Returns indices of initialized LazyHariGraph instances.
@@ -36,19 +37,19 @@ class Dynamics:
     """
 
     def __init__(self):
-        self.lazy_hari_graphs: List[LazyGraph] = []
-        self.groups: List[List[int]] = []
+        self.lazy_hari_graphs: list[LazyGraph] = []
+        self.groups: list[list[int]] = []
 
     @property
-    def initialized(self) -> List[int]:
+    def initialized(self) -> list[int]:
         """
-        List of indices of initialized LazyHariGraph instances.
+        list of indices of initialized LazyHariGraph instances.
 
         Iterates over each LazyHariGraph in the list `lazy_hari_graphs` and checks if it's initialized.
         Returns a list containing the indices of all the LazyHariGraphs that are initialized.
 
         Returns:
-            List[int]: Indices of all initialized LazyHariGraph instances.
+            list[int]: Indices of all initialized LazyHariGraph instances.
         """
         return [index for index, graph in enumerate(self.lazy_hari_graphs) if graph.is_initialized]
 
@@ -64,14 +65,14 @@ def uncluster(self):
             graph.uninitialize()
 
     @classmethod
-    def read_network(cls, network_files: Union[str, List[str]], opinion_files: List[str], load_request: Dict[str, Any] = {}) -> Dynamics:
+    def read_network(cls, network_files: Union[str, list[str]], opinion_files: list[str], load_request: dict[str, Any] = {}) -> Dynamics:
         """
         Reads a list of network files and a list of opinion files to create LazyHariGraph objects
         and appends them to the lazy_hari_graphs list of a HariDynamics instance.
 
         Parameters:
-            network_files (Union[str, List[str]]): Either a single path or a list of paths to the network files.
-            opinion_files (List[str]): A list of paths to the opinion files.
+            network_files (Union[str, list[str]]): Either a single path or a list of paths to the network files.
+            opinion_files (list[str]): A list of paths to the opinion files.
 
         Returns:
             HariDynamics: An instance of HariDynamics with lazy_hari_graphs populated.
@@ -107,14 +108,14 @@ def read_network(cls, network_files: Union[str, List[str]], opinion_files: List[
         return dynamics_instance
 
     @classmethod
-    def read_json(cls, json_files: Union[str, List[str]], load_request: Dict[str, Any] = {}) -> Dynamics:
+    def read_json(cls, json_files: Union[str, list[str]], load_request: dict[str, Any] = {}) -> Dynamics:
         """
         Reads a list of network files and a list of opinion files to create LazyHariGraph objects
         and appends them to the lazy_hari_graphs list of a HariDynamics instance.
 
         Parameters:
-            network_files (Union[str, List[str]]): Either a single path or a list of paths to the network files.
-            opinion_files (List[str]): A list of paths to the opinion files.
+            network_files (Union[str, list[str]]): Either a single path or a list of paths to the network files.
+            opinion_files (list[str]): A list of paths to the opinion files.
 
         Returns:
             HariDynamics: An instance of HariDynamics with lazy_hari_graphs populated.
@@ -251,23 +252,23 @@ def group(self, num_intervals: int, interval_size: int = 1, offset: int = 0):
             self.groups.append(
                 list(range(start_index, min(end_index, total_length))))
 
-    def get_grouped_graphs(self) -> List[List[LazyGraph]]:
+    def get_grouped_graphs(self) -> list[list[LazyGraph]]:
         """
         Retrieves grouped LazyHariGraph objects based on the indices in self.groups.
 
         Returns:
-            List[List[LazyHariGraph]]: List of lists where each sublist contains LazyHariGraph objects.
+            list[list[LazyHariGraph]]: list of lists where each sublist contains LazyHariGraph objects.
         """
         return [[self.lazy_hari_graphs[i] for i in group_indices]
                 for group_indices in self.groups]
 
-    def merge_nodes_by_mapping(self, mapping: Tuple[int]):
+    def merge_nodes_by_mapping(self, mapping: tuple[int]):
         """
         Merge nodes in each LazyHariGraph based on the provided mapping.
         If a graph was already initialized, uninitialize it first.
 
         Parameters:
-            mapping (Tuple[int]): A dictionary representing how nodes should be merged.
+            mapping (tuple[int]): A dictionary representing how nodes should be merged.
         """
 
         for lazy_graph in self.lazy_hari_graphs:

hari_plotter/graph.py

@@ -3,10 +3,8 @@
 import json
 import os
 import random
-import re
-import warnings
 from itertools import combinations, permutations
-from typing import Dict, List, Optional, Set, Tuple, Type, Union
+from typing import Optional, Type, Union
 
 import networkx as nx
 import numpy as np
@@ -53,13 +51,13 @@ def set_gatherer(self, new_gatherer: Type[NodeEdgeGatherer]) -> None:
         """
         self.gatherer = new_gatherer(self)
 
-    def add_parameters_to_nodes(self, nodes: Optional[List[Tuple[int]]] = None) -> None:
+    def add_parameters_to_nodes(self, nodes: Optional[list[tuple[int]]] = None) -> None:
         """
         Adds or updates parameters for the specified nodes based on the current gatherer's criteria. If no nodes
         are specified, parameters are added or updated for all nodes in the graph.
 
         Parameters:
-            nodes (Optional[List[Tuple[int]]]): List of node identifiers to update. If None, updates all nodes.
+            nodes (Optional[list[tuple[int]]]): list of node identifiers to update. If None, updates all nodes.
         """
         nodes = nodes or list(self.nodes)
         parameters = self.gatherer.gather_everything()
@@ -185,13 +183,13 @@ def make_degroot_converging(self, seed: Optional[int] = None) -> None:
             for u, v in incoming_edges:
                 self.edges[u, v]['Influence'] /= total_influence
 
-    def mean_graph(self, images: List['Graph']) -> 'Graph':
+    def mean_graph(self, images: list['Graph']) -> 'Graph':
         """
         Calculates the mean graph from a list of HariGraph instances. The mean graph's nodes and edges have attributes
         that are the average of the corresponding attributes in the input graphs.
 
         Parameters:
-            images (List['HariGraph']): A list of HariGraph instances from which to calculate the mean graph.
+            images (list['HariGraph']): A list of HariGraph instances from which to calculate the mean graph.
 
         Returns:
             'HariGraph': A new HariGraph instance representing the mean of the input graphs.
@@ -314,9 +312,9 @@ def write_network(self, network_file: str, opinion_file: str, delimiter=','):
 
             # Iterate over the node mapping in sorted order of integer IDs
             for node, node_idx in sorted(node_mapping.items(), key=lambda x: x[1]):
-                # List of nodes influencing the current node
+                # list of nodes influencing the current node
                 neighbors = list(self.predecessors(node))
-                # List of influence weights from each neighbor
+                # list of influence weights from each neighbor
                 weights = [self[neighbor][node]['Influence']
                            for neighbor in neighbors]
                 # Use the mapping to get integer IDs for the neighbors
@@ -487,11 +485,11 @@ def by_deletion(cls, n: int, factor: float) -> Graph:
 
     @classmethod
     def strongly_connected_components(
-            cls, cluster_sizes: List[int], inter_cluster_edges: int, mean_opinion: float = 0.5, seed: int = None) -> Graph:
+            cls, cluster_sizes: list[int], inter_cluster_edges: int, mean_opinion: float = 0.5, seed: int = None) -> Graph:
         """
         Creates a HariGraph instance with multiple strongly connected components.
 
-        :param cluster_sizes: List[int], sizes of the clusters.
+        :param cluster_sizes: list[int], sizes of the clusters.
         :param inter_cluster_edges: int, number of edges between the components.
         :param mean_opinion: float, mean opinion of the graph.
         :param seed: int, random seed.
@@ -616,14 +614,14 @@ def dynamics_example_step(self, t: float):
 
     # ---- Merge Methods ----
 
-    def get_cluster_mapping(self) -> List[List[Tuple[int]]]:
+    def get_cluster_mapping(self) -> list[list[tuple[int]]]:
         """
         Generates a list of nodes in the unclustered graph to be clustered to get the current graph
         :return: A list representing the current clusters in the graph.
         """
         return sorted([sorted([(element,) for element in node]) for node in self.nodes])
 
-    def merge_nodes(self, i: Tuple[int], j: Tuple[int]):
+    def merge_nodes(self, i: tuple[int], j: tuple[int]):
         """
         Merges two nodes in the graph into a new node.
 
@@ -633,17 +631,17 @@ def merge_nodes(self, i: Tuple[int], j: Tuple[int]):
         and the old nodes are removed.
 
         Parameters:
-            i Tuple[int]: The identifier for the first node to merge.
-            j Tuple[int]: The identifier for the second node to merge.
+            i tuple[int]: The identifier for the first node to merge.
+            j tuple[int]: The identifier for the second node to merge.
         """
         self.gatherer.merge_nodes(i, j)
 
-    def merge_clusters(self, clusters: List[List[Tuple[int]]], labels: Union[List[str], None] = None, merge_remaining=False):
+    def merge_clusters(self, clusters: list[list[tuple[int]]], labels: Union[list[str], None] = None, merge_remaining=False):
         """
         Merges clusters of nodes in the graph into new nodes. Optionally merges the remaining nodes into an additional cluster.
 
         Parameters:
-            clusters (Union[List[Set[int]], Dict[int, int]]): A list where each element is a set containing
+            clusters (Union[list[Set[int]], dict[int, int]]): A list where each element is a set containing
                                     the IDs of the nodes in a cluster to be merged or a dictionary mapping old node IDs
                                     to new node IDs.
             merge_remaining (bool): If True, merge the nodes not included in clusters into an additional cluster. Default is False.
@@ -663,7 +661,7 @@ def find_clusters(self, max_opinion_difference: float = 0.1, min_influence: floa
             min_influence (float): Minimum required influence to form a cluster, adjusted by the size of the node.
 
         Returns:
-            List[List[int]]: A list of lists, where each inner list represents a cluster of node identifiers.
+            list[list[int]]: A list of lists, where each inner list represents a cluster of node identifiers.
         """
         clusters = []
         visited_nodes = set()
@@ -713,12 +711,12 @@ def find_clusters(self, max_opinion_difference: float = 0.1, min_influence: floa
 
         return clusters
 
-    def merge_by_intervals(self, intervals: List[float]):
+    def merge_by_intervals(self, intervals: list[float]):
         """
         Merges nodes into clusters based on the intervals defined by the input list of opinions.
 
         Parameters:
-            intervals (List[float]): A sorted list of opinions representing the boundaries of the intervals.
+            intervals (list[float]): A sorted list of opinions representing the boundaries of the intervals.
         """
         if not intervals:
             raise ValueError("Intervals list cannot be empty")
@@ -818,7 +816,7 @@ def opinions(self):
         return {node: self.nodes[node]['Opinion'] for node in self.nodes}
 
     @opinions.setter
-    def opinions(self, values: Union[int, float, Dict[Tuple[int]:float]]):
+    def opinions(self, values: Union[int, float, dict[tuple[int]:float]]):
         if isinstance(values, (int, float)):
             for node in self.nodes:
                 self.nodes[node]['Opinion'] = values