22
33import collections
44from pathlib import Path
5- from typing import List , Optional , Callable , OrderedDict , Dict , Union , Iterable , cast
5+ from typing import Callable , OrderedDict , Iterable , cast
66
77import networkx
88import numpy as np
@@ -41,8 +41,8 @@ class Mesh2VecBase:
4141 def __init__ (
4242 self ,
4343 distance : int ,
44- hyper_edges : Dict [str , List [str ]],
45- vtx_ids : Optional [ List [ str ]] = None ,
44+ hyper_edges : dict [str , list [str ]],
45+ vtx_ids : list [ str ] | None = None ,
4646 calc_strategy : str = "dfs" ,
4747 ):
4848 # pylint: disable=line-too-long
@@ -83,7 +83,7 @@ def __init__(
8383 check_vtx_ids (vtx_ids , hyper_edges )
8484
8585 self ._distance : int = distance
86- self ._hyper_edges : OrderedDict [str , List [str ]] = collections .OrderedDict (hyper_edges )
86+ self ._hyper_edges : OrderedDict [str , list [str ]] = collections .OrderedDict (hyper_edges )
8787
8888 self ._vtx_idx_to_ids = collections .OrderedDict (enumerate (vtx_ids )) # type: ignore
8989 self ._vtx_ids_to_idx = {vtx_ids : i for i , vtx_ids in enumerate (vtx_ids )}
@@ -204,7 +204,7 @@ def from_file(hg_file: Path, distance: int, calc_strategy: str = "dfs") -> "Mesh
204204 hyper_edges_ids_to_vtx_ids = {line_split [0 ]: line_split [1 :] for line_split in lines_split }
205205 return Mesh2VecBase (distance = distance , hyper_edges = hyper_edges_ids_to_vtx_ids )
206206
207- def get_nbh (self , vtx : str , dist : int ) -> List [str ]:
207+ def get_nbh (self , vtx : str , dist : int ) -> list [str ]:
208208 """
209209 Get a list of neighbors with the exact distance ``dist`` of a given vertex ``vtx``
210210
@@ -229,10 +229,10 @@ def get_nbh(self, vtx: str, dist: int) -> List[str]:
229229 def aggregate_categorical (
230230 self ,
231231 feature : str ,
232- dist : Union [ List [ int ], int ] ,
233- categories : Optional [ Union [ List [ str ], List [int ]]] = None ,
234- default_value : Optional [ Union [ int , str ]] = None ,
235- ) -> Union [ str , List [str ] ]:
232+ dist : list [ int ] | int ,
233+ categories : list [ str ] | list [int ] | None = None ,
234+ default_value : int | str | None = None ,
235+ ) -> str | list [str ]:
236236 """
237237 For categorical features, aggregate the numbers of occurrences of each categorical value.
238238 This results in a new aggregated ``feature`` for each categorical value. If ``feature`` is
@@ -264,7 +264,7 @@ def aggregate_categorical(
264264 check_distance_arg (dist_to_check , self )
265265 check_feature_available (feature , self )
266266 feature_names = []
267- feature_categories : List [str ] | npt .NDArray [np .str_ ]
267+ feature_categories : list [str ] | npt .NDArray [np .str_ ]
268268 if categories is not None :
269269 feature_categories = [str (category ) for category in categories ] + ["NONE" ]
270270 else :
@@ -286,15 +286,15 @@ def aggregate_categorical(
286286 def aggregate (
287287 self ,
288288 feature : str ,
289- dist : Union [ List [ int ], int ] ,
290- aggr : Union [
291- Callable [[np .ndarray ], Union [ float , int , str ]],
292- Callable [[np .ndarray , Union [ float , int , str ]], Union [ float , int , str ]],
293- ] ,
294- aggr_name : Optional [ str ] = None ,
289+ dist : list [ int ] | int ,
290+ aggr : (
291+ Callable [[np .ndarray ], float | int | str ]
292+ | Callable [[np .ndarray , float | int | str ], float | int | str ]
293+ ) ,
294+ aggr_name : str | None = None ,
295295 agg_add_ref : bool = False ,
296296 default_value : float = 0.0 ,
297- ) -> Union [ str , List [str ] ]:
297+ ) -> str | list [str ]:
298298 # pylint: disable=line-too-long,too-many-arguments,too-many-positional-arguments
299299 """
300300 Aggregate features from neighborhoods for each distance in ``dist``
@@ -366,15 +366,14 @@ def _collect_feature_values(
366366 self ,
367367 feature_name : str ,
368368 dist : int ,
369- default_value : Optional [Union [float , int , str ]],
370- aggr : Optional [
371- Union [
372- Callable [[np .ndarray ], Union [float , int , str ]],
373- Callable [[np .ndarray , Union [float , int , str ]], Union [float , int , str ]],
374- ]
375- ] = None ,
376- ref_values : Optional [List [Union [float , int , str ]]] = None ,
377- ) -> List [Union [float , int , str ]]:
369+ default_value : float | int | str | None ,
370+ aggr : (
371+ Callable [[np .ndarray ], float | int | str ]
372+ | Callable [[np .ndarray , float | int | str ], float | int | str ]
373+ | None
374+ ) = None ,
375+ ref_values : list [float | int | str ] | None = None ,
376+ ) -> list [float | int | str ]:
378377 """helper method to collect and aggregate data from all hyper nodes"""
379378 # pylint: disable=too-many-arguments,too-many-positional-arguments,line-too-long
380379 check_distance_arg (dist , self )
@@ -386,11 +385,10 @@ def _collect_feature_values(
386385 if dist == 0 :
387386 return feature
388387
389- # `aggr` and `default_value` are explicitly defined in method's signature, so we just
390- # ignore typing here and let it fail in case of wrong usage.
388+ # Ignore typing here and let it fail in case of wrong usage.
391389 if ref_values is None :
392390 if aggr is not None : # fast: use nan_to_num over the whole array
393- aggr = cast (Callable [[np .ndarray ], Union [ float , int , str ] ], aggr )
391+ aggr = cast (Callable [[np .ndarray ], float | int | str ], aggr )
394392 return np .nan_to_num (
395393 [aggr (feature [neighborhood ]) for neighborhood in self ._neighborhoods [dist ]],
396394 nan = default_value , # type: ignore[arg-type]
@@ -408,7 +406,7 @@ def _collect_feature_values(
408406 raise TypeError ("When `ref_values` is given, `aggr` function to compare is needed." )
409407
410408 # compare to reference value is needed
411- aggr = cast (Callable [[np .ndarray , Union [ float , int , str ]], Union [ float , int , str ] ], aggr )
409+ aggr = cast (Callable [[np .ndarray , float | int | str ], float | int | str ], aggr )
412410 return [
413411 np .nan_to_num (
414412 aggr (feature [neighborhood ], ref_values [i ]),
@@ -421,7 +419,7 @@ def add_features_from_csv(
421419 self ,
422420 csv_file : Path ,
423421 with_header : bool = False ,
424- columns : Optional [ List [ str ]] = None ,
422+ columns : list [ str ] | None = None ,
425423 ) -> None :
426424 """Map the content of a CSV file to the vertices of the hypergraph.
427425
@@ -467,7 +465,7 @@ def add_features_from_dataframe(self, df: pd.DataFrame) -> None:
467465 raise ValueError (f"Feature { new_columns_name } already exists" )
468466 self ._features = self ._features .merge (df , how = "left" , on = "vtx_id" , validate = "1:1" )
469467
470- def to_dataframe (self , vertices : Optional [ Iterable [str ]] = None ) -> pd .DataFrame :
468+ def to_dataframe (self , vertices : Iterable [str ] | None = None ) -> pd .DataFrame :
471469 """
472470 Returns a Pandas dataframe with all the beforehand aggregated feature columns. If
473471 ``vertices`` is not ``None`` and iterable, the dataframe is only generated for vertices
@@ -477,7 +475,7 @@ def to_dataframe(self, vertices: Optional[Iterable[str]] = None) -> pd.DataFrame
477475 return self ._aggregated_features [self ._features in vertices ].copy ()
478476 return self ._aggregated_features .copy ()
479477
480- def to_array (self , vertices : Optional [ Iterable [str ]] = None ) -> np .ndarray :
478+ def to_array (self , vertices : Iterable [str ] | None = None ) -> np .ndarray :
481479 """
482480 Returns a numpy array with all the beforehand aggregated feature columns. If
483481 ``vertices`` is not ``None`` and iterable, the array is only generated for vertices
@@ -489,15 +487,15 @@ def get_max_distance(self) -> int:
489487 """returns the distance value used to generate the hypergraph neighborhood"""
490488 return self ._distance
491489
492- def available_features (self ) -> List [str ]:
490+ def available_features (self ) -> list [str ]:
493491 """returns a list the names of all features"""
494492 return self ._features .drop ("vtx_id" , axis = 1 ).keys ().tolist ()
495493
496- def available_aggregated_features (self ) -> List [str ]:
494+ def available_aggregated_features (self ) -> list [str ]:
497495 """returns a list the names of all aggregated features"""
498496 return self ._aggregated_features .drop ("vtx_id" , axis = 1 ).keys ().tolist ()
499497
500- def vtx_ids (self ) -> List [str ]:
498+ def vtx_ids (self ) -> list [str ]:
501499 """returns a list the ids of all hyper vertices"""
502500 return list (self ._vtx_ids_to_idx .keys ())
503501
0 commit comments