COVID-19 Community Level Query Tool

ccl.py

#!/usr/bin/env python3
# encoding=utf-8
# vim: set nobomb :

from enum import IntEnum, unique
from dataclasses import dataclass, field
from typing import Optional, Tuple
from datetime import datetime, timedelta
import json
from argparse import ArgumentParser
from locale import atof, setlocale, LC_NUMERIC
from itertools import starmap
from functools import partial


try:
  from requests import Request, Session
  use_rlib = True
except ImportError:
  from urllib.request import urlopen, Request
  use_rlib = False


@unique
class CommunityLevel(IntEnum):
  LOW = 0
  MEDIUM = 1
  HIGH = 2


@dataclass(order=True, frozen=True)
class CLRecord:
  state: str
  county: str
  county_fips: str = field(compare=False)
  covid_cases_per_100k: int = field(compare=False)
  date_updated: datetime
  county_population: Optional[int] = field(compare=False, default=None)
  health_service_area_number: Optional[int] = field(compare=False, default=None)
  health_service_area: Optional[str] = field(compare=False, default=None)
  health_service_area_population: Optional[int] = field(compare=False, default=None)
  covid_inpatient_bed_utilization: Optional[str] = field(compare=False, default=None)
  covid_hospital_admissions_per_100k: Optional[int] = field(compare=False, default=None)
  covid_19_community_level: Optional[CommunityLevel] = field(compare=False, default=None)

  def __post_init__(self):
    super().__setattr__('date_updated', datetime.fromisoformat(self.date_updated))
    if self.covid_hospital_admissions_per_100k is not None:
      super().__setattr__('covid_hospital_admissions_per_100k', atof(self.covid_hospital_admissions_per_100k))
    super().__setattr__('covid_cases_per_100k', atof(self.covid_cases_per_100k))
    if self.health_service_area_number is not None:
      super().__setattr__('health_service_area_number', int(atof(self.health_service_area_number)))
    if self.health_service_area_population is not None:
      super().__setattr__('health_service_area_population', int(atof(self.health_service_area_population)))
    if self.county_population is not None:
      super().__setattr__('county_population', int(atof(self.county_population)))
    if self.covid_19_community_level is not None:
      super().__setattr__('covid_19_community_level', CommunityLevel[self.covid_19_community_level.upper()])


def kwmap(function, iterable):
  for args in iterable:
    yield function(**args)


urlbase = 'https://data.cdc.gov/resource/3nnm-4jni.json'
useragent = 'PyCCL/2'
apptoken = 'mM3osziuGbfO3gs079bNPJ0Q2'
uaandtoken = {'User-Agent': useragent, 'X-App-Token': apptoken}


def request_cl_rlib(session: 'Session', state: str, county: str) -> list[CLRecord]:
  with session.get(urlbase, params={'state': state, '$where': "starts_with(county, '{}')".format(county)}, headers=uaandtoken) as res:
    jobjs = res.json() # type: list[dict]
  return list(kwmap(CLRecord, jobjs))


def request_cl(state: str, county: str) -> list[CLRecord]:
  r = Request(url='{}?state={}&$where=starts_with(county,%20%27{}%27)'.format(urlbase, state, county), headers=uaandtoken)
  with urlopen(r) as res:
    jobjs = json.loads(res.read()) # type: list[dict]
  return list(kwmap(CLRecord, jobjs))


delay = timedelta(days=14)


def earliest_nomask_date(cls: list[CLRecord]) -> Optional[datetime]:
  try:
    last_bad = sorted(filter(lambda c: c.covid_19_community_level > CommunityLevel.MEDIUM, cls))[-1]
  except IndexError as e:
    # See if it's never been high, or just no data
    if cls:
      return None
    else:
      raise e
  earliest_possible = last_bad.date_updated + delay
  if earliest_possible < datetime.now():
    return None
  else:
    return earliest_possible


def parse_state_county(arg: str) -> Tuple[str, str]:
  return tuple(arg.split(','))


