[docs]defcreate_topology(regions,ldc,mdc):""" Create a topology from group of regions. Will check for root regions, if there's not exactly 1 root region a :class:`~.topology.region.RegionGroup` will be created as new root. :param regions: Any iterable of regions. :type regions: Iterable :param ldc: Least dominant corner of the topology. Forms the suggested outer bounds of the topology together with the `mdc`. :param mdc: Most dominant corner of the topology. Forms the suggested outer bounds of the topology together with the `mdc`. """regions=list(regions)iflen(roots:=get_root_regions(regions))==1:topology=roots[0]else:topology=RegionGroup(children=roots,name="topology")hint=box_layout(ldc,mdc)topology.do_layout(hint)returntopology
[docs]defget_partitions(regions):""" Get all the partitions belonging to the group of regions and their subregions. :param regions: Any iterable of regions. :type regions: Iterable """defcollect_deps(region,ignore):# get_dependencies can be overridden, so `list` it to avoid mutation of userdatadeps=list(region.get_dependencies())regions=[dfordindepsifnotis_partition(d)]parts=set(pforpindepsifis_partition(p))ignore.update(parts)forrinregions:# Only check unchecked regionsifrnotinignore:ignore.add(r)parts.update(collect_deps(r,ignore))returnpartspartitions=set()forregioninregions:partitions.update(collect_deps(region,set()))returnpartitions
[docs]defis_partition(obj):""" Checks if an object is a partition. """return(hasattr(obj,"get_layout")andhasattr(obj,"to_chunks")andhasattr(obj,"chunk_to_voxels"))
[docs]defis_region(obj):""" Checks if an object is a region. """returnhasattr(obj,"get_layout")andhasattr(obj,"do_layout")
[docs]defget_root_regions(regions):""" Get all the root regions, not belonging to any other region in the given group. :param regions: Any iterable of regions. :type regions: Iterable """managed=set()defcollect_deps(region,ignore):# get_dependencies can be overridden, so `list` it to avoid mutation of userdatadeps=list(region.get_dependencies()).copy()fordepindeps:# Only check unchecked regions, ignore visited & partitionsifdepnotinignoreandis_region(dep):ignore.add(dep)extra_deps=collect_deps(dep,ignore)ignore.update(extra_deps)returndeps# Give `managed` as the mutable ignore arg so that it is filled with all regionals# encountered as dependencies by the `collect_deps` recursive function.forregioninregions:collect_deps(region,managed)returnlist(set(list(regions))-managed)