1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
|
#!/usr/bin/env python3
import os
import include_homekit
from datetime import datetime
from typing import Tuple, List, Optional
from argparse import ArgumentParser
from homekit.config import config, AppConfigUnit
from homekit.database import SimpleState
from homekit.api import WebApiClient
class OpenwrtLoggerConfig(AppConfigUnit):
@classmethod
def schema(cls) -> Optional[dict]:
return dict(
database_name_template=dict(type='string', required=True)
)
def parse_line(line: str) -> Tuple[int, str]:
space_pos = line.index(' ')
date = line[:space_pos]
rest = line[space_pos+1:]
return (
int(datetime.strptime(date, "%Y-%m-%dT%H:%M:%S%z").timestamp()),
rest
)
if __name__ == '__main__':
parser = ArgumentParser()
parser.add_argument('--file', type=str, required=True,
help='openwrt log file')
parser.add_argument('--access-point', type=int, required=True,
help='access point number')
arg = config.load_app(OpenwrtLoggerConfig, parser=parser)
state = SimpleState(name=config.app_config['database_name_template'].replace('{ap}', str(arg.access_point)),
default=dict(seek=0, size=0))
fsize = os.path.getsize(arg.file)
if fsize < state['size']:
state['seek'] = 0
with open(arg.file, 'r') as f:
if state['seek']:
# jump to the latest read position
f.seek(state['seek'])
# read till the end of the file
content = f.read()
# save new position
state['seek'] = f.tell()
state['size'] = fsize
lines: List[Tuple[int, str]] = []
if content != '':
for line in content.strip().split('\n'):
if not line:
continue
try:
lines.append(parse_line(line))
except ValueError:
lines.append((0, line))
api = WebApiClient()
api.log_openwrt(lines, arg.access_point)
|
