Social venue

SocialVenue

Bases: Group

Source code in june/groups/leisure/social_venue.py

class SocialVenue(Group):
    """ """
    max_size = np.inf

    def __init__(self, area=None):
        super().__init__()  # Pass disease_config to the parent class
        self.area = area
        # Use 0 as domain_id in non-MPI mode
        self.domain_id = 0 if not mpi_available else mpi_rank


    def add(self, person, activity="leisure"):
        """

        Args:
            person: 
            activity: (Default value = "leisure")

        """
        self.subgroups[0].append(person)
        setattr(person.subgroups, activity, self.subgroups[0])

    @property
    def super_area(self):
        """ """
        return self.area.super_area

    @property
    def get_coordinates(self):
        """ """
        if self.area is None:
            return
        else:
            return self.area.coordinates

get_coordinates property

super_area property

add(person, activity='leisure')

Parameters:

Name	Type	Description	Default
person			required
activity		(Default value = "leisure")	'leisure'

Source code in june/groups/leisure/social_venue.py

def add(self, person, activity="leisure"):
    """

    Args:
        person: 
        activity: (Default value = "leisure")

    """
    self.subgroups[0].append(person)
    setattr(person.subgroups, activity, self.subgroups[0])

SocialVenueError

Bases: BaseException

Source code in june/groups/leisure/social_venue.py

class SocialVenueError(BaseException):
    """ """
    pass

SocialVenues

Bases: Supergroup

Source code in june/groups/leisure/social_venue.py

