Browse Source

Prototype code for trace validation

master
Markus Becker 2 years ago
parent
commit
d26c95b0f1
  1. 3
      setup.py
  2. 95
      validate/__init__.py

3
setup.py

@ -15,7 +15,8 @@ setup(
entry_points={
'console_scripts': [
# add entry points here
'location-ingest = ingest:cmd',
'loclib-ingest = ingest:cmd',
'loclib-validate = validate:cmd',
],
},
python_requires=">=3",

95
validate/__init__.py

@ -0,0 +1,95 @@
import argparse
import logging
import sqlite3
from tqdm import tqdm
from typing import List, Callable, NewType
logger = logging.getLogger(__name__)
class Sample:
def __init__(self, lat, lon, timestamp):
super().__init__()
self.lat = lat
self.lon = lon
self.timestamp = timestamp
class Trace:
def __init__(self, id: int, user: int):
super().__init__()
self.data : List[Sample] = []
self.id = id
self.user = user
def append(self, sample: Sample):
self.data.append(sample)
class TraceRating:
def __init__(self, score: float, trace: Trace):
super().__init__()
self.score = score
class TestResults:
def __init__(self, good: float = 0.8):
super().__init__()
self.traceRatings = []
self.numGood = 0
self.numBad = 0
self._good = good
def add(self, rating: TraceRating) -> None:
self.traceRatings.append(rating)
if rating.score >= self._good:
self.numGood += 1
else:
self.numBad += 1
def length(self) -> int:
return len(self.traceRatings)
def __repr__(self):
return f"+{self.numGood}|-{self.numBad}/{self.length()} @({self._good})"
Metric = Callable[[List[Sample], Sample, float], float]
metrics: List[Metric] = []
def cmd() -> None:
parser = argparse.ArgumentParser(description='Test location database for trustworthiness')
parser.add_argument('database', nargs="+", help='Target location database')
parser.add_argument('-v', '--verbose', help='Additional logging', action='store_true', dest='verbose')
args = parser.parse_args()
logging.basicConfig(level=logging.WARNING, format="%(asctime)-15s %(levelname)-7s [%(name)s] %(message)s")
if args.verbose:
logging.getLogger().setLevel(logging.DEBUG)
logger.info("Verbose logging is enabled")
logger.debug(args.database)
for db in args.database:
print(db, testDB(db))
def testDB(dbpath: str, weights: List[float] = []) -> TestResults:
results = TestResults()
conn = sqlite3.connect(dbpath)
traceCursor = conn.cursor()
sampleCursor = conn.cursor()
for traceRow in traceCursor.execute("SELECT * FROM traces ORDER BY id ASC"):
logger.debug(traceRow)
trace = Trace(traceRow[0], traceRow[1])
traceData = sampleCursor.execute(
"SELECT lat, lon, unixtime FROM samples WHERE trace = ? ORDER BY unixtime",
[traceRow[0]]
)
for sampleRow in traceData:
trace.append(Sample(*sampleRow))
score = testTrace(trace)
logger.debug(score)
sampleCursor.close()
traceCursor.close()
return results
def testTrace(trace: Trace) -> float:
pass
Loading…
Cancel
Save