Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
table="all",
subscriber_identifier="msisdn",
ignore_nulls=True,
subscriber_subset=None,
):
"""
"""
if location == "any" and spatial_unit != make_spatial_unit("cell"):
raise ValueError(
"Invalid parameter combination: location='any' can only be used with cell spatial unit."
)
self.start = standardise_date(start)
self.stop = standardise_date(stop)
self.location = location
self.ul = SubscriberLocations(
self.start,
self.stop,
spatial_unit=spatial_unit,
hours=hours,
table=table,
subscriber_identifier=subscriber_identifier,
ignore_nulls=ignore_nulls,
subscriber_subset=subscriber_subset,
)
self.table = self.ul.table
self.subscriber_identifier = self.ul.subscriber_identifier
def __init__(
self,
start,
stop,
spatial_unit: AnySpatialUnit = make_spatial_unit("cell"),
hours="all",
table="all",
):
self.start = standardise_date(start)
self.stop = standardise_date(stop)
self.spatial_unit = spatial_unit
self.hours = hours
self.table = table
self.ul = UniqueLocations(
SubscriberLocations(
start=self.start,
stop=self.stop,
spatial_unit=self.spatial_unit,
hours=self.hours,
table=self.table,
)
)
super().__init__()
stop,
spatial_unit: Optional[AnySpatialUnit] = None,
hours="all",
table="all",
subscriber_identifier="msisdn",
*,
ignore_nulls=True,
subscriber_subset=None,
):
"""
"""
self.start = standardise_date(start)
self.stop = standardise_date(stop)
if spatial_unit is None:
self.spatial_unit = make_spatial_unit("admin", level=3)
else:
self.spatial_unit = spatial_unit
self.hours = hours
self.table = table
self.subscriber_identifier = subscriber_identifier
self.subscriber_locs = SubscriberLocations(
start=self.start,
stop=self.stop,
spatial_unit=self.spatial_unit,
hours=self.hours,
table=self.table,
subscriber_identifier=self.subscriber_identifier,
ignore_nulls=ignore_nulls,
subscriber_subset=subscriber_subset,
def __init__(
self,
start: str,
stop: str,
statistic: str = "avg",
*,
hours: Union[str, Tuple[int, int]] = "all",
tables: Union[str, List[str]] = "all",
subscriber_identifier: str = "msisdn",
subscriber_subset: Optional[Query] = None,
direction: Union[str, Direction] = Direction.OUT,
):
self.start = standardise_date(start)
self.stop = standardise_date(stop)
self.hours = hours
self.tables = tables
self.subscriber_identifier = subscriber_identifier
self.direction = Direction(direction)
column_list = [
self.subscriber_identifier,
"datetime",
*self.direction.required_columns,
]
self.statistic = statistic.lower()
if self.statistic not in valid_stats:
raise ValueError(
"{} is not a valid statistic. Use one of {}".format(
self.statistic, valid_stats
def __init__(
self,
start,
stop,
volume="total",
statistic="sum",
*,
subscriber_identifier="msisdn",
hours="all",
subscriber_subset=None,
):
self.start = standardise_date(start)
self.stop = standardise_date(stop)
self.subscriber_identifier = subscriber_identifier
self.hours = hours
self.volume = volume
self.statistic = statistic.lower()
self.tables = "events.mds"
if self.statistic not in valid_stats:
raise ValueError(
"{} is not a valid statistic. Use one of {}".format(
self.statistic, valid_stats
)
)
if self.volume not in {"total", "upload", "download"}:
raise ValueError(f"{self.volume} is not a valid volume.")
def __init__(
self,
start,
stop,
*,
hours="all",
tables="all",
subscriber_identifier="msisdn",
direction: Union[str, Direction] = Direction.BOTH,
exclude_self_calls=True,
subscriber_subset=None,
):
self.tables = tables
self.start = standardise_date(start)
self.stop = standardise_date(stop)
self.hours = hours
self.direction = Direction(direction)
self.subscriber_identifier = subscriber_identifier
self.exclude_self_calls = exclude_self_calls
self.tables = tables
column_list = [
self.subscriber_identifier,
"msisdn_counterpart",
*self.direction.required_columns,
]
self.unioned_query = EventsTablesUnion(
self.start,
self.stop,
columns=column_list,
columns,
tables=None,
hours="all",
subscriber_subset=None,
subscriber_identifier="msisdn",
):
"""
"""
if isinstance(tables, str) and tables.lower() == "all":
logger.warn(
"EventsTablesUnion will soon stop accepting the argument tables='all'. Use tables=None instead."
)
tables = None
self.start = standardise_date(start)
self.stop = standardise_date(stop)
self.columns = columns
self.tables = self._parse_tables(tables)
if "*" in columns and len(self.tables) != 1:
raise ValueError(
"Must give named tables when combining multiple event type tables."
)
self.date_subsets = self._make_table_list(
hours=hours,
subscriber_subset=subscriber_subset,
subscriber_identifier=subscriber_identifier,
)
super().__init__()
def __init__(
self,
start,
stop,
statistic="sum",
*,
subscriber_identifier="msisdn",
hours="all",
subscriber_subset=None,
):
self.start = standardise_date(start)
self.stop = standardise_date(stop)
self.subscriber_identifier = subscriber_identifier
self.hours = hours
self.statistic = statistic.lower()
self.tables = "events.topups"
if self.statistic not in valid_stats:
raise ValueError(
"{} is not a valid statistic. Use one of {}".format(
self.statistic, valid_stats
)
)
column_list = [
self.subscriber_identifier,
"datetime",
def __init__(
self,
start,
stop,
*,
hours="all",
tables="all",
subscriber_identifier="msisdn",
direction: Union[str, Direction] = Direction.BOTH,
exclude_self_calls=True,
subscriber_subset=None,
):
self.start = standardise_date(start)
self.stop = standardise_date(stop)
self.hours = hours
self.direction = Direction(direction)
self.subscriber_identifier = subscriber_identifier
self.exclude_self_calls = exclude_self_calls
self.tables = tables
column_list = [
self.subscriber_identifier,
"msisdn_counterpart",
*self.direction.required_columns,
]
self.unioned_query = EventsTablesUnion(
self.start,
self.stop,
def __init__(self, start_date, end_date):
self.start_date = parse_datestring(start_date)
self.end_date = parse_datestring(end_date)
self.start_date_as_str = standardise_date(start_date)
self.end_date_as_str = standardise_date(end_date)
self.one_day_past_end_date = self.end_date + dt.timedelta(days=1)
self.one_day_past_end_date_as_str = standardise_date(self.one_day_past_end_date)