add Gitea CI

.gitea/workflows/lint.yml

@@ -0,0 +1,23 @@
+name: lint
+run-name: lint is launched by ${{ github.actor }}
+on: [push]
+jobs:
+  check:
+    runs-on: ubuntu-latest
+    name: lint
+    steps:
+      - run: echo "🎉 The job was automatically triggered by a ${{ github.event_name }} event."
+      - run: echo "🐧 This job is now running on a ${{ runner.os }} server hosted by Gitea!"
+      - run: echo "🔎 The name of your branch is ${{ github.ref }} and your repository is ${{ github.repository }}."
+      - name: Check out repository code
+        uses: actions/checkout@v4
+      - name: Create Python virtualenv
+        run: |
+          mkdir -p .cache
+          python -m venv .
+          source bin/activate
+          XDG_CACHE_HOME=.cache pip install pylint-venv pylint
+      - name: Lint
+        run: XDG_CACHE_HOME=.cache ./bin/pylint -d E0401 *.py 
+      - run: echo "🍏 This job's status is ${{ job.status }}."
+

pacer2influxdb.py

@@ -1,60 +1,59 @@
 #!/usr/bin/env python3
-from influxdb import InfluxDBClient
+
+"""Export Pacer data to InfluxDB"""
+
 import datetime
 import time
 import sqlite3 as lite
 import argparse
+from influxdb import InfluxDBClient
 
-def main(host='localhost', port=8086, user='root', password='root', dbname='demo', dbfile='demo.db'):
+def main(args):
     """Instantiate the connection to the InfluxDB client."""
-
-    db = InfluxDBClient(host, port, user, password, dbname)
-    measurements = db.get_list_measurements()
-    ts = 0
+    influx_db = InfluxDBClient(args.host, args.port, args.user, args.password, args.dbname)
+    measurementime_stamp = influx_db.get_list_measurementime_stamp()
+    time_stamp = 0
     lastdate = 0
-    laststeps = 0
-    if measurements != []:
-        lastentry = db.query('SELECT LAST("steps") FROM "steps"')
+    lastime_stampteps = 0
+    if measurementime_stamp != []:
+        lastentry = influx_db.query('SELECT LAST("steps") FROM "steps"')
+        pointime_stamp = lastentry.get_pointime_stamp('steps')
+        lastdate = list(pointime_stamp)[0]['time']
+        time_stamp = time.mktime(datetime.datetime.strptime(lastdate, '%Y-%m-%dT%H:%M:%SZ')
+                         .timetuple())
 
-        points = lastentry.get_points('steps')
-        lastdate = list(points)[0]['time']
-        ts = time.mktime(datetime.datetime.strptime(lastdate, '%Y-%m-%dT%H:%M:%SZ').timetuple())
-        if ts == datetime.datetime.now().timestamp() // 86400 * 86400 + time.timezone:
-                points = lastentry.get_points('steps')
-                laststeps = list(points)[0]['last']
+        if time_stamp == datetime.datetime.now().timestamp() // 86400 * 86400 + time.timezone:
+            pointime_stamp = lastentry.get_pointime_stamp('steps')
+            lastime_stampteps = list(pointime_stamp)[0]['last']
 
     if args.verbose:
-        print("last entry is %s, ts is %s, number of steps is %s\n" % (lastdate,ts,laststeps))
+        print(f'last entry is {lastdate}, time_stamp is {time_stamp}, \
+        number of steps is {lastime_stampteps}\n')
 
-    con = lite.connect(dbfile)
+    con = lite.connect(args.dbfile)
     with con:
-
         cur = con.cursor()
-        cur.execute("SELECT recordedForDate,steps,distanceInMeters,activeTimeInSeconds,calories FROM dailyActivityLog WHERE recordedForDate >= %s AND steps > %s" % (ts,laststeps))
-
+        cur.execute(f'SELECT recordedForDate,steps,distanceInMeters,activeTimeInSeconds,calories \
+                     FROM dailyActivityLog \
+                     WHERE recordedForDate >= {time_stamp} AND steps > {lastime_stampteps}')
         while True:
-
             row = cur.fetchone()
-            if row == None:
+            if row is None:
                 break
             mytime = datetime.datetime.fromtimestamp(row[0]).strftime('%Y-%m-%dT%H:%M:%SZ')
-            data = [
-                   {"measurement":"steps",
+            data = [{"measurement":"steps",
                     "time":mytime,
                     "fields": {
                               "steps":row[1],
                               "distanceInMeters":row[2],
                               "activeTimeInSeconds":row[3],
                               "calories":row[4]
-                    }
-                }
-                ]
-
+                              }
+                    }]
             if args.verbose:
-                print("writing data for %s" % (mytime)) 
-
-            db.write_points(data)
-        db.close()
+                print(f'writing data for {mytime}')
+            influx_db.write_pointime_stamp(data)
+        influx_db.close()
 
 def parse_args():
     """Parse the args from main."""
@@ -80,7 +79,5 @@ def parse_args():
                         help='sqlite (pacer) database name')
     return parser.parse_args()
 
-
 if __name__ == '__main__':
-    args = parse_args()
-main(host=args.host, port=args.port, user=args.user, password=args.password, dbname=args.dbname, dbfile=args.dbfile)
+    main(parse_args())