class SocialVenues(Supergroup):
    """ """
    venue_class = SocialVenue

    def __init__(self, social_venues: List[venue_class], make_tree=True):
        super().__init__(members=social_venues)
        # Use a safer logging approach that works in both MPI and non-MPI modes
        current_rank = 0 if not mpi_available else mpi_rank
        logger.info(f"Domain {current_rank} has {len(self)} {self.spec}(s)")
        self.ball_tree = None
        if make_tree:
            if not social_venues:
                logger.warning(f"No social venues of type {self.spec} in this domain")
            else:
                self.make_tree()

    @classmethod
    def from_coordinates(
        cls,
        coordinates: List[np.array],
        super_areas: Optional[SuperAreas],
        max_distance_to_area=10,
        **kwargs,
    ):
        """

        Args:
            coordinates (List[np.array]): 
            super_areas (Optional[SuperAreas]): 
            max_distance_to_area: (Default value = 10)
            **kwargs: 

        """
        if len(coordinates) == 0:
            return cls([], **kwargs)

        if super_areas:
            super_areas, distances = super_areas.get_closest_super_areas(
                coordinates, k=1, return_distance=True
            )
            distances_close = np.where(distances < max_distance_to_area)
            coordinates = coordinates[distances_close]
            super_areas = [super_areas[i] for i in distances_close[0]]

        social_venues = []

        for i, coord in enumerate(coordinates):
            sv = cls.venue_class()  # Pass disease_config here
            if super_areas:
                super_area = super_areas[i]
            else:
                super_area = None
            sv.coordinates = coord
            if super_areas:
                area, dist = Areas(super_area.areas).get_closest_area(
                    coordinates=coord, return_distance=True
                )

                if dist > max_distance_to_area:
                    continue

                sv.area = area
            social_venues.append(sv)

        return cls(social_venues, **kwargs)

    @classmethod
    def for_super_areas(
        cls, super_areas: List[SuperArea], coordinates_filename: str = None
    ):
        """

        Args:
            super_areas (List[SuperArea]): 
            coordinates_filename (str, optional): (Default value = None)

        """
        if coordinates_filename is None:
            coordinates_filename = cls.default_coordinates_filename
        sv_coordinates = pd.read_csv(coordinates_filename, index_col=0).values

        # Generate social venues using the coordinates
        social_venues_instance = cls.from_coordinates(
            sv_coordinates, super_areas=super_areas
        )

        # Prepare sample data for visualization
        venue_data = [
            {
                "Venue ID": venue.id,
                "Super Area": venue.area.super_area.name if venue.area and venue.area.super_area else "Unknown",
                "Coordinates": venue.coordinates,
                "Assigned Area": venue.area.name if venue.area else "None",
            }
            for venue in social_venues_instance.members
        ]

        # Convert sample data to DataFrame for better readability
        df_venues = pd.DataFrame(venue_data)
        print("\n===== Gaps in ID represent the venues that were created that did not fulfill the criteria required =====")
        print("\n===== Sample of Social Venues Created from Coordinates =====")
        print(df_venues)

        return social_venues_instance


    @classmethod
    def for_areas(cls, areas: Areas, coordinates_filename: str = None):
        """

        Args:
            areas (Areas): 
            coordinates_filename (str, optional): (Default value = None)

        """
        if coordinates_filename is None:
            coordinates_filename = cls.default_coordinates_filename
        super_areas = SuperAreas([area.super_area for area in areas])
        return cls.for_super_areas(super_areas, coordinates_filename)

    @classmethod
    def for_geography(cls, geography: Geography, coordinates_filename: str = None):
        """

        Args:
            geography (Geography): 
            coordinates_filename (str, optional): (Default value = None)

        """
        if coordinates_filename is None:
            coordinates_filename = cls.default_coordinates_filename

        return cls.for_super_areas(geography.super_areas, coordinates_filename)

    @classmethod
    def distribute_for_areas(
        cls,
        areas: List[Area],
        venues_per_capita: float = None,
        venues_per_area: int = None,
    ):
        """Generates social venues in the given areas.

        Args:
            areas (List[Area]): list of areas to generate the venues in
            venues_per_capita (float, optional): number of venues per person in each area. (Default value = None)
            venues_per_area (int, optional): number of venues in each area. (Default value = None)

        """
        if venues_per_area is not None and venues_per_capita is not None:
            raise SocialVenueError(
                "Please specify only one of venues_per_capita or venues_per_area."
            )
        social_venues = []
        if venues_per_area is not None:
            for area in areas:
                for _ in range(venues_per_area):
                    sv = cls.venue_class()
                    sv.area = area
                    social_venues.append(sv)
        elif venues_per_capita is not None:
            for area in areas:
                area_population = len(area.people)
                for _ in range(int(np.ceil(venues_per_capita * area_population))):
                    sv = cls.venue_class()
                    sv.area = area
                    sv.coordinates = area.coordinates
                    social_venues.append(sv)
        else:
            raise SocialVenueError(
                "Specify one of venues_per_capita or venues_per_area"
            )
        return cls(social_venues)

    @classmethod
    def distribute_for_super_areas(
        cls, super_areas: List[SuperArea], venues_per_super_area=1, venues_per_capita=1
    ):
        """Generates social venues in the given super areas.

        Args:
            super_areas (List[SuperArea]): list of areas to generate the venues in
            venues_per_super_area: how many venus per super_area to generate (Default value = 1)
            venues_per_capita: (Default value = 1)

        """
        if venues_per_super_area is not None and venues_per_capita is not None:
            raise SocialVenueError(
                "Please specify only one of venues_per_capita or venues_per_area."
            )
        social_venues = []
        if venues_per_super_area is not None:
            for area in super_areas:
                for _ in range(venues_per_super_area):
                    sv = cls.venue_class()
                    sv.area = area
                    social_venues.append(sv)
        elif venues_per_capita is not None:
            for super_area in super_areas:
                super_area_population = len(super_area.people)
                for _ in range(int(np.ceil(venues_per_capita * super_area_population))):
                    sv = cls.venue_class()
                    area = Areas(super_area.areas).get_closest_area(
                        coordinates=super_area.coordinates
                    )
                    sv.area = area
                    sv.coordinates = area.coordinates
                    social_venues.append(sv)
        else:
            raise SocialVenueError(
                "Specify one of venues_per_capita or venues_per_area"
            )
        return cls(social_venues)

    def make_tree(self):
        """ """
        self.ball_tree = BallTree(
            np.array([np.deg2rad(sv.coordinates) for sv in self]), metric="haversine"
        )

    def add_to_areas(self, areas: Areas):
        """Adds all venues to the closest super area

        Args:
            areas (Areas): 

        """
        for venue in self:
            if not hasattr(venue, "coordinates"):
                raise SocialVenueError(
                    "Can't add to super area if venues don't have coordiantes."
                )
            venue.area = areas.get_closest_areas(venue.coordinates)[0]

    def get_closest_venues(self, coordinates, k=1):
        """Queries the ball tree for the closests venues.

        Args:
            coordinates: coordinates in the format [Latitude, Longitude]
            k: number of neighbours desired (Default value = 1)

        """
        if not self.members:
            return
        if self.ball_tree is None:
            raise SocialVenueError("Initialise ball tree first with self.make_tree()")
        venue_idxs = self.ball_tree.query(
            np.deg2rad(coordinates).reshape(1, -1), return_distance=False, k=k
        ).flatten()
        social_venues = self.members
        return [social_venues[idx] for idx in venue_idxs]

    def get_venues_in_radius(self, coordinates, radius=5):
        """Queries the ball tree for the closests venues.

        Args:
            coordinates: coordinates in the format [Latitude, Longitude]
            radius: radius in km to query (Default value = 5)

        """
        if not self.members:
            return
        if self.ball_tree is None:
            raise SocialVenueError("Initialise ball tree first with self.make_tree()")
        radius = radius / earth_radius
        venue_idxs, _ = self.ball_tree.query_radius(
            np.deg2rad(coordinates).reshape(1, -1),
            r=radius,
            sort_results=True,
            return_distance=True,
        )
        venue_idxs = venue_idxs[0]
        if not venue_idxs.size:
            return None
        social_venues = self.members
        return [social_venues[idx] for idx in venue_idxs]

    def get_leisure_subgroup(self, person, subgroup_type, to_send_abroad):
        """

        Args:
            person: 
            subgroup_type: 
            to_send_abroad: 

        """
        return self[subgroup_type]

