Skip to content

City

Cities

Bases: Supergroup

A collection of cities.

Source code in june/geography/city.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
class Cities(Supergroup):
    """A collection of cities."""

    def __init__(self, cities: List[City], ball_tree=True):
        super().__init__(cities)
        self.members_by_name = {city.name: city for city in cities}
        if ball_tree:
            self._ball_tree = self._construct_ball_tree()

    @classmethod
    def for_super_areas(
        cls,
        super_areas: List[SuperArea],
        city_super_areas_filename=default_cities_filename,
    ):
        """Initialises the cities which are on the given super areas.

        Args:
            super_areas (List[SuperArea]): 
            city_super_areas_filename: (Default value = default_cities_filename)

        """
        city_super_areas = pd.read_csv(city_super_areas_filename)
        city_super_areas = city_super_areas.loc[
            city_super_areas.super_area.isin(
                [super_area.name for super_area in super_areas]
            )
        ]
        city_super_areas.reset_index(inplace=True)
        city_super_areas.set_index("city", inplace=True)
        cities = []
        for city in city_super_areas.index.unique():
            super_area_names = city_super_areas.loc[city, "super_area"]
            if type(super_area_names) == str:
                super_area_names = [super_area_names]
            else:
                super_area_names = super_area_names.values.astype(str)
            city = City(name=city, super_areas=super_area_names)
            lats = []
            lons = []
            for super_area_name in super_area_names:
                super_area = super_areas.members_by_name[super_area_name]
                super_area.city = city
                lats.append(super_area.coordinates[0])
                lons.append(super_area.coordinates[1])
            city.coordinates = _calculate_centroid(lats, lons)
            city.super_area = super_areas.get_closest_super_area(city.coordinates)
            cities.append(city)
        return cls(cities)

    @classmethod
    def for_geography(
        cls, geography: Geography, city_super_areas_filename=default_cities_filename
    ):
        """

        Args:
            geography (Geography): 
            city_super_areas_filename: (Default value = default_cities_filename)

        """
        return cls.for_super_areas(
            super_areas=geography.super_areas,
            city_super_areas_filename=city_super_areas_filename,
        )

    def _construct_ball_tree(self):
        """Constructs a NN tree with the haversine metric for the cities."""
        coordinates = np.array([np.deg2rad(city.coordinates) for city in self if city.coordinates is not None])
        if coordinates.size == 0:
            # No valid coordinates, return None to indicate no ball tree
            return None
        ball_tree = BallTree(coordinates, metric="haversine")
        return ball_tree

    def get_closest_cities(self, coordinates, k=1, return_distance=False):
        """

        Args:
            coordinates: 
            k: (Default value = 1)
            return_distance: (Default value = False)

        """
        coordinates = np.array(coordinates)
        if self._ball_tree is None:
            raise ValueError("Cities initialised without a BallTree - no cities with valid coordinates found")
        if coordinates.shape == (2,):
            coordinates = coordinates.reshape(1, -1)
        if return_distance:
            distances, indcs = self._ball_tree.query(
                np.deg2rad(coordinates), return_distance=return_distance, k=k
            )
            if coordinates.shape == (1, 2):
                cities = [self[idx] for idx in indcs[0]]
                return cities, distances[0] * earth_radius
            else:
                cities = [self[idx] for idx in indcs[:, 0]]
                return cities, distances[:, 0] * earth_radius
        else:
            indcs = self._ball_tree.query(
                np.deg2rad(coordinates), return_distance=return_distance, k=k
            )
            cities = [self[idx] for idx in indcs[0]]
            return cities

    def get_by_name(self, city_name):
        """

        Args:
            city_name: 

        """
        return self.members_by_name[city_name]

    def get_closest_city(self, coordinates):
        """

        Args:
            coordinates: 

        """
        return self.get_closest_cities(coordinates, k=1, return_distance=False)[0]

    def get_closest_commuting_city(self, coordinates):
        """

        Args:
            coordinates: 

        """
        cities_by_distance = self.get_closest_cities(coordinates, k=len(self.members))
        for city in cities_by_distance:
            if city.stations.members:
                return city
        logger.warning("No commuting city in this world.")

for_geography(geography, city_super_areas_filename=default_cities_filename) classmethod

Parameters:

Name Type Description Default
geography Geography
required
city_super_areas_filename

(Default value = default_cities_filename)

 default_cities_filename
