Skip to content

flowmachine.utils

Source: flowmachine/utils.py

Various simple utilities.

Enum Statistic

Statistic
Source: flowmachine/utils.py

Valid statistics for use with postgres. Convert a column name to a stat by doing f"{statistic:column_name}". For columns defined as a variable, do f"{statistic:'{column_name_variable}'}"

Members

  • COUNT: count

  • SUM: sum

  • AVG: avg

  • MAX: max

  • MIN: min

  • MEDIAN: median

  • STDDEV: stddev

  • VARIANCE: variance

  • MODE: mode

convert_dict_keys_to_strings

convert_dict_keys_to_strings(input_dict)
Source: flowmachine/utils.py

Return a copy of the input dictionary where all keys are converted to strings. This is necessary if the dictionary needs to be valid JSON.

Parameters

  • input_dict: dict

    The input dictionary.

Returns

  • dict

    A copy of input_dict with all keys converted to strings.

get_dist_query_string

get_dist_query_string(*, lon1, lat1, lon2, lat2)
Source: flowmachine/utils.py

function for getting the distance query string between two lon-lat points.

get_name_and_alias

get_name_and_alias(column_name: str) -> Tuple[str]
Source: flowmachine/utils.py

Given a column name string, return the column name and alias (if there is one), or return the provided column name twice if there is no alias.

Returns

  • typing.Tuple[str]

Examples

get_name_and_alias("col AS alias")
  ('col', 'alias')
get_name_and_alias("col")
  ('col', 'col')
get_name_and_alias("table.col")
  ('table.col', 'col')
get_name_and_alias("table.col as alias")
  ('table.col', 'alias')

list_of_dates

list_of_dates(start, stop)
Source: flowmachine/utils.py

Parameters

  • start, stop: str

    yyyy-mm-dd format datestrings.

Returns

  • list of yyyy-mm-dd format datestrings between start and stop (inclusive)

make_where

make_where(condition) -> str
Source: flowmachine/utils.py

Create an sql where clause from a list of conditions.

Parameters

  • condition: list of str, str

    Where conditions to combine

Returns

  • str

    A where clause or empty string if all clauses are empty

Examples

make_where("")
""
make_where("NOT outgoing")
"WHERE NOT outgoing"
make_where(["NOT outgoing", "duration > 5")
"WHERE NOT outgoing AND duration > 5"

parse_datestring

parse_datestring(datestring: Union[str, datetime.datetime, datetime.date]) -> datetime.datetime
Source: flowmachine/utils.py

Parameters

  • datestring: typing.Union[str, datetime.datetime, datetime.date]

    ISO string date, or a datetime or date object

Returns

  • datetime.datetime

pretty_sql

pretty_sql(
    sql,
    compact_lists_margin=0,
    split_string_literals_threshold=0,
    special_functions=True,
    comma_at_eoln=True,
)
Source: flowmachine/utils.py

Prettify and validate the syntax of an SQL query, using pglast

Parameters

  • sql: str

    SQL to prettyify and validate

  • compact_lists_margin: int, default 0

    Use compact form for lists shorter than this

  • split_string_literals_threshold: int, default 0

    Split strings (in the sql) longer than this threshold

  • special_functions: bool, default True

    Translate some special functions to their more commonly used forms

  • comma_at_eoln

Returns

  • str

    The prettified string.

proj4string

proj4string(conn, crs=None)
Source: flowmachine/utils.py

Provide a proj4 string for the input, or by default return the wsg84 proj4 string.

Parameters

  • conn: Connection

    FlowMachine db connection to use to get the proj4 string

  • crs: str, int

    An integer EPSG code, or a proj4 string

Returns

  • str

    Proj4 string for the input crs.

sort_recursively

sort_recursively(d)
Source: flowmachine/utils.py

Helper function to sort a dictionary recursively (including any dicts or sequences inside it).

Parameters

  • d: dict

    Input dictionary (can be nested).

Returns

  • dict

    The sorted dictionary.

standardise_date

standardise_date(date: Union[str, datetime.date, datetime.datetime]) -> Union[str, NoneType]
Source: flowmachine/utils.py

Turn a date, datetime, or date string into a standardised date string (YYYY-MM-DD HH:MM:SS).

Parameters

  • date: typing.Union[str, datetime.date, datetime.datetime]

    Date-like-thing to standardise.

Returns

  • typing.Union[str, NoneType]

    YYYY-MM-DD HH:MM:SS formatted date string or None if input was None.

standardise_date_to_datetime

standardise_date_to_datetime(date: Union[str, datetime.date, datetime.datetime]) -> Union[datetime.datetime, NoneType]
Source: flowmachine/utils.py

As standardise_date but returns a datetime object or None if input was None Parameters ---------- date : str, date or datetime Date-like-thing to standardise.

Returns

  • typing.Union[datetime.datetime, NoneType]

time_period_add

time_period_add(date, n, unit="days")
Source: flowmachine/utils.py

Adds n days to the date (represented as a string). Or alternatively add hours or minutes to the date.

Parameters

  • date: str

    Date to add to in yyyy-mm-dd format

  • n: int

    Number of units to add

  • unit: str, default 'days'

    Type of unit to add to the date

Returns

  • str

    Altered date string

Examples

time_period_add('2016-01-01', 3)
'2016-01-04 00:00:00'

to_nested_list

to_nested_list(d)
Source: flowmachine/utils.py

Helper function to recursively convert an input dictionary to a nested list.

Parameters

  • d: dict

    Input dictionary (can be nested).

Returns

  • list of tuples