add_to_areas(areas)

Adds all venues to the closest super area

Parameters:

Name	Type	Description	Default
areas	Areas		required

Source code in june/groups/leisure/social_venue.py

def add_to_areas(self, areas: Areas):
    """Adds all venues to the closest super area

    Args:
        areas (Areas): 

    """
    for venue in self:
        if not hasattr(venue, "coordinates"):
            raise SocialVenueError(
                "Can't add to super area if venues don't have coordiantes."
            )
        venue.area = areas.get_closest_areas(venue.coordinates)[0]

distribute_for_areas(areas, venues_per_capita=None, venues_per_area=None) classmethod

Generates social venues in the given areas.

Parameters:

Name	Type	Description	Default
areas	List[Area]	list of areas to generate the venues in	required
venues_per_capita	float	number of venues per person in each area. (Default value = None)	None
venues_per_area	int	number of venues in each area. (Default value = None)	None

Source code in june/groups/leisure/social_venue.py

@classmethod
def distribute_for_areas(
    cls,
    areas: List[Area],
    venues_per_capita: float = None,
    venues_per_area: int = None,
):
    """Generates social venues in the given areas.

    Args:
        areas (List[Area]): list of areas to generate the venues in
        venues_per_capita (float, optional): number of venues per person in each area. (Default value = None)
        venues_per_area (int, optional): number of venues in each area. (Default value = None)

    """
    if venues_per_area is not None and venues_per_capita is not None:
        raise SocialVenueError(
            "Please specify only one of venues_per_capita or venues_per_area."
        )
    social_venues = []
    if venues_per_area is not None:
        for area in areas:
            for _ in range(venues_per_area):
                sv = cls.venue_class()
                sv.area = area
                social_venues.append(sv)
    elif venues_per_capita is not None:
        for area in areas:
            area_population = len(area.people)
            for _ in range(int(np.ceil(venues_per_capita * area_population))):
                sv = cls.venue_class()
                sv.area = area
                sv.coordinates = area.coordinates
                social_venues.append(sv)
    else:
        raise SocialVenueError(
            "Specify one of venues_per_capita or venues_per_area"
        )
    return cls(social_venues)

distribute_for_super_areas(super_areas, venues_per_super_area=1, venues_per_capita=1) classmethod

Generates social venues in the given super areas.

Parameters:

Name	Type	Description	Default
super_areas	List[SuperArea]	list of areas to generate the venues in	required
venues_per_super_area		how many venus per super_area to generate (Default value = 1)	1
venues_per_capita		(Default value = 1)	1