Source code in june/geography/city.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
@classmethod
def for_geography(
    cls, geography: Geography, city_super_areas_filename=default_cities_filename
):
    """

    Args:
        geography (Geography): 
        city_super_areas_filename: (Default value = default_cities_filename)

    """
    return cls.for_super_areas(
        super_areas=geography.super_areas,
        city_super_areas_filename=city_super_areas_filename,
    )

for_super_areas(super_areas, city_super_areas_filename=default_cities_filename) classmethod

Initialises the cities which are on the given super areas.

Parameters:

Name Type Description Default
super_areas List[SuperArea]
required
city_super_areas_filename

(Default value = default_cities_filename)

 default_cities_filename
Source code in june/geography/city.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
@classmethod
def for_super_areas(
    cls,
    super_areas: List[SuperArea],
    city_super_areas_filename=default_cities_filename,
):
    """Initialises the cities which are on the given super areas.

    Args:
        super_areas (List[SuperArea]): 
        city_super_areas_filename: (Default value = default_cities_filename)

    """
    city_super_areas = pd.read_csv(city_super_areas_filename)
    city_super_areas = city_super_areas.loc[
        city_super_areas.super_area.isin(
            [super_area.name for super_area in super_areas]
        )
    ]
    city_super_areas.reset_index(inplace=True)
    city_super_areas.set_index("city", inplace=True)
    cities = []
    for city in city_super_areas.index.unique():
        super_area_names = city_super_areas.loc[city, "super_area"]
        if type(super_area_names) == str:
            super_area_names = [super_area_names]
        else:
            super_area_names = super_area_names.values.astype(str)
        city = City(name=city, super_areas=super_area_names)
        lats = []
        lons = []
        for super_area_name in super_area_names:
            super_area = super_areas.members_by_name[super_area_name]
            super_area.city = city
            lats.append(super_area.coordinates[0])
            lons.append(super_area.coordinates[1])
        city.coordinates = _calculate_centroid(lats, lons)
        city.super_area = super_areas.get_closest_super_area(city.coordinates)
        cities.append(city)
    return cls(cities)

get_by_name(city_name)

Parameters:

Name Type Description Default
city_name
required
Source code in june/geography/city.py 
248
249
250
251
252
253
254
255
def get_by_name(self, city_name):
    """

    Args:
        city_name: 

    """
    return self.members_by_name[city_name]

get_closest_cities(coordinates, k=1, return_distance=False)

Parameters:

Name Type Description Default
coordinates
required
k

(Default value = 1)

 1
return_distance

(Default value = False)

 False
Source code in june/geography/city.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def get_closest_cities(self, coordinates, k=1, return_distance=False):
    """

    Args:
        coordinates: 
        k: (Default value = 1)
        return_distance: (Default value = False)

    """
    coordinates = np.array(coordinates)
    if self._ball_tree is None:
        raise ValueError("Cities initialised without a BallTree - no cities with valid coordinates found")
    if coordinates.shape == (2,):
        coordinates = coordinates.reshape(1, -1)
    if return_distance:
        distances, indcs = self._ball_tree.query(
            np.deg2rad(coordinates), return_distance=return_distance, k=k
        )
        if coordinates.shape == (1, 2):
            cities = [self[idx] for idx in indcs[0]]
            return cities, distances[0] * earth_radius
        else:
            cities = [self[idx] for idx in indcs[:, 0]]
            return cities, distances[:, 0] * earth_radius
    else:
        indcs = self._ball_tree.query(
            np.deg2rad(coordinates), return_distance=return_distance, k=k
        )
        cities = [self[idx] for idx in indcs[0]]
        return cities

get_closest_city(coordinates)

Parameters:

Name Type Description Default
coordinates
required
Source code in june/geography/city.py 
257
258
259
260
261
262
263
264
def get_closest_city(self, coordinates):
    """

    Args:
        coordinates: 

    """
    return self.get_closest_cities(coordinates, k=1, return_distance=False)[0]

get_closest_commuting_city(coordinates)

Parameters:

Name Type Description Default
coordinates
required
Source code in june/geography/city.py 
266
267
268
269
270
271
272
273
274
275
276
277
def get_closest_commuting_city(self, coordinates):
    """

    Args:
        coordinates: 

    """
    cities_by_distance = self.get_closest_cities(coordinates, k=len(self.members))
    for city in cities_by_distance:
        if city.stations.members:
            return city
    logger.warning("No commuting city in this world.")

City

A city is a collection of areas, with some added methods for functionality, such as commuting or local lockdowns.

