Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return self.__class__.__name__ + "()"
def __eq__(self, other):
return isinstance(other, CellSpatialUnit)
def __hash__(self):
# We may never need CellSpatialUnits to be hashable, but we define this
# just in case.
return hash(str(self))
@property
def canonical_name(self) -> str:
return "cell"
class GeomSpatialUnit(SpatialUnitMixin, Query):
"""
Base class for spatial units that map location IDs in
connection.location_table to geographic locations.
Parameters
----------
geom_table_column_names : str or list
Name(s) of the column(s) to fetch from geom_table.
location_id_column_names : str or list
Name(s) of the column(s) which identify the locations.
Must be a subset of the column_names for this query.
geom_table : str or flowmachine.Query, optional
Name of the table containing the geography information.
Can be either the name of a table, with the schema, or a
flowmachine.Query object.
Defaults to connection.location_table
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""
Classes that deal with mapping cells (or towers or sites)
to a spatial level, mostly be performing a spatial join.
Examples of this include CellToAdmin or CellToGrid.
"""
from typing import List
from ...core import Query, Grid
class CellToPolygon(Query):
"""
Class that maps a cell with a lat-lon to a geographical
region.
Parameters
----------
column_name : str, optional
The name of the column to fetch from the geometry
table in the database. Can also be a list of names.
polygon_table : str, or flowmachine.Query optional
name of the table containing the geography information.
Can be either the name of a table, with the schema, a flowmachine.Query
object, or a string representing a query.
geom_col : str, default 'geom'
column that defines the geography.
"""
def _get_subtable(self):
"""
Private method which takes the table and returns a query
representing the object. This is necessary as the table can
be passed in a variety of ways.
"""
if issubclass(self.polygon_table.__class__, Query):
return f"({self.polygon_table.get_query()}) AS polygon"
elif "select " in self.polygon_table.lower():
return f"({self.polygon_table}) AS polygon"
else:
return f"{self.polygon_table} AS polygon"
return self.unioned.column_names
def _make_query(self):
return f"SELECT DISTINCT unioned.subscriber FROM ({self.unioned.get_query()}) unioned"
def as_set(self):
"""
Returns all unique subscribers as a set.
"""
return {u[0] for u in get_db().fetch(self.get_query())}
SubsetDates = EventTableSubset # Backwards compatibility for unpicking queries from db
class SubscriberLocationSubset(Query):
"""
Query to get a subset of users who have made min_calls
number of calls within a given region during
period of time from start to stop.
Parameters
----------
start : datetime
Start time to filter query.
stop : datetime
Stop time to filter query.
spatial_unit : flowmachine.core.spatial_unit.*SpatialUnit, default admin3
Spatial unit to which subscriber locations will be mapped. See the
docstring of make_spatial_unit for more information.
min_calls : int
"""
def __init__(self, lon, lat, radius, name):
self.lon, self.lat = lon, lat
self.point_sql = (
f"ST_GeomFromText('POINT ({self.lon} {self.lat})',4326)::geography"
)
self.name = name
self.radius = radius
def __repr__(self):
return f"Circle(lon={self.lon},lat={self.lat},radius={self.radius},name={self.name})"
class CircleGeometries(GeoDataMixin, Query):
"""
This class will form the required geometries for a set of circles defined by
an iterable of Circle objects. This class is used by CircleRasterPops which calculates
population based on a raster.
Parameters
----------
circles : iterable of Circle
The circular regions for which geometries are required
Examples
--------
>>> cl = Circle(2, 3, 4, 'bob')
>>> c = CircleGeometries([cl])
>>> c.head()
geom name
"""
Calculates an event score for each event based
on a scoring dictionary.
"""
from typing import Dict, Union, Tuple, Optional
from typing import List
from ..utilities import EventsTablesUnion
from ...core import Query, location_joined_query, make_spatial_unit
from ...core.spatial_unit import AnySpatialUnit
from flowmachine.utils import standardise_date
class EventScore(Query):
"""
Represents an event score class.
This class assigns a score to each event based on the hour of the day and
the day of the week. The scores can be useful to cluster a set of events
based on its signature. Such type of analysis reduces the dimensionality of
the problem by projecting a given event pattern onto the real line.
This class returns a table with scores averaged across the requested spatial unit
per subscriber.
Parameters
----------
score_hour : list of float
A length 24 list containing numerical scores between -1 and 1, where entry 0 is midnight.
score_dow : dict
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import structlog
import warnings
from typing import List
from ...core import Query
from ...core.context import get_db
from ...core.errors import MissingDateError
from .event_table_subset import EventTableSubset
from flowmachine.utils import standardise_date
logger = structlog.get_logger("flowmachine.debug", submodule=__name__)
class EventsTablesUnion(Query):
"""
Takes a list of subtables, subsets each of them
by date and selects a specified list of columns
from the result and unions (i.e. appends) all
of these tables. This class is mostly used as an
intermediate for other classes.
Parameters
----------
start, stop : str
ISO-format date
columns :
list of columns to select
tables : str or list of strings, default 'all'
Can be a sting of a single table (with the schema)
or a list of these. The keyword all is to select all
locinfo.date_of_last_service,
{columns}
FROM
{self.location_info_table_fqn} AS locinfo
INNER JOIN
{self._get_subtable()}
ON ST_within(
locinfo.geom_point::geometry,
ST_SetSRID(polygon.{self.geom_col}, 4326)::geometry
)
"""
return tower_admins
class CellToAdmin(Query):
"""
Maps all cells (aka sites) to a admin region. This is a thin wrapper to
the more general class CellToPolygon, which assumes that you have
the standard set-up.
Parameters
----------
level : {'adminN'}
One of admin1, admin2 etc.
column_name : str, optional
Pass a string of the column to use as the
identifier of the admin region. By default
this will be admin*pcod. But you may wish
to use something else, such as admin3name.
"""
from .events_tables_union import EventsTablesUnion
from ...core import Query, make_spatial_unit
from ...core.context import get_db
from ...core.spatial_unit import AnySpatialUnit
from numpy import inf
import structlog
from flowmachine.utils import standardise_date
logger = structlog.get_logger("flowmachine.debug", submodule=__name__)
valid_subscriber_identifiers = ("msisdn", "imei", "imsi")
class UniqueSubscribers(Query):
"""
Class representing the set of all unique subscribers in our interactions
table.
Parameters
----------
start : str
iso format date range for the beginning of the time frame,
e.g. 2016-01-01 or 2016-01-01 14:03:01
stop : str
As above
hours : tuple of ints, default 'all'
Subset the result within certain hours, e.g. (4,17)
This will subset the query only with these hours, but
across all specified days. Or set to 'all' to include
all hours.