---
-name: test-linux
+name: cov-submit
"on":
push: {}
jobs:
- test-linux:
+ cov-submit:
runs-on: ubuntu-latest
steps:
- name: Install requirements
# NOTE: container lacks OS dist for testresources/launchpadlib
run: |
+ pip install coveralls==3.3.1
pip install testresources==2.0.1
- make _deps
+ pip install -r requirements.txt
+ pip install -r requirements-test.txt
# TODO: Tests for: python-argcomplete (tab-completion)
- name: Test
export NUTRA_HOME=$(pwd)/tests/.nutra.test
make _test
- - name: Lint
- run: make _lint
-
- - name: Install
- run: make install
-
- - name: Basic Tests / CLI / Integration
- run: |
- n -v
- nutra -d init -y
- nutra --no-pager nt
- nutra --no-pager sort -c 789
- nutra --no-pager search ultraviolet mushrooms
- nutra --no-pager anl 9050
- nutra --no-pager recipe
- nutra day tests/resources/day/human-test.csv
+ - name: Submit coverage report / coveralls
+ env:
+ GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ run: coveralls --service=github
jobs:
test-linux:
- runs-on: ubuntu-latest
+ runs-on: ubuntu-18.04
+
+ strategy:
+ matrix:
+ python-version: ["3.4.10", "3.6", "3.8", "3.10"]
steps:
- name: Checkout
with:
submodules: recursive
+ - name: Set up Python ${{ matrix.python-version }}
+ uses: actions/setup-python@v4
+ with:
+ python-version: ${{ matrix.python-version }}
+
- name: Reload Cache / pip
uses: actions/cache@v3
with:
path: ~/.cache/pip
- key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements*.txt') }}
+ key: ${{ runner.os }}-${{ matrix.python-version }}-pip-${{ hashFiles('requirements*.txt') }}
restore-keys: |
${{ runner.os }}-pip-
- name: Install requirements
# NOTE: container lacks OS dist for testresources/launchpadlib
run: |
- pip install testresources==2.0.1
- make _deps
-
- # TODO: Tests for: python-argcomplete (tab-completion)
- - name: Test
- run: |
- export NUTRA_HOME=$(pwd)/tests/.nutra.test
- make _test
+ # NOTE: This is the latest version on GitHub for Python 3.4
+ if [[ "$(python --version)" =~ "3.4." ]]; then
+ pip install colorama==0.4.1;
+ fi
- - name: Lint
- run: make _lint
+ pip install -r requirements.txt
- name: Install
+ env:
+ PY_SYS_INTERPRETER: python3
run: make install
- name: Basic Tests / CLI / Integration
push: {}
jobs:
- test-win32:
+ windows-latest:
runs-on: windows-latest
steps:
restore-keys: |
${{ runner.os }}-pip-
- - name: Install requirements
- run: |
- make _deps
-
- # TODO: Tests for: python-argcomplete (tab-completion)
- - name: Test
- run: |
- $curDir = (Get-Location).toString()
- $Env:NUTRA_HOME = $curDir + '/tests/.nutra.test'
- make _test
-
- - name: Lint
- run: make _lint
-
- name: Install
run: make install
+++ /dev/null
----
-dist: xenial
-os: ["linux"]
-language: python
-python:
- - 3.4
- - 3.5
- - 3.6
- - 3.8
- - 3.9
- - pypy3
-cache: pip
-git:
- submodules: true
-
-before_install:
- - >
- if [[ "$(python --version)" =~ "3.4." ]]; then
- grep -v skip_empty setup.cfg > setup.cfg2 && mv setup.cfg2 setup.cfg;
- fi
-
-install:
- - python -m pip install coveralls
- - python -m pip install colorama==0.4.1
- - python -m pip install coverage>=4.5.4
- - python -m pip install -r requirements.txt
- - python -m pip install -r requirements-test-win_xp-ubu1604.txt
-
-script:
- - coverage run -m pytest -v -s -p no:cacheprovider tests/
- - coverage report
- - if [[ "$(python --version)" =~ "3.6." && "$(which pypy)" == "" ]]; then coveralls; fi
REQ_TEST := requirements-test.txt
REQ_OLD := requirements-test-win_xp-ubu1604.txt
+PIP_OPT_ARGS ?=
+
.PHONY: _deps
_deps:
$(PIP) install wheel
- $(PIP) install -r requirements.txt
- - $(PIP) install -r $(REQ_OPT)
- - $(PIP) install -r $(REQ_LINT)
- - $(PIP) install -r $(REQ_TEST) || (echo "\r\nTEST REQs failed. Trying old version" && $(PIP) install -r $(REQ_OLD))
+ $(PIP) install $(PIP_OPT_ARGS) -r requirements.txt
+ - $(PIP) install $(PIP_OPT_ARGS) -r $(REQ_OPT)
+ - $(PIP) install $(PIP_OPT_ARGS) -r $(REQ_LINT)
+ - $(PIP) install $(PIP_OPT_ARGS) -r $(REQ_TEST) || \
+ echo "TEST REQs failed. Try with '--user' flag, or old version: $(PIP) install -r $(REQ_OLD)"
.PHONY: deps
deps: _venv _deps ## Install requirements
.PHONY: install
install: ## pip install nutra
$(PY_SYS_INTERPRETER) -m pip install wheel
- $(PY_SYS_INTERPRETER) -m pip install .
+ $(PY_SYS_INTERPRETER) -m pip install . || $(PY_SYS_INTERPRETER) -m pip install --user .
$(PY_SYS_INTERPRETER) -m pip show nutra
- $(PY_SYS_INTERPRETER) -c 'import shutil; print(shutil.which("nutra"));'
nutra -v
__url__ = "https://github.com/nutratech/cli"
# Sqlite target versions
-__db_target_nt__ = "0.0.5"
+__db_target_nt__ = "0.0.6"
__db_target_usda__ = "0.0.8"
USDA_XZ_SHA256 = "25dba8428ced42d646bec704981d3a95dc7943240254e884aad37d59eee9616a"
NTSQLITE_BUILDPATH = os.path.join(ROOT_DIR, "ntsqlite", "sql", NT_DB_NAME)
NTSQLITE_DESTINATION = os.path.join(NUTRA_HOME, NT_DB_NAME)
-print(NTSQLITE_BUILDPATH)
-print(NTSQLITE_DESTINATION)
# Check Python version
PY_MIN_VER = (3, 4, 0)
# Run function
if args_dict:
- return parser.func(args=parser)
- return parser.func()
+ # Make sure the parser.func() always returns: Tuple[Int, Any]
+ return parser.func(args=parser) # type: ignore
+ return parser.func() # type: ignore
# Otherwise print help
arg_parser.print_help()
raise
finally:
if DEBUG:
- exc_time = time.time() - start_time
+ exc_time = time.time() - start_time # type: ignore
print("\nExecuted in: %s ms" % round(exc_time * 1000, 1))
print("Exit code: %s" % exit_code)
"""Main module for things related to argparse"""
+import argparse
from ntclient.argparser import funcs as parser_funcs
from ntclient.argparser import types
-def build_subcommands(subparsers) -> None:
+def build_subcommands(subparsers: argparse._SubParsersAction) -> None:
"""Attaches subcommands to main parser"""
build_init_subcommand(subparsers)
build_nt_subcommand(subparsers)
################################################################################
# Methods to build subparsers, and attach back to main arg_parser
################################################################################
-def build_init_subcommand(subparsers) -> None:
+def build_init_subcommand(subparsers: argparse._SubParsersAction) -> None:
"""Self running init command"""
init_parser = subparsers.add_parser(
"init", help="setup profiles, USDA and NT database"
init_parser.set_defaults(func=parser_funcs.init)
-def build_nt_subcommand(subparsers) -> None:
+def build_nt_subcommand(subparsers: argparse._SubParsersAction) -> None:
"""Lists out nutrients details with computed totals and averages"""
nutrient_parser = subparsers.add_parser(
"nt", help="list out nutrients and their info"
nutrient_parser.set_defaults(func=parser_funcs.nutrients)
-def build_search_subcommand(subparsers) -> None:
+def build_search_subcommand(subparsers: argparse._SubParsersAction) -> None:
"""Search: terms [terms ... ]"""
search_parser = subparsers.add_parser(
"search", help="search foods by name, list overview info"
search_parser.set_defaults(func=parser_funcs.search)
-def build_sort_subcommand(subparsers) -> None:
+def build_sort_subcommand(subparsers: argparse._SubParsersAction) -> None:
"""Sort foods ranked by nutr_id, per 100g or 200kcal"""
sort_parser = subparsers.add_parser("sort", help="sort foods by nutrient ID")
sort_parser.add_argument(
sort_parser.set_defaults(func=parser_funcs.sort)
-def build_analyze_subcommand(subparsers) -> None:
+def build_analyze_subcommand(subparsers: argparse._SubParsersAction) -> None:
"""Analyzes (foods only for now)"""
analyze_parser = subparsers.add_parser("anl", help="analyze food(s)")
analyze_parser.add_argument(
analyze_parser.set_defaults(func=parser_funcs.analyze)
-def build_day_subcommand(subparsers) -> None:
+def build_day_subcommand(subparsers: argparse._SubParsersAction) -> None:
"""Analyzes a DAY.csv, uses new colored progress bar spec"""
day_parser = subparsers.add_parser(
"day", help="analyze a DAY.csv file, RDAs optional"
day_parser.set_defaults(func=parser_funcs.day)
-def build_recipe_subcommand(subparsers) -> None:
+def build_recipe_subcommand(subparsers: argparse._SubParsersAction) -> None:
"""View, add, edit, delete recipes"""
recipe_parser = subparsers.add_parser("recipe", help="list and analyze recipes")
recipe_subparsers = recipe_parser.add_subparsers(title="recipe subcommands")
"""Current home to subparsers and service-level logic"""
+import argparse
import os
from ntclient import services
-def init(args):
+def init(args: argparse.Namespace) -> tuple:
"""Wrapper init method for persistence stuff"""
return services.init(yes=args.yes)
################################################################################
# Nutrients, search and sort
################################################################################
-def nutrients():
+def nutrients(): # type: ignore
"""List nutrients"""
return services.usda.list_nutrients()
-def search(args):
- """Searches all dbs, foods, recipes, recents and favorites."""
+def search(args: argparse.Namespace) -> tuple:
+ """Searches all dbs, foods, recipes, recent items and favorites."""
if args.top:
return services.usda.search(
words=args.terms, fdgrp_id=args.fdgrp_id, limit=args.top
return services.usda.search(words=args.terms, fdgrp_id=args.fdgrp_id)
-def sort(args):
+def sort(args: argparse.Namespace) -> tuple:
"""Sorts based on nutrient id"""
if args.top:
return services.usda.sort_foods(args.nutr_id, by_kcal=args.kcal, limit=args.top)
################################################################################
# Analysis and Day scoring
################################################################################
-def analyze(args):
+def analyze(args: argparse.Namespace) -> tuple:
"""Analyze a food"""
food_ids = args.food_id
grams = args.grams
return services.analyze.foods_analyze(food_ids, grams)
-def day(args):
+def day(args: argparse.Namespace) -> tuple:
"""Analyze a day's worth of meals"""
day_csv_paths = args.food_log
day_csv_paths = [os.path.expanduser(x) for x in day_csv_paths]
################################################################################
# Recipes
################################################################################
-def recipes():
+def recipes() -> tuple:
"""Return recipes"""
return services.recipe.recipes_overview()
-def recipe(args):
+def recipe(args: argparse.Namespace) -> tuple:
"""Return recipe view (analysis)"""
return services.recipe.recipe_overview(args.recipe_id)
-def recipe_import(args):
+def recipe_import(args: argparse.Namespace) -> tuple:
"""Add a recipe"""
# TODO: custom serving sizes, not always in grams?
return services.recipe.recipe_import(args.path)
-def recipe_delete(args):
+def recipe_delete(args: argparse.Namespace) -> tuple:
"""Delete a recipe"""
return services.recipe.recipe_delete(args.recipe_id)
import os
-def file_path(string):
+def file_path(path_in: str) -> str:
"""Returns file if it exists, else raises argparse error"""
- if os.path.isfile(string):
- return string
- raise argparse.ArgumentTypeError('FileNotFoundError: "%s"' % string)
+ if os.path.isfile(path_in):
+ return path_in
+ raise argparse.ArgumentTypeError('FileNotFoundError: "%s"' % path_in)
-def file_or_dir_path(string):
+def file_or_dir_path(path_in: str) -> str:
"""Returns path if it exists, else raises argparse error"""
- if os.path.exists(string):
- return string
- raise argparse.ArgumentTypeError('FileNotFoundError: "%s"' % string)
+ if os.path.exists(path_in):
+ return path_in
+ raise argparse.ArgumentTypeError('FileNotFoundError: "%s"' % path_in)
"""Temporary [wip] module for more visual (& colorful) RDA output"""
-def nutprogbar(food_amts, food_analyses, nutrients):
+def nutprogbar(food_amts: dict, food_analyses: list, nutrients: dict) -> dict:
"""Returns progress bars, colorized, for foods analyses"""
- def tally():
+ def tally() -> None:
for nut in nut_percs:
# TODO: get RDA values from nt DB, tree node nested organization
print(nut)
- food_analyses = {
+ food_analyses_dict = {
x[0]: {y[1]: y[2] for y in food_analyses if y[0] == x[0]} for x in food_analyses
}
for food_id, grams in food_amts.items():
# r = grams / 100.0
- analysis = food_analyses[food_id]
+ analysis = food_analyses_dict[food_id]
for nutrient_id, amt in analysis.items():
if nutrient_id not in nut_amts:
nut_amts[nutrient_id] = amt
-Subproject commit 47c6db2a418db690ba1b8e407756240ec2f296f3
+Subproject commit f76955d55293d166f833befbab2c52447d0cd07e
"""Main SQL persistence module, shared between USDA and NT databases"""
import sqlite3
+from collections.abc import Sequence
# ------------------------------------------------
def sql_entries(sql_result: sqlite3.Cursor) -> list:
"""Formats and returns a `sql_result()` for console digestion and output"""
# TODO: return object: metadata, command, status, errors, etc?
- rows = sql_result.fetchall()
+ rows = sql_result.fetchall()
return rows
cur = con.cursor()
result = cur.execute("SELECT * FROM version;").fetchall()
+
close_con_and_cur(con, cur, commit=False)
- return result[-1][1]
+ return str(result[-1][1])
def close_con_and_cur(
- con: sqlite3.Connection, cur: sqlite3.Cursor, commit=True
+ con: sqlite3.Connection, cur: sqlite3.Cursor, commit: bool = True
) -> None:
"""Cleans up, commits, and closes after an SQL command is run"""
# Main query methods
# ------------------------------------------------
def _prep_query(
- con: sqlite3.Connection, query: str, db_name: str, values=None
-) -> tuple:
- """@param values: tuple | list"""
+ con: sqlite3.Connection, query: str, db_name: str, values: Sequence = ()
+) -> sqlite3.Cursor:
+ """
+ Run a query and return a cursor object ready for row extraction.
+ @param con: sqlite3.Connection object
+ @param query: query string, e.g. SELECT * FROM version;
+ @param db_name: (nt | usda) database name [TODO: enum]
+ @param values: (tuple | list)
+ empty for bare queries, tuple for single, and list for many
+ @return: A sqlite3.Cursor object with populated return values.
+ """
from ntclient import DEBUG # pylint: disable=import-outside-toplevel
# TODO: separate `entry` & `entries` entity for single vs. bulk insert?
if values:
if isinstance(values, list):
- result = cur.executemany(query, values)
+ cur.executemany(query, values)
else: # tuple
- result = cur.execute(query, values)
+ cur.execute(query, values)
else:
- result = cur.execute(query)
+ cur.execute(query)
- return cur, result
+ return cur
def _sql(
con: sqlite3.Connection,
query: str,
db_name: str,
- values=None,
+ values: Sequence = (),
) -> list:
"""@param values: tuple | list"""
- cur, result = _prep_query(con, query, db_name, values)
+ cur = _prep_query(con, query, db_name, values)
# TODO: print "<number> SELECTED", or other info
# BASED ON command SELECT/INSERT/DELETE/UPDATE
con: sqlite3.Connection,
query: str,
db_name: str,
- values=None,
+ values: Sequence = (),
) -> tuple:
"""@param values: tuple | list"""
- cur, result = _prep_query(con, query, db_name, values)
+ cur = _prep_query(con, query, db_name, values)
- result = sql_entries_headers(result)
+ result = sql_entries_headers(cur)
close_con_and_cur(con, cur)
return result
"""Nutratracker DB specific sqlite module"""
import os
import sqlite3
+from collections.abc import Sequence
from ntclient import (
NT_DB_NAME,
def nt_ver() -> str:
"""Gets version string for nt.sqlite3 database"""
+
con = nt_sqlite_connect(version_check=False)
return version(con)
# TODO: is this logic (and these error messages) the best?
# what if .isdir() == True ? Fails with stacktrace?
os.rename(NTSQLITE_BUILDPATH, NTSQLITE_DESTINATION)
- if not nt_ver() == __db_target_nt__:
+ if nt_ver() != __db_target_nt__:
raise SqlInvalidVersionError(
"ERROR: nt target [{0}] mismatch actual [{1}], ".format(
__db_target_nt__, nt_ver()
# ------------------------------------------------
# SQL connection & utility methods
# ------------------------------------------------
-
-
-def nt_sqlite_connect(version_check=True) -> sqlite3.Connection:
+def nt_sqlite_connect(version_check: bool = True) -> sqlite3.Connection:
"""Connects to the nt.sqlite3 file, or throws an exception"""
+
db_path = os.path.join(NUTRA_HOME, NT_DB_NAME)
if os.path.isfile(db_path):
con = sqlite3.connect(db_path)
raise SqlConnectError("ERROR: nt database doesn't exist, please run `nutra init`")
-def sql(query, values=None) -> list:
+def sql(query: str, values: Sequence = ()) -> list:
"""Executes a SQL command to nt.sqlite3"""
+
con = nt_sqlite_connect()
return _sql(con, query, db_name="nt", values=values)
-def sql_headers(query, values=None) -> tuple:
+def sql_headers(query: str, values: Sequence = ()) -> tuple:
"""Executes a SQL command to nt.sqlite3"""
+
con = nt_sqlite_connect()
return _sql_headers(con, query, db_name="nt", values=values)
from ntclient.persistence.sql.nt import sql, sql_headers
-def sql_nt_next_index(table=None):
+def sql_nt_next_index(table: str) -> int:
"""Used for previewing inserts"""
+ # noinspection SqlResolve
query = "SELECT MAX(id) as max_id FROM %s;" % table # nosec: B608
return int(sql(query)[0]["max_id"])
################################################################################
# Recipe functions
################################################################################
-def sql_recipe(recipe_id):
+def sql_recipe(recipe_id: int) -> list:
"""Selects columns for recipe_id"""
- query = "SELECT * FROM recipes WHERE id=?;"
+ query = "SELECT * FROM recipe WHERE id=?;"
return sql(query, values=(recipe_id,))
-def sql_recipes():
+def sql_recipes() -> tuple:
"""Show recipes with selected details"""
query = """
SELECT
return sql_headers(query)
-def sql_analyze_recipe(recipe_id):
+def sql_analyze_recipe(recipe_id: int) -> list:
"""Output (nutrient) analysis columns for a given recipe_id"""
query = """
SELECT
return sql(query, values=(recipe_id,))
-def sql_recipe_add():
+def sql_recipe_add() -> list:
"""TODO: method for adding recipe"""
query = """
"""
import sqlite3
import tarfile
import urllib.request
+from collections.abc import Sequence
from ntclient import NUTRA_HOME, USDA_DB_NAME, __db_target_usda__
from ntclient.persistence.sql import _sql, _sql_headers, version
from ntclient.utils.exceptions import SqlConnectError, SqlInvalidVersionError
-def usda_init(yes=False) -> None:
+def usda_init(yes: bool = False) -> None:
"""On-boarding function. Downloads tarball and unpacks usda.sqlite3 file"""
def input_agree() -> str:
"""Download USDA tarball from BitBucket and extract to storage folder"""
if yes or input_agree().lower() == "y":
- # TODO: save with version in filename? Don't re-download tarball, just extract?
+ # TODO: save with version in filename?
+ # Don't re-download tarball, just extract?
save_path = os.path.join(NUTRA_HOME, "%s.tar.xz" % USDA_DB_NAME)
# Download usda.sqlite3.tar.xz
print("==> done downloading %s" % USDA_DB_NAME)
- # TODO: handle resource moved on Bitbucket or version mismatch due to manual overwrite?
+ # TODO: handle resource moved on Bitbucket
+ # or version mismatch due to manual overwrite?
url = (
"https://bitbucket.org/dasheenster/nutra-utils/downloads/{0}-{1}.tar.xz".format(
USDA_DB_NAME, __db_target_usda__
"INFO: usda.sqlite3 target [{0}] doesn't match actual [{1}], ".format(
__db_target_usda__, usda_ver()
)
- + "static resource (no user data lost).. downloading and extracting correct version"
+ + "static resource (no user data lost).. "
+ "downloading and extracting correct version"
)
download_extract_usda()
)
-def usda_sqlite_connect(version_check=True) -> sqlite3.Connection:
+def usda_sqlite_connect(version_check: bool = True) -> sqlite3.Connection:
"""Connects to the usda.sqlite3 file, or throws an exception"""
# TODO: support as customizable env var ?
db_path = os.path.join(NUTRA_HOME, USDA_DB_NAME)
if os.path.isfile(db_path):
con = sqlite3.connect(db_path)
- # con.row_factory = sqlite3.Row # see: https://chrisostrouchov.com/post/python_sqlite/
+ # con.row_factory = sqlite3.Row # see:
+ # https://chrisostrouchov.com/post/python_sqlite/
# Verify version
if version_check and usda_ver() != __db_target_usda__:
return version(con)
-def sql(query, values=None, version_check=True) -> list:
- """Executes a SQL command to usda.sqlite3"""
+def sql(query: str, values: Sequence = (), version_check: bool = True) -> list:
+ """
+ Executes a SQL command to usda.sqlite3
+
+ @param query: Input SQL query
+ @param values: Union[tuple, list] Leave as empty tuple for no values,
+ e.g. bare query. Populate a tuple for a single insert. And use a list for
+ cur.executemany()
+ @param version_check: Ignore mismatch version, useful for "meta" commands
+ @return: List of selected SQL items
+ """
con = usda_sqlite_connect(version_check=version_check)
return _sql(con, query, db_name="usda", values=values)
-def sql_headers(query, values=None, version_check=True) -> tuple:
- """Executes a SQL command to usda.sqlite3 [WITH HEADERS]"""
+def sql_headers(query: str, values: Sequence = (), version_check: bool = True) -> tuple:
+ """
+ Executes a SQL command to usda.sqlite3 [WITH HEADERS]
+
+ @param query: Input SQL query
+ @param values: Union[tuple, list] Leave as empty tuple for no values,
+ e.g. bare query. Populate a tuple for a single insert. And use a list for
+ cur.executemany()
+ @param version_check: Ignore mismatch version, useful for "meta" commands
+ @return: List of selected SQL items
+ """
con = usda_sqlite_connect(version_check=version_check)
################################################################################
# Basic functions
################################################################################
-def sql_fdgrp():
+def sql_fdgrp() -> dict:
"""Shows food groups"""
query = "SELECT * FROM fdgrp;"
return {x[0]: x for x in result}
-def sql_food_details(food_ids=None) -> list:
+def sql_food_details(_food_ids: set = None) -> list:
"""Readable human details for foods"""
- if food_ids is None:
+ if not _food_ids:
query = "SELECT * FROM food_des;"
else:
# TODO: does sqlite3 driver support this? cursor.executemany() ?
query = "SELECT * FROM food_des WHERE id IN (%s);"
- food_ids = ",".join(str(x) for x in set(food_ids))
+ food_ids = ",".join(str(x) for x in set(_food_ids))
query = query % food_ids
return sql(query)
return sql_headers(query)
-def sql_servings(food_ids) -> list:
+def sql_servings(_food_ids: set) -> list:
"""Food servings"""
# TODO: apply connective logic from `sort_foods()` IS ('None') ?
query = """
WHERE
serv.food_id IN (%s);
"""
- food_ids = ",".join(str(x) for x in set(food_ids))
+ # FIXME: support this kind of thing by library code & parameterized queries
+ food_ids = ",".join(str(x) for x in set(_food_ids))
return sql(query % food_ids)
-def sql_analyze_foods(food_ids) -> list:
+def sql_analyze_foods(food_ids: set) -> list:
"""Nutrient analysis for foods"""
query = """
SELECT
food_des.id IN (%s);
"""
# TODO: parameterized queries
- food_ids = ",".join(str(x) for x in set(food_ids))
- return sql(query % food_ids)
+ food_ids_concat = ",".join(str(x) for x in set(food_ids))
+ return sql(query % food_ids_concat)
################################################################################
# Sort
################################################################################
-def sql_sort_helper1(nutrient_id) -> list:
+def sql_sort_helper1(nutrient_id: int) -> list:
"""Selects relevant bits from nut_data for sorting"""
query = """
return sql(query % (NUTR_ID_KCAL, nutrient_id))
-def sql_sort_foods(nutr_id) -> list:
+def sql_sort_foods(nutr_id: int) -> list:
"""Sort foods by nutr_id per 100 g"""
query = """
return sql(query % nutr_id)
-def sql_sort_foods_by_kcal(nutr_id) -> list:
+def sql_sort_foods_by_kcal(nutr_id: int) -> list:
"""Sort foods by nutr_id per 200 kcal"""
# TODO: use parameterized queries
from ntclient.services import analyze, recipe, usda
-def init(yes=False):
+def init(yes: bool = False) -> tuple:
"""
+ Main init method for downloading USDA and creating NT databases.
TODO: Check for:
1. .nutra folder
2. usda
3a. nt
3b. default profile?
4. prefs.json
+
+ @param yes: bool (Skip prompting for [Y/n] in stdin)
+ @return: tuple[int, bool]
"""
+
print("Nutra directory ", end="")
if not os.path.isdir(NUTRA_HOME):
os.makedirs(NUTRA_HOME, 0o755)
################################################################################
# Foods
################################################################################
-def foods_analyze(food_ids, grams=None):
+def foods_analyze(food_ids: set, grams: int = 0) -> tuple:
"""
Analyze a list of food_ids against stock RDA values
TODO: from ntclient.utils.nutprogbar import nutprogbar
analyses = {}
for analysis in raw_analyses:
food_id = analysis[0]
- if grams is not None:
+ if grams:
anl = (analysis[1], round(analysis[2] * grams / 100, 2))
else:
anl = (analysis[1], analysis[2])
analyses[food_id].append(anl)
serving = sql_servings(food_ids)
- food_des = sql_food_details(food_ids)
- food_des = {x[0]: x for x in food_des}
+ food_des_rows = sql_food_details(food_ids)
+ food_des = {x[0]: x for x in food_des_rows}
nutrients = sql_nutrients_overview()
rdas = {x[0]: x[1] for x in nutrients.values()}
################################################################################
# Day
################################################################################
-def day_analyze(day_csv_paths, rda_csv_path=None):
+def day_analyze(day_csv_paths: str, rda_csv_path: str = str()) -> tuple:
"""Analyze a day optionally with custom RDAs,
e.g. nutra day ~/.nutra/rocky.csv -r ~/.nutra/dog-rdas-18lbs.csv
TODO: Should be a subset of foods_analyze
"""
from ntclient import DEBUG # pylint: disable=import-outside-toplevel
- if rda_csv_path is not None:
+ if rda_csv_path:
with open(rda_csv_path, encoding="utf-8") as file_path:
rda_csv_input = csv.DictReader(
row for row in file_path if not row.startswith("#")
logs.append(log)
# Inject user RDAs
- nutrients = [list(x) for x in sql_nutrients_overview().values()]
+ nutrients_lists = [list(x) for x in sql_nutrients_overview().values()]
for rda in rdas:
nutrient_id = int(rda["id"])
_rda = float(rda["rda"])
- for nutrient in nutrients:
- if nutrient[0] == nutrient_id:
- nutrient[1] = _rda
+ for _nutrient in nutrients_lists:
+ if _nutrient[0] == nutrient_id:
+ _nutrient[1] = _rda
if DEBUG:
- substr = "{0} {1}".format(_rda, nutrient[2]).ljust(12)
- print("INJECT RDA: {0} --> {1}".format(substr, nutrient[4]))
- nutrients = {x[0]: x for x in nutrients}
+ substr = "{0} {1}".format(_rda, _nutrient[2]).ljust(12)
+ print("INJECT RDA: {0} --> {1}".format(substr, _nutrient[4]))
+ nutrients = {x[0]: x for x in nutrients_lists}
# Analyze foods
foods_analysis = {}
if entry["id"]:
food_id = int(entry["id"])
grams = float(entry["grams"])
- for nutrient in foods_analysis[food_id]:
- nutr_id = nutrient[0]
- nutr_per_100g = nutrient[1]
+ for _nutrient2 in foods_analysis[food_id]:
+ nutr_id = _nutrient2[0]
+ nutr_per_100g = _nutrient2[1]
nutr_val = grams / 100 * nutr_per_100g
if nutr_id not in nutrient_totals:
nutrient_totals[nutr_id] = nutr_val
return 0, nutrients_totals
-def day_format(analysis, nutrients, buffer=None):
+# TODO: why not this...? nutrients: Mapping[int, tuple]
+def day_format(analysis: dict, nutrients: dict, buffer: int = 0) -> None:
"""Formats day analysis for printing to console"""
- def print_header(header):
+ def print_header(header: str) -> None:
print(Fore.CYAN, end="")
print("~~~~~~~~~~~~~~~~~~~~~~~~~~~")
print("--> %s" % header)
print("~~~~~~~~~~~~~~~~~~~~~~~~~~~")
print(Style.RESET_ALL)
- def print_macro_bar(_fat, _net_carb, _pro, _kcals_max, _buffer=None):
+ def print_macro_bar(
+ _fat: float, _net_carb: float, _pro: float, _kcals_max: float, _buffer: int = 0
+ ) -> None:
_kcals = fat * 9 + net_carb * 4 + _pro * 4
p_fat = (_fat * 9) / _kcals
+ Style.RESET_ALL
)
- def print_nute_bar(_n_id, amount, _nutrients):
+ def print_nute_bar(_n_id: int, amount: float, _nutrients: dict) -> tuple:
nutrient = _nutrients[_n_id]
rda = nutrient[1]
tag = nutrient[3]
print_nute_bar(n_id, analysis[n_id], nutrients)
# TODO: below
print(
- "work in progress...some minor fields with negligible data, they are not shown here"
+ "work in progress... "
+ "some minor fields with negligible data, they are not shown here"
)
)
-def recipes_overview():
+def recipes_overview() -> tuple:
"""Shows overview for all recipes"""
recipes = sql_recipes()[1]
return 0, results
-def recipe_overview(recipe_id):
+def recipe_overview(recipe_id: int) -> tuple:
"""Shows single recipe overview"""
recipe = sql_analyze_recipe(recipe_id)
name = recipe[0][1]
print(name)
- food_ids = {x[2]: x[3] for x in recipe}
- food_names = {x[0]: x[3] for x in sql_food_details(food_ids.keys())}
- food_analyses = sql_analyze_foods(food_ids.keys())
+ food_ids_dict = {x[2]: x[3] for x in recipe}
+ food_ids = set(food_ids_dict.keys())
+ food_names = {x[0]: x[3] for x in sql_food_details(food_ids)}
+ food_analyses = sql_analyze_foods(food_ids)
table = tabulate(
- [[food_names[food_id], grams] for food_id, grams in food_ids.items()],
+ [[food_names[food_id], grams] for food_id, grams in food_ids_dict.items()],
headers=["food", "g"],
)
print(table)
# tabulate nutrient RDA %s
nutrients = sql_nutrients_overview()
# rdas = {x[0]: x[1] for x in nutrients.values()}
- progbars = nutprogbar(food_ids, food_analyses, nutrients)
+ progbars = nutprogbar(food_ids_dict, food_analyses, nutrients)
print(progbars)
return 0, recipe
-def recipe_import(file_path):
+def recipe_import(file_path: str) -> tuple:
"""Import a recipe to SQL database"""
- def extract_id_from_filename(path):
+ def extract_id_from_filename(path: str) -> int:
filename = str(os.path.basename(path))
if (
"[" in filename
):
# TODO: try, raise: print/warn
return int(filename.split("[")[1].split("]")[0])
- return None
+ return 0 # zero is falsy
if os.path.isfile(file_path):
# TODO: better logic than this
return 1, False
-def recipe_add(name, food_amts):
+def recipe_add(name: str, food_amts: dict) -> tuple:
"""Add a recipe to SQL database"""
print()
print("New recipe: " + name + "\n")
- food_names = {x[0]: x[2] for x in sql_food_details(food_amts.keys())}
+ food_ids = set(food_amts.keys())
+ food_names = {x[0]: x[2] for x in sql_food_details(food_ids)}
results = []
for food_id, grams in food_amts.items():
return 1, False
-def recipe_delete(recipe_id):
+def recipe_delete(recipe_id: int) -> tuple:
"""Deletes recipe by ID, along with any FK constraints"""
recipe = sql_recipe(recipe_id)[0]
from ntclient.utils import NUTR_ID_KCAL, NUTR_IDS_AMINOS, NUTR_IDS_FLAVONES
-def list_nutrients():
+def list_nutrients() -> tuple:
"""Lists out nutrients with basic details"""
+
from ntclient import PAGING # pylint: disable=import-outside-toplevel
headers, nutrients = sql_nutrients_details()
################################################################################
# Sort
################################################################################
-def sort_foods(nutrient_id, by_kcal, limit=DEFAULT_RESULT_LIMIT):
+def sort_foods(
+ nutrient_id: int, by_kcal: bool, limit: int = DEFAULT_RESULT_LIMIT
+) -> tuple:
"""Sort, by nutrient, either (amount / 100 g) or (amount / 200 kcal)"""
# TODO: sub shrt_desc for long if available, and support config.FOOD_NAME_TRUNC
- def print_results(_results, _nutrient_id):
+ def print_results(_results: list, _nutrient_id: int) -> list:
"""Prints truncated list for sort"""
nutrients = sql_nutrients_overview()
nutrient = nutrients[_nutrient_id]
foods,
)
)
- foods.sort(key=lambda x: x[1], reverse=True)
+ # Sort by nutr_val
+ foods.sort(key=lambda x: int(x[1]), reverse=True)
foods = foods[:limit]
- food_ids = {x[0] for x in foods}
# Gets fdgrp and long_desc
+ food_ids = {x[0] for x in foods}
food_des = {x[0]: x for x in sql_food_details(food_ids)}
for food in foods:
food_id = food[0]
################################################################################
# Search
################################################################################
-def search(words, fdgrp_id=None, limit=DEFAULT_RESULT_LIMIT):
+def search(words: list, fdgrp_id: int = 0, limit: int = DEFAULT_RESULT_LIMIT) -> tuple:
"""Searches foods for input"""
- def tabulate_search(_results):
+ def tabulate_search(_results: list) -> list:
"""Makes search results more readable"""
# Current terminal size
# TODO: display "nonzero/total" report nutrients, aminos, and flavones..
from fuzzywuzzy import fuzz # pylint: disable=import-outside-toplevel
food_des = sql_food_details()
- if fdgrp_id is not None:
+ if fdgrp_id:
food_des = list(filter(lambda x: x[1] == fdgrp_id, food_des))
query = " ".join(words)
- scores = {f[0]: fuzz.token_set_ratio(query, f[2]) for f in food_des}
- scores = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:limit]
+ _scores = {f[0]: fuzz.token_set_ratio(query, f[2]) for f in food_des}
+ # NOTE: fuzzywuzzy reports score as an int, not float
+ scores = sorted(_scores.items(), key=lambda x: int(x[1]), reverse=True)
+ scores = scores[:limit]
food_ids = {x[0] for x in scores}
nut_data = sql_analyze_foods(food_ids)
else:
foods_nutrients[food_id][nutr_id] = nutr_val
- def search_results(_scores):
- """Generates search results, consumable by tabulate"""
+ def search_results(_scores: list) -> list:
+ """
+ Generates search results, consumable by tabulate
+
+ @param _scores: List[tuple]
+ @return: List[dict]
+ """
_results = []
for score in _scores:
_food_id = score[0]
score = score[1]
- food = food_des[_food_id]
+ food = food_des_dict[_food_id]
_fdgrp_id = food[1]
long_desc = food[2]
shrt_desc = food[3]
result = {
"food_id": _food_id,
"fdgrp_id": _fdgrp_id,
- # TODO: get more details from another function, maybe enhance food_details() ?
- # is that useful tho?
+ # TODO: get more details from another function,
+ # maybe enhance food_details() ? Is that useful tho?
# "fdgrp_desc": cache.fdgrp[fdgrp_id]["fdgrp_desc"],
# "data_src": cache.data_src[data_src_id]["name"],
"long_desc": shrt_desc if shrt_desc else long_desc,
return _results
# TODO: include C/F/P macro ratios as column?
- food_des = {f[0]: f for f in food_des}
+ # TODO: is this defined in the best place? It's accessed once in a helper function.
+ food_des_dict = {f[0]: f for f in food_des}
results = search_results(scores)
tabulate_search(results)
-autopep8~=1.6
-bandit~=1.7
-black~=22.3
-doc8~=0.11
-flake8~=4.0
-mypy~=0.960
-pylint~=2.13
-yamllint~=1.27
+autopep8>=1.6
+bandit>=1.7
+black>=22.3
+doc8>=0.11
+flake8>=4.0
+mypy>=0.960
+pylint>=2.13
+types-tabulate>=0.8.11
+yamllint>=1.27
-coverage~=6.0
-pytest~=7.0
+coverage>=6.0
+pytest>=7.0
skip_covered = True
+
[pycodestyle]
max-line-length = 88
+
[flake8]
per-file-ignores =
# Allow unused imports in __init__.py files
W503, # line break before binary operator
+
[isort]
line_length=88
known_first_party=ntclient
include_trailing_comma=true
+
[mypy]
-ignore_missing_imports = True
show_error_codes = True
-disable_error_code = import
+;show_error_context = True
+;pretty = True
+
+disallow_incomplete_defs = True
+disallow_untyped_defs = True
disallow_untyped_calls = True
disallow_untyped_decorators = True
-strict_optional = False
+
+warn_return_any = True
warn_redundant_casts = True
+warn_unreachable = True
+
warn_unused_ignores = True
+warn_unused_configs = True
+warn_incomplete_stub = True
+
+# Our test, they don't return a value typically
+[mypy-tests.*]
+disallow_untyped_defs = False
+
+# Our "sql" package, in ntclient/ntsqlite
+[mypy-sql]
+ignore_missing_imports = True
+
+# 3rd party packages missing types
+[mypy-argcomplete,colorama,coverage,fuzzywuzzy,psycopg2.*,setuptools]
+ignore_missing_imports = True
+
from ntclient.core import nutprogbar
from ntclient.ntsqlite.sql import build_ntsqlite
from ntclient.persistence.sql.nt import nt_ver
+from ntclient.persistence.sql.nt.funcs import sql_nt_next_index
from ntclient.persistence.sql.usda import funcs as usda_funcs
from ntclient.persistence.sql.usda import sql as _usda_sql
from ntclient.persistence.sql.usda import usda_ver
version = nt_ver()
assert version == __db_target_nt__
+ next_index = sql_nt_next_index("bf_eq")
+ assert next_index
+
# TODO: add more tests, it used to poll biometrics
projects / nutratech / cli.git / commitdiff