Source code in june/geography/city.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class City:
    """A city is a collection of areas, with some added methods for functionality,
    such as commuting or local lockdowns.

    """

    external = False

    _id = count()

    def __init__(
        self,
        super_areas: List[str] = None,
        super_area: SuperArea = None,
        name: str = None,
        coordinates=None,
    ):
        """
        Initialises a city. A city is defined by a collection of ``super_areas`` and is located at one particular ``super_area``.
        The location to one ``super_area`` is necessary for domain parallelisation.

        Parameters
        ----------
        super_areas:
            A list of super area names
        super_area:
            The ``SuperArea`` instance of where the city resides
        name
            The city name
        coordinates
            A tuple or array of floats indicating latitude and longitude of the city (in degrees).
        """
        self.id = next(self._id)
        self.super_area = super_area
        self.super_areas = super_areas
        self.name = name
        self.super_stations = None
        self.city_stations = []
        self.inter_city_stations = []
        self.coordinates = coordinates
        self.internal_commuter_ids = set()  # internal commuters in the city

    @classmethod
    def from_file(cls, name, city_super_areas_filename=default_cities_filename):
        """

        Args:
            name: 
            city_super_areas_filename: (Default value = default_cities_filename)

        """
        city_super_areas_df = pd.read_csv(city_super_areas_filename)
        city_super_areas_df.set_index("city", inplace=True)
        return cls.from_df(name=name, city_super_areas_df=city_super_areas_df)

    @classmethod
    def from_df(cls, name, city_super_areas_df):
        """

        Args:
            name: 
            city_super_areas_df: 

        """
        city_super_areas = city_super_areas_df.loc[name].values
        return cls(super_areas=city_super_areas, name=name)

    def get_commute_subgroup(self, person):
        """Gets the commute subgroup of the person. We first check if
        the person is in the list of the internal city commuters. If not,
        we then check if the person is a commuter in their closest city station.
        If none of the above, then that person doesn't need commuting.

        Args:
            person: 

        """
        if not self.has_stations:
            return
        if person.id in self.internal_commuter_ids:
            internal_station = self.city_stations[
                randint(0, len(self.city_stations) - 1)
            ]
            return internal_station.get_commute_subgroup()
        else:
            closest_inter_city_station = (
                person.super_area.closest_inter_city_station_for_city[self.name]
            )
            if person.id in closest_inter_city_station.commuter_ids:
                return closest_inter_city_station.get_commute_subgroup()

    def get_closest_inter_city_station(self, coordinates):
        """

        Args:
            coordinates: 

        """
        return self.inter_city_stations.get_closest_station(coordinates)

    @property
    def has_stations(self):
        """ """
        return ((self.city_stations is not None) and (len(self.city_stations) > 0)) or (
            (self.inter_city_stations is not None) and len(self.inter_city_stations) > 0
        )

has_stations property

__init__(super_areas=None, super_area=None, name=None, coordinates=None)

Initialises a city. A city is defined by a collection of super_areas and is located at one particular super_area. The location to one super_area is necessary for domain parallelisation.

Parameters

super_areas: A list of super area names super_area: The SuperArea instance of where the city resides name The city name coordinates A tuple or array of floats indicating latitude and longitude of the city (in degrees).

Source code in june/geography/city.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    super_areas: List[str] = None,
    super_area: SuperArea = None,
    name: str = None,
    coordinates=None,
):
    """
    Initialises a city. A city is defined by a collection of ``super_areas`` and is located at one particular ``super_area``.
    The location to one ``super_area`` is necessary for domain parallelisation.

    Parameters
    ----------
    super_areas:
        A list of super area names
    super_area:
        The ``SuperArea`` instance of where the city resides
    name
        The city name
    coordinates
        A tuple or array of floats indicating latitude and longitude of the city (in degrees).
    """
    self.id = next(self._id)
    self.super_area = super_area
    self.super_areas = super_areas
    self.name = name
    self.super_stations = None
    self.city_stations = []
    self.inter_city_stations = []
    self.coordinates = coordinates
    self.internal_commuter_ids = set()  # internal commuters in the city

from_df(name, city_super_areas_df) classmethod

Parameters:

Name Type Description Default
name
required
city_super_areas_df
required
Source code in june/geography/city.py 
89
90
91
92
93
94
95
96
97
98
99
@classmethod
def from_df(cls, name, city_super_areas_df):
    """

    Args:
        name: 
        city_super_areas_df: 

    """
    city_super_areas = city_super_areas_df.loc[name].values
    return cls(super_areas=city_super_areas, name=name)