def main():
  setlocale(LC_NUMERIC, '')

  parser = ArgumentParser(description='When can the vaccinated quit wearing masks?')
  parser.add_argument('location', type=parse_state_county, nargs='+', help='a "State,County" to look up')
  args = parser.parse_args()

  if use_rlib:
    with Session() as s:
      res = zip(args.location, starmap(partial(request_cl_rlib, s), args.location))
  else:
    res = zip(args.location, starmap(request_cl, args.location))

  for (state, county), rec in res:
    try:
      print('{} County, {}, {}'.format(county, state, earliest_nomask_date(rec)))
    except IndexError as e:
      print('{} County, {}, {}'.format(county, state, e))

if __name__ == '__main__':
  main()

cctl.py

#!/usr/bin/env python3
# encoding=utf-8
# vim: set nobomb :

from enum import IntEnum, unique
from dataclasses import dataclass, field
from typing import Optional, Tuple
from datetime import datetime, timedelta
import json
from argparse import ArgumentParser
from locale import atof, setlocale, LC_NUMERIC
from itertools import starmap
from functools import partial


try:
  from requests import Request, Session
  use_rlib = True
except ImportError:
  from urllib.request import urlopen, Request
  use_rlib = False


@unique
class CommunityTransmissionLevel(IntEnum):
  LOW = 0
  MODERATE = 1
  SUBSTANTIAL = 2
  HIGH = 3


@dataclass(order=True, frozen=True)
class CTLRecord:
  state_name: str
  county_name: str
  fips_code: str = field(compare=False)
  report_date: datetime
  cases_per_100k_7_day_count: float = field(compare=False)
  community_transmission_level: CommunityTransmissionLevel = field(compare=False)
  percent_test_results_reported: Optional[float] = field(compare=False, default=None)

  def __post_init__(self):
    super().__setattr__('report_date', datetime.fromisoformat(self.report_date))
    super().__setattr__('cases_per_100k_7_day_count', atof(self.cases_per_100k_7_day_count))
    if self.percent_test_results_reported is not None:
      super().__setattr__('percent_test_results_reported', atof(self.percent_test_results_reported))
    super().__setattr__('community_transmission_level', CommunityTransmissionLevel[self.community_transmission_level.upper()])


def kwmap(function, iterable):
  for args in iterable:
    yield function(**args)


urlbase = 'https://data.cdc.gov/resource/8396-v7yb.json'
useragent = 'PyCCTL/1'
apptoken = 'mM3osziuGbfO3gs079bNPJ0Q2'
uaandtoken = {'User-Agent': useragent, 'X-App-Token': apptoken}


def request_ctl_rlib(session: 'Session', state: str, county: str) -> list[CTLRecord]:
  with session.get(urlbase, params={'state_name': state, 'county_name': '{} County'.format(county)}, headers=uaandtoken) as res:
    jobjs = res.json() # type: list[dict]
  return list(kwmap(CTLRecord, jobjs))


def request_ctl(state: str, county: str) -> list[CTLRecord]:
  r = Request(url='{}?state_name={}&county_name={}%20County'.format(urlbase, state, county), headers=uaandtoken)
  with urlopen(r) as res:
    jobjs = json.loads(res.read()) # type: list[dict]
  return list(kwmap(CTLRecord, jobjs))


delay = timedelta(days=14)


def earliest_nomask_date(ctls: list[CTLRecord]) -> Optional[datetime]:
  last_bad = sorted(filter(lambda c: c.community_transmission_level > CommunityTransmissionLevel.MODERATE, ctls))[-1]
  earliest_possible = last_bad.report_date + delay
  if earliest_possible < datetime.now():
    return None
  else:
    return earliest_possible


def parse_state_county(arg: str) -> Tuple[str, str]:
  return tuple(arg.split(','))


def main():
  setlocale(LC_NUMERIC, '')

  parser = ArgumentParser(description='When can the vaccinated quit wearing masks?')
  parser.add_argument('location', type=parse_state_county, nargs='+', help='a "State,County" to look up')
  args = parser.parse_args()

  if use_rlib:
    with Session() as s:
      res = zip(args.location, starmap(partial(request_ctl_rlib, s), args.location))
  else:
    res = zip(args.location, starmap(request_ctl, args.location))

  for (state, county), rec in res:
    print('{} County, {}, {}'.format(county, state, earliest_nomask_date(rec)))

if __name__ == '__main__':
  main()

This is for the old "Community Transmission Level" measurement. No word yet on datasets for COVID-19 Community Levels.

Now it's here!

ccl.py has now been updated to handle the March 31st, 2022 field name/format changes.