db-manager/src/pis.py

import yaml, hashlib
import logger as log
import mongo

REBUILD :bool = False  # Set to True to force rebuild

log.out("pis.py: PIS Module Loaded", "DBUG")
file_location :str = "/app/data/pis/gwr.yaml" # Production & Testing
#file_location :str = "/home/fred.boniface/git/owlboard/db-manager/data/pis/gwr.yaml" # Local Development

def runUpdate():
  if (not requiresUpdate()):
    log.out('pis.runUpdate: PIS Codes do not need updating', 'INFO')
    return
  log.out(f"pis.runUpdate: Update required", "INFO")
  pis_data = load()
  pis_parsed = parse(pis_data)
  pis_indexes = ["stops", "tiplocs"]
  mongo.dropCollection("pis")
  mongo.putMany("pis", pis_parsed, pis_indexes)
  
def requiresUpdate():
  if REBUILD:
    return True
  currentHash = mongo.getMetaHash("pis")
  with open(file_location, "r") as f:
    text = f.read()
    newHash = hashlib.md5(text.encode()).hexdigest()
  log.out(f"pis.requiresUpdate: Existing PIS Hash: {currentHash}","INFO")
  log.out(f"pis.requiresUpdate: New PIS hash: {newHash}", "INFO")
  if currentHash is None or newHash != currentHash:
    log.out("pis.requiredUpdate: PIS Data requires updating", "INFO")
    mongo.putMetaHash("pis", newHash)
    return True
  log.out("pis.requiredUpdate: PIS Data is up to date", "INFO")
  return False

def load():
    with open(file_location, "r") as data:
        try:
            pis = yaml.safe_load(data)
            return pis["pis"]
        except yaml.YAMLError as exc:
            log.out(f"pis.load: Error loading YAML: {exc}", "EROR")
            return exc

def parse(codeList):
    StartLen = len(codeList)
    log.out(f"pis.parse: codeList starting length: {StartLen}", "DBUG")
    log.out(f"pis.parse: Removing duplicate codes & adding TIPLOCs")
    for i in codeList:
        stops = i['stops']
        code = i['code']
        for ii in codeList:
            if stops == ii['stops'] and code != ii['code']:
                log.out(f"Identical stopping pattern found: {ii['code']}","DBUG")
                codeList.remove(ii) # Instead of removing, I should add a property (duplicate: true),
                                    # then I can filter this out on the backend when searching by start
                                    # and end stations and just use query one for other queries, this
                                    # means that when searching by code, a perfectly valid code won't
                                    # show 0 results.
        tiplocs = []
        for iii in stops:
          tiplocs.append(getTiploc(iii))
        i['tiplocs'] = tiplocs
    log.out(f"pis.parse: Removed {StartLen - len(codeList)} duplicates", "INFO")
    return codeList

def getTiploc(crs :str):
  CRS = crs.upper()
  query = {
    '3ALPHA': CRS
  }
  try:
    res = mongo.query("stations", query)
    if 'TIPLOC' in res:
      return res['TIPLOC']
  except Exception as e:
    log.out(f"pis.getTiploc: Error finding tiploc: {query}", "EROR")
    log.out(f"ERROR: {e}", "EROR")