from_file(name, city_super_areas_filename=default_cities_filename) classmethod

Parameters:

Name Type Description Default
name
required
city_super_areas_filename

(Default value = default_cities_filename)

 default_cities_filename
Source code in june/geography/city.py 
76
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def from_file(cls, name, city_super_areas_filename=default_cities_filename):
    """

    Args:
        name: 
        city_super_areas_filename: (Default value = default_cities_filename)

    """
    city_super_areas_df = pd.read_csv(city_super_areas_filename)
    city_super_areas_df.set_index("city", inplace=True)
    return cls.from_df(name=name, city_super_areas_df=city_super_areas_df)

get_closest_inter_city_station(coordinates)

Parameters:

Name Type Description Default
coordinates
required
Source code in june/geography/city.py 
125
126
127
128
129
130
131
132
def get_closest_inter_city_station(self, coordinates):
    """

    Args:
        coordinates: 

    """
    return self.inter_city_stations.get_closest_station(coordinates)

get_commute_subgroup(person)

Gets the commute subgroup of the person. We first check if the person is in the list of the internal city commuters. If not, we then check if the person is a commuter in their closest city station. If none of the above, then that person doesn't need commuting.

Parameters:

Name Type Description Default
person
required
Source code in june/geography/city.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def get_commute_subgroup(self, person):
    """Gets the commute subgroup of the person. We first check if
    the person is in the list of the internal city commuters. If not,
    we then check if the person is a commuter in their closest city station.
    If none of the above, then that person doesn't need commuting.

    Args:
        person: 

    """
    if not self.has_stations:
        return
    if person.id in self.internal_commuter_ids:
        internal_station = self.city_stations[
            randint(0, len(self.city_stations) - 1)
        ]
        return internal_station.get_commute_subgroup()
    else:
        closest_inter_city_station = (
            person.super_area.closest_inter_city_station_for_city[self.name]
        )
        if person.id in closest_inter_city_station.commuter_ids:
            return closest_inter_city_station.get_commute_subgroup()

ExternalCity

Bases: ExternalGroup

This a city that lives outside the simulated domain.

Source code in june/geography/city.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
class ExternalCity(ExternalGroup):
    """This a city that lives outside the simulated domain."""

    external = True

    def __init__(self, id, domain_id, coordinates=None, commuter_ids=None, name=None):
        super().__init__(spec="city", domain_id=domain_id, id=id)
        self.internal_commuter_ids = commuter_ids or set()
        self.city_stations = []
        self.inter_city_stations = []
        self.super_area = None
        self.coordinates = coordinates
        self.name = name

    @property
    def has_stations(self):
        """ """
        return len(self.city_stations) > 0

    def get_commute_subgroup(self, person):
        """Gets the commute subgroup of the person. We first check if
        the person is in the list of the internal city commuters. If not,
        we then check if the person is a commuter in their closest city station.
        If none of the above, then that person doesn't need commuting.

        Args:
            person: 

        """
        if not self.has_stations:
            return
        if person.id in self.internal_commuter_ids:
            internal_station = self.city_stations[
                randint(0, len(self.city_stations) - 1)
            ]
            return internal_station.get_commute_subgroup()
        else:
            closest_inter_city_station = (
                person.super_area.closest_inter_city_station_for_city[self.name]
            )
            if person.id in closest_inter_city_station.commuter_ids:
                return closest_inter_city_station.get_commute_subgroup()

has_stations property

get_commute_subgroup(person)

Gets the commute subgroup of the person. We first check if the person is in the list of the internal city commuters. If not, we then check if the person is a commuter in their closest city station. If none of the above, then that person doesn't need commuting.

Parameters:

Name Type Description Default
person
required
Source code in june/geography/city.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def get_commute_subgroup(self, person):
    """Gets the commute subgroup of the person. We first check if
    the person is in the list of the internal city commuters. If not,
    we then check if the person is a commuter in their closest city station.
    If none of the above, then that person doesn't need commuting.

    Args:
        person: 

    """
    if not self.has_stations:
        return
    if person.id in self.internal_commuter_ids:
        internal_station = self.city_stations[
            randint(0, len(self.city_stations) - 1)
        ]
        return internal_station.get_commute_subgroup()
    else:
        closest_inter_city_station = (
            person.super_area.closest_inter_city_station_for_city[self.name]
        )
        if person.id in closest_inter_city_station.commuter_ids:
            return closest_inter_city_station.get_commute_subgroup()