Source code in june/groups/leisure/social_venue.py

@classmethod
def distribute_for_super_areas(
    cls, super_areas: List[SuperArea], venues_per_super_area=1, venues_per_capita=1
):
    """Generates social venues in the given super areas.

    Args:
        super_areas (List[SuperArea]): list of areas to generate the venues in
        venues_per_super_area: how many venus per super_area to generate (Default value = 1)
        venues_per_capita: (Default value = 1)

    """
    if venues_per_super_area is not None and venues_per_capita is not None:
        raise SocialVenueError(
            "Please specify only one of venues_per_capita or venues_per_area."
        )
    social_venues = []
    if venues_per_super_area is not None:
        for area in super_areas:
            for _ in range(venues_per_super_area):
                sv = cls.venue_class()
                sv.area = area
                social_venues.append(sv)
    elif venues_per_capita is not None:
        for super_area in super_areas:
            super_area_population = len(super_area.people)
            for _ in range(int(np.ceil(venues_per_capita * super_area_population))):
                sv = cls.venue_class()
                area = Areas(super_area.areas).get_closest_area(
                    coordinates=super_area.coordinates
                )
                sv.area = area
                sv.coordinates = area.coordinates
                social_venues.append(sv)
    else:
        raise SocialVenueError(
            "Specify one of venues_per_capita or venues_per_area"
        )
    return cls(social_venues)

for_areas(areas, coordinates_filename=None) classmethod

Parameters:

Name	Type	Description	Default
areas	Areas		required
coordinates_filename	str	(Default value = None)	None

Source code in june/groups/leisure/social_venue.py

@classmethod
def for_areas(cls, areas: Areas, coordinates_filename: str = None):
    """

    Args:
        areas (Areas): 
        coordinates_filename (str, optional): (Default value = None)

    """
    if coordinates_filename is None:
        coordinates_filename = cls.default_coordinates_filename
    super_areas = SuperAreas([area.super_area for area in areas])
    return cls.for_super_areas(super_areas, coordinates_filename)

for_geography(geography, coordinates_filename=None) classmethod

Parameters:

Name	Type	Description	Default
geography	Geography		required
coordinates_filename	str	(Default value = None)	None

Source code in june/groups/leisure/social_venue.py

@classmethod
def for_geography(cls, geography: Geography, coordinates_filename: str = None):
    """

    Args:
        geography (Geography): 
        coordinates_filename (str, optional): (Default value = None)

    """
    if coordinates_filename is None:
        coordinates_filename = cls.default_coordinates_filename

    return cls.for_super_areas(geography.super_areas, coordinates_filename)

for_super_areas(super_areas, coordinates_filename=None) classmethod

Parameters:

Name	Type	Description	Default
super_areas	List[SuperArea]		required
coordinates_filename	str	(Default value = None)	None

Source code in june/groups/leisure/social_venue.py

@classmethod
def for_super_areas(
    cls, super_areas: List[SuperArea], coordinates_filename: str = None
):
    """

    Args:
        super_areas (List[SuperArea]): 
        coordinates_filename (str, optional): (Default value = None)

    """
    if coordinates_filename is None:
        coordinates_filename = cls.default_coordinates_filename
    sv_coordinates = pd.read_csv(coordinates_filename, index_col=0).values

    # Generate social venues using the coordinates
    social_venues_instance = cls.from_coordinates(
        sv_coordinates, super_areas=super_areas
    )

    # Prepare sample data for visualization
    venue_data = [
        {
            "Venue ID": venue.id,
            "Super Area": venue.area.super_area.name if venue.area and venue.area.super_area else "Unknown",
            "Coordinates": venue.coordinates,
            "Assigned Area": venue.area.name if venue.area else "None",
        }
        for venue in social_venues_instance.members
    ]

    # Convert sample data to DataFrame for better readability
    df_venues = pd.DataFrame(venue_data)
    print("\n===== Gaps in ID represent the venues that were created that did not fulfill the criteria required =====")
    print("\n===== Sample of Social Venues Created from Coordinates =====")
    print(df_venues)

    return social_venues_instance

from_coordinates(coordinates, super_areas, max_distance_to_area=10, **kwargs) classmethod

Parameters:

Name	Type	Description	Default
coordinates	List[array]		required
super_areas	Optional[SuperAreas]		required
max_distance_to_area		(Default value = 10)	10
**kwargs			{}

Source code in june/groups/leisure/social_venue.py

@classmethod
def from_coordinates(
    cls,
    coordinates: List[np.array],
    super_areas: Optional[SuperAreas],
    max_distance_to_area=10,
    **kwargs,
):
    """

    Args:
        coordinates (List[np.array]): 
        super_areas (Optional[SuperAreas]): 
        max_distance_to_area: (Default value = 10)
        **kwargs: 

    """
    if len(coordinates) == 0:
        return cls([], **kwargs)

    if super_areas:
        super_areas, distances = super_areas.get_closest_super_areas(
            coordinates, k=1, return_distance=True
        )
        distances_close = np.where(distances < max_distance_to_area)
        coordinates = coordinates[distances_close]
        super_areas = [super_areas[i] for i in distances_close[0]]

    social_venues = []

    for i, coord in enumerate(coordinates):
        sv = cls.venue_class()  # Pass disease_config here
        if super_areas:
            super_area = super_areas[i]
        else:
            super_area = None
        sv.coordinates = coord
        if super_areas:
            area, dist = Areas(super_area.areas).get_closest_area(
                coordinates=coord, return_distance=True
            )

            if dist > max_distance_to_area:
                continue

            sv.area = area
        social_venues.append(sv)

    return cls(social_venues, **kwargs)

get_closest_venues(coordinates, k=1)

Queries the ball tree for the closests venues.

Parameters:

Name	Type	Description	Default
coordinates		coordinates in the format [Latitude, Longitude]	required
k		number of neighbours desired (Default value = 1)	1

Source code in june/groups/leisure/social_venue.py

def get_closest_venues(self, coordinates, k=1):
    """Queries the ball tree for the closests venues.

    Args:
        coordinates: coordinates in the format [Latitude, Longitude]
        k: number of neighbours desired (Default value = 1)

    """
    if not self.members:
        return
    if self.ball_tree is None:
        raise SocialVenueError("Initialise ball tree first with self.make_tree()")
    venue_idxs = self.ball_tree.query(
        np.deg2rad(coordinates).reshape(1, -1), return_distance=False, k=k
    ).flatten()
    social_venues = self.members
    return [social_venues[idx] for idx in venue_idxs]

get_leisure_subgroup(person, subgroup_type, to_send_abroad)

Parameters:

Name	Type	Description	Default
person			required
subgroup_type			required
to_send_abroad			required

Source code in june/groups/leisure/social_venue.py

def get_leisure_subgroup(self, person, subgroup_type, to_send_abroad):
    """

    Args:
        person: 
        subgroup_type: 
        to_send_abroad: 

    """
    return self[subgroup_type]

get_venues_in_radius(coordinates, radius=5)

Queries the ball tree for the closests venues.

Parameters:

Name	Type	Description	Default
coordinates		coordinates in the format [Latitude, Longitude]	required
radius		radius in km to query (Default value = 5)	5

Source code in june/groups/leisure/social_venue.py

def get_venues_in_radius(self, coordinates, radius=5):
    """Queries the ball tree for the closests venues.

    Args:
        coordinates: coordinates in the format [Latitude, Longitude]
        radius: radius in km to query (Default value = 5)

    """
    if not self.members:
        return
    if self.ball_tree is None:
        raise SocialVenueError("Initialise ball tree first with self.make_tree()")
    radius = radius / earth_radius
    venue_idxs, _ = self.ball_tree.query_radius(
        np.deg2rad(coordinates).reshape(1, -1),
        r=radius,
        sort_results=True,
        return_distance=True,
    )
    venue_idxs = venue_idxs[0]
    if not venue_idxs.size:
        return None
    social_venues = self.members
    return [social_venues[idx] for idx in venue_idxs]

make_tree()

Source code in june/groups/leisure/social_venue.py

def make_tree(self):
    """ """
    self.ball_tree = BallTree(
        np.array([np.deg2rad(sv.coordinates) for sv in self]), metric="haversine"
    )