sheppy/ths-datenlogger
1
0
Fork 0
mirror of https://github.com/FAUSheppy/ths-datenlogger synced 2025-12-06 04:11:34 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

Initial

Browse Source
This commit is contained in:
Sheppy
2018-01-25 03:20:21 +01:00
commit 585a2ea40c
18 changed files with 1115 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
.gitignore vendored Normal file
View File

@@ -0,0 +1,9 @@
*.png
*.dbf
__py*
*.swp
*.dbf
*.DBF
*.xls
build/
dist/

83
config_parse.py Normal file
View File

@@ -0,0 +1,83 @@
import configparser
import sys
conf = None
default_conf = None
def parse_config():
global conf
global default_conf
conf = configparser.ConfigParser()
conf.read("ths_config.txt")
default_conf = configparser.ConfigParser()
default_conf.read("ths_readonly_default.conf")
if conf == None or (len(conf.sections()) == 0 and len(default_conf.sections()) == 0):
print("Error: Missing configuration file, cannot continue")
raise Exception("Missing configuration file")
def get_keys(like=None):
ret = conf["plot"].keys()
if like != None:
ret = list(filter(lambda x:like in x,ret))
if len(ret) == 0:
print("No options that contain the string '%s'"%like)
return ""
return ret
def change_cfg(key,value):
global conf
confs = conf["plot"]
v = str(value)
key = str(key)
if key not in confs:
return False
else:
confs[key] = value
return True
def CFG(tag):
global conf
global default_conf
if conf == None:
parse_config()
if len(default_conf.sections()) > 0:
default_confs = default_conf["plot"]
else:
default_confs = None
confs = conf["plot"]
if tag in confs:
return parse_cfg(confs[tag])
elif default_confs != None and tag in default_confs:
print("Warning: %s no found in configuration, defaulting to %s" % (str(tag),str(default_conf[tag])),sys.stderr)
return parse_cfg(default_confs[tag])
else:
raise Exception("Error: configuration option %s not found in configuration and no default value for it, cannot continue, exit." % str(tag))
def parse_cfg(c):
if c == None:
raise Exception("Config key (%s) found but has no value. Cannot continue, exit." % str(c))
c = c.strip("'")
c = c.strip('"')
if c in ["yes","ja","True","Yes","Ja","true"]:
return True
if c in ["no","nein","False","No","Nein","false"]:
return False
try:
return int(c)
except ValueError:
pass
try:
return float(c)
except ValueError:
pass
return c
CFG("show_avg")

8
constants.py Normal file
View File

@@ -0,0 +1,8 @@
#!/usr/bin/python3
from config_parse import CFG
GLOBAL_FONT = {'family':CFG("font"),'weight':'normal','size': CFG("global_text_size")}
BASE_PATH=CFG("default_source_path")
SOURCE_PATH=CFG("default_target_dir")
FIGURE=0
AXIS=1
CALLBACK=2

61
frontend_new.py Normal file
View File

@@ -0,0 +1,61 @@
#!/usr/bin/python3
import sys
import tkinter
from tkinter import filedialog
import plot_main
import input_backend
import frontend_utils as futils
from config_parse import CFG
from language import LAN
l = LAN[CFG("language")]
tk = tkinter.Tk()
tk.withdraw()
def main_repl(datapoints,path,date1=None,date2=None,done1=False,done2=False):
### READ IN DATES ###
futils.info_list(datapoints)
while not done1:
date1,done1 = futils.input_date_repl(datapoints,startdate=True)
while not done2:
date2,done2 = futils.input_date_repl(datapoints,startdate=False)
### CHECK DATES ###
done1,done2 = futils.check_dates(path,date1,date2)
if not done1 or not done2:
main_repl(datapoints,path,date1,date2,done1,done2)
else:
plot_main.plot(datapoints,path,date1,date2)
def selection_repl(path):
if path != None:
datapoints = input_backend.read_in_file(path)
if CFG("debug_no_interactive"):
plot_main.plot(datapoints,path)
return None
main_repl(datapoints,path)
else:
tmp=input( "\n -> Type 'n' or 'new' <ENTER> to restart with another file\n -> Type 'r' or 'restart'<ENTER> to use the current file again\n (restart or selecting the same file WILL OVERRIDE the picture you just generated!)\n -> 'c'<ENTER> oder 'c <CONFIG_OPTION_NAME> <NEW_VALUE>' um Konfigurationsoptionen zu ändern\n -> Or press just <ENTER> to exit: ")
if tmp == None or tmp == "":
return None
elif tmp in ["r","restart"]:
return path
elif tmp in ["n","new"]:
return futils.open_file()
elif tmp.startswith('c'):
raise NotImplementedError("On the fly configuration not yet implemented.")
else:
return path
def main():
### PREVENT MULTICORE SUPPORT ###
if CFG("enable_multicore_support"):
raise NotImplementedError("multiprocessing not fully implemented")
### PROMT TO OPEN FILE ###
FILE_READY = False
while True:
path = selection_repl(futils.open_file())
if path == None:
break

217
frontend_utils.py Normal file
View File

@@ -0,0 +1,217 @@
#!/usr/bin/python3
import sys
import tkinter
import plot_main
import config_parse
from config_parse import CFG
from language import LAN
from datetime import datetime
l = LAN[CFG("language")]
timeformat = "%d.%m.%y %H:%M:%S (%A)"
def parse_date_from_user_input(s,end_of_day=False,datapoints=None):
today = datetime.now()
day = 0
month = 0
year = 0
hour = 0
minute = 0
second = 0
## EMPTY ##
if s == None or s == "":
return None
## TIME ##
if len(s.split(" ")) > 1:
time = s.split(" ")[1]
time_a = time.split(":")
if len(time_a) > 0:
hour = int(time_a[0])
elif end_of_day > 0:
hour = 23
if len(time_a) > 1:
minute = int(time_a[1])
elif end_of_day:
minute = 59
if len(time_a) > 2:
second = int(time_a[2])
elif end_of_day:
second = 59
elif end_of_day:
hour = 23
minute = 59
second = 59
## DATE ##
tmp = s.split(" ")[0]
## allow more speperators ##
sep = None
for c in ["-",".",","]:
if c in tmp:
sep = c
break
if sep == None:
sep = "-"
tmp = tmp.strip(sep)
if len(tmp.split(sep)) == 0:
raise ValueError("Invalid Date '%s'"%str(s))
else:
date_a = tmp.split(sep)
if len(date_a) > 0:
day = int(date_a[0])
if len(date_a) > 1:
month = int(date_a[1])
if len(date_a) > 2:
year = int(date_a[2])
if year == 0:
if today.month > month:
year = today.year
else:
year = today.year-1
if month == 0:
if today.day > day and today.year == year:
month = today.month
else:
month = today.month-1
if month < 1:
month = 12-month
ret = datetime(year,month,day,hour,minute,second)
try:
times = datapoints[CFG("plot_temperatur_key")].times
if ( ret > max(times) or ret < min(times) ) and min(times).day < ret.day < max(times).day and min(times).month == max(times).month:
month = min(times.month)
except Exception as e:
print("Warning, magic date selection failed for an unknown reason")
ret = datetime(year,month,day,hour,minute,second)
return ret
def info_list(datapoints):
if len(datapoints.keys()) > 0:
print("Erster Datensatz: "+min(datapoints[list(datapoints.keys())[0]].times).strftime(timeformat))
print("Letzer Datensatz: "+max(datapoints[list(datapoints.keys())[0]].times).strftime(timeformat))
print("Anzahl Datensätze: "+str(len(datapoints[list(datapoints.keys())[0]].times)))
else:
print("Keine Datesätze gefunden!")
print_sep_line(True)
def input_date_repl(datapoints,startdate=True):
date = None
while True:
try:
if startdate:
ret = input(l["input_first_date"])
else:
ret = input(l["input_second_date"])
except EOFError:
return (date,True)
except KeyboardInterrupt:
sys.exit(2)
if ret in ["h","help","hilfe"]:
if startdate:
print(l["input_first_date_help"])
else:
print(l["input_second_date_help"])
continue
elif ret == "list":
info_list(datapoints)
continue
else:
try:
if startdate:
date=parse_date_from_user_input(ret,datapoints=datapoints)
else:
date=parse_date_from_user_input(ret,True,datapoints)
return (date,True)
except ValueError as e:
print(l["cannot_parse_date"] + "( was: {} )\n".format(ret))
return (None,False)
def print_sep_line(ln=False):
if not ln:
print("-----------------------------------------------")
else:
print("-----------------------------------------------",end='')
def check_dates(path,date1,date2,options=""):
print_sep_line()
if options!="":
print("Config options: %s"%options)
print("Datei: %s"%path)
if date1 == None and date2 == None:
print("Info: Keine Zeitbeschränkung gewählt. Alle vorhandenen Daten werden verwendet.")
return (True,True)
elif date1 == None:
print("Alle Werte vor %s"%date2.strftime(timeformat))
elif date2 == None:
print("Alle Werte nach %s"%date1.strftime(timeformat))
else:
print("Start: %s\nEnde: %s"%(date1.strftime(timeformat),date2.strftime(timeformat)))
FIRST=True
while(True):
try:
if FIRST:
ret = input("Stimmt das so?\n -> <ENTER> für ja/weiter\n -> 's'<ENTER> für Startzeit ändern\n -> 'e'<ENTER> für Endzeit ändern\n -> 'b'<ENTER> für beides ändern\n -> 'exit'<ENTER> to exit\n---> ")
else:
ret = input("Versuchen sie es nochmal: ")
except EOFError:
ret = ""
if ret == 's':
return (False,True)
elif ret == 'e':
return (True,False)
elif ret == 'b':
return (False,False)
elif ret.startswith('c '):
tmp = config_options(ret)
if tmp == "":
pass
else:
options += "\n "+tmp
check_dates(path,date1,date2,options)
elif ret == "":
return (True,True)
elif ret == "exit":
sys.exit(0)
FIRST=False
def open_file():
front_end_source_path = CFG("default_source_path")
if CFG("use_input_filename"):
f = front_end_source_path + CFG("input_filename")
return f
path=None
path=tkinter.filedialog.askopenfilename(filetypes=(("DBF/XLS Files",("*.DBF","*.dbf","*.xls","*.XLS")),("All Files","*.*")))
if path == None or path=="":
print("Error: No file selected!")
return None
try:
open(path,'r').close()
except IOError:
print("Error: Unable to open selected file, perhaps it does no longer exist or you have insufficient permissions to open it?")
return None
return path
def config_options(string):
opt = ""
arg = string.split(" ")
if len(arg) == 2:
for l in config_parse.get_keys(arg[1]):
print(l)
elif len(arg) == 3:
if not config_parse.change_cfg(arg[1],arg[2]):
print("Option %s does not exist."%str(arg[1]))
else:
opt += "set %s %s"%(arg[1],arg[2])
print("set %s %s"%(arg[1],arg[2]))
else:
print("Ussage: c <configname> <value> / c <part_of_config_name> / c (= list all ), e to exit")
return opt

13
init.py Executable file
View File

@@ -0,0 +1,13 @@
#!/usr/bin/python3
import frontend_new
import sys
if __name__ == "__main__":
try:
frontend_new.main()
sys.exit(0)
except KeyboardInterrupt as e:
sys.exit(1)
except Exception as e:
print(e)
input("Ein Fehler ist aufgetreten, <ENTER> um zu beenden, wenn dieser Fehler unerwartet war -> Mail!")
sys.exit(1)

29
init.spec Normal file
View File

@@ -0,0 +1,29 @@
# -*- mode: python -*-
block_cipher = None
a = Analysis(['init.py'],
pathex=['Z:\\home\\ik15ydit\\reps\\random-code\\ths'],
binaries=[],
datas=[],
hiddenimports=[],
hookspath=[],
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher)
pyz = PYZ(a.pure, a.zipped_data,
cipher=block_cipher)
exe = EXE(pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
name='init',
debug=False,
strip=False,
upx=True,
runtime_tmpdir=None,
console=True )

133
input_backend.py Normal file
View File

@@ -0,0 +1,133 @@
#!/usr/bin/python3
from config_parse import CFG
from datetime import datetime, timedelta
from dbfread import DBF
import plot_timeutils
line_colors = ['b', 'r', 'g', 'c', 'm', 'y']
tname = CFG("temperatur_plot_name")
hname = CFG("humidity_plot_name")
dname = CFG("dewcels_plot_name")
color_id = 0
class Data:
def __init__(self,name,plot=False):
global color_id,line_colors
self.name = name
self.color=line_colors[color_id%len(line_colors)]
color_id += 1
self.data = []
self.times = []
self.plot = plot
## no idea on what kind of drugs I was when i wrote this function (it is somewhat ingenious though) ##
def get_timeframe(self, callback,date1=None,date2=None):
r=dict()
for t,c in zip(self.times,self.data):
t = callback(t,date1,date2)
if t == None:
continue
if t in r:
r[t]+=[c]
else:
r.update({t:[c]})
arr_t = []
arr_v = []
for k,v in r.items():
arr_t += [k]
arr_v += [sum(v)/len(v)]
arr_t = [x for x,_ in sorted(zip(arr_t,arr_v))]
arr_v = [x for _,x in sorted(zip(arr_t,arr_v))]
return (arr_t,arr_v)
def parse_line(datapoints,line,timekey,keys,time_parser,timeformat=None):
# This function expects:
# - datapoints { String:DataObject }
# - line { String:Any }
# - timekey String (key for timevalue in 'line')
# - keys [ (String,String) ] (source_key in 'line' to target_key in 'datapoints')
time = time_parser(line[ timekey ],timeformat)
for key in keys:
datapoints[ key[1] ].data += [ line[ key[0] ] ]
datapoints[ key[1] ].times += [ time ]
def read_in_file(path,backend=None):
global tname
global hname
global dname
global opath
datapoints = dict()
pt=CFG("plot_temperatur_key")
ph=CFG("plot_humidity_key")
pd=CFG("plot_dewcels_key")
## NAME PADDING ##
max_name_len = max(len(tname),len(hname),len(dname))
while len(tname) < max_name_len:
tname += " "
while len(hname) < max_name_len:
hname += " "
while len(dname) < max_name_len:
dname += " "
datapoints.update({ pt:Data( tname,CFG("plot_temperatur") ) })
datapoints[pt].color = CFG("temperatur_color")
datapoints.update({ ph:Data( hname,CFG("plot_humidity") ) })
datapoints[ph].color = CFG("humidity_color")
datapoints.update({ pd:Data( dname,CFG("plot_dewcels") ) })
datapoints[pd].color = CFG("dewcels_color")
if path == None:
raise Exception("Path in plot.read_in was None")
elif backend != None:
backend(path)
elif path.endswith(".DBF") or path.endswith(".dbf"):
dbfread(path,datapoints,pt,ph,pd)
elif path.endswith(".xls") or path.endswith(".XLS"):
csvread(path,datapoints,pt,ph,pd)
else:
raise NotImplementedError("Cannot determine filetype, cannot continue. Exit.")
check_read_in(datapoints)
return datapoints
def dbfread(path,datapoints,pt,ph,pd):
for record in DBF(path):
parse_line(datapoints,record,'DATETIME',[ ('TEMPCELS',pt) , ('HUMIDITY',ph) , ('DEWCELS',pd) ] ,plot_timeutils.time_from_dbf)
def csvread(path,datapoints,pt,ph,pd):
count = 0;
with open(path) as f:
for l in f:
if l.startswith(">>") or l.startswith("--") or l.startswith("NO."):
count += 1
continue
else:
row_arg = list(map(lambda s:s.replace(" ","").replace(",","."),l.split("\t")))
row = {"temp":None,"hum":None,"taupunkt":None,"datetime":None}
row["datetime"] = row_arg[1]+row_arg[2]
row["temp"] = float(row_arg[3])
row["hum"] = float(row_arg[4])
row["taupunkt"] = float(row_arg[5])
parse_line(datapoints,row,'datetime',[ ('temp',pt) , ('hum',ph) , ('taupunkt',pd) ],\
plot_timeutils.time_from_csv,timeformat="%d-%m-%Y%H:%M:%S")
print("Info: Ignored %d lines at beginning of file"%count)
def check_read_in(datapoints):
good = False
for v in datapoints.values():
if len(v.times) != len(v.data):
print("more timestamps than data (or visa versa), this indicates that the file is corrupted, cannot continue")
good = False
break
if len(v.times) > 1:
good = True
if not good:
input("reading input file failed for an unknown reason, <ENTER> to exit")
import sys
sys.exit(1)

23
language.py Normal file
View File

@@ -0,0 +1,23 @@
end="\nDrücken sie 'STRG + C' ('STEUERUNG CANCEL') <ENTER> um das Program zu beenden"
input_date = "Geben sie den Zeitpunkt an, an dem der Plot {} soll! \n\
\nDatum/Uhrzeit im Format 'DD-MM-YYYY HH:MM:SS'. Wird die Uhrzeit weggelassen, \n\
so wird 00:00:00 (Startzeit) bzw. 23:59:59 (Endzeit) angenommen.\n\
Werden Jahr oder Monat weggelassen wird (versucht) ein passendes Datum zu wählen.\n\
Lassen sie die Zeile leer um mit dem ersten existierenden Wert anzufangen \n\n\
list<ENTER> um eine Übersicht über die gefunden Datenwerte zu erhalten\n\n\
Beispiele für Formate (angenommen es ist der 12.01.2017): \n\n\
11 12 -> 11.01.2017 12:00:00 \n\
11-01 -> 11.01.2017 00:00:00 \n\
13 -> 13.12.2016 00:00:00 \n\
13-01-2017 -> 13.01.2017 00:00:00 \n\
13-1-2017 17:1:4 -> 13.01.2017 17:01:04 (nuller können also weggelassen werden)\n"
hilfe=" oder h/help/hilfe <ENTER> für Hilfe\n"
LAN = {"DE":{},"EN":{}}
LAN["DE"]["input_first_date_help"] = input_date.format("beginnen") + end + "\n"
LAN["DE"]["input_second_date_help"] = input_date.format("enden") + end + "\n"
LAN["DE"]["input_first_date"] = "\nStartzeit"+hilfe+"(Format: DD-MM-YY HH:MM:SS): "
LAN["DE"]["input_second_date"] = "\nEndzeit "+hilfe+"(Format: DD-MM-YY HH:MM:SS): "
LAN["DE"]["cannot_parse_date"] = "Konnte Datum/Uhrzeit nicht verarbeiten! \n"
LAN["DE"]["dstart_bigger_dend"] = "Startzeit > Endzeit. MÖÖÖÖP \n"

231
plot_graphutils.py Normal file
View File

@@ -0,0 +1,231 @@
#!/usr/bin/python3
from config_parse import CFG
from datetime import datetime, timedelta
import matplotlib
matplotlib.use(CFG("use_gui_backend"))
import matplotlib.pyplot as plt
import matplotlib.dates
import matplotlib.ticker as ticker
from constants import *
import math
import plot_timeutils
matplotlib.rc('font', **GLOBAL_FONT)
def getlimits_y(y):
ymax = max(y)+CFG("empty_space_above_plot")
y_min_height = CFG("yaxis_minnimum_hight")
if y_min_height != 0 and y_min_height > ymax:
ymax = y_min_height
y_start_val = CFG("yaxis_start_value")
if y_start_val < min(y) or CFG("yaxis_force_start_value"):
ymin=y_start_val
else:
ymin=min(y)
return (ymin,ymax)
def avg(array):
return sum(array)/float(len(array))
def legend_box_contents(name,y):
if CFG("show_min"):
name += " min: %.1f,"%min(y)
if CFG("show_max"):
name += " max: %.1f,"%max(y)
if CFG("show_avg"):
name += " Mittelwert: %.1f,"% avg(y)
return name.rstrip(",")
def general_background_setup(tup,ymin,ymax,x):
unix_x = list(map(plot_timeutils.unix,x))
### SET AXIS LIMITS ###
tup[AXIS].set_ylim([ymin,ymax])
tup[AXIS].set_xlim([plot_timeutils.unix(min(x)),plot_timeutils.unix(max(x))])
if CFG("draw_thresholds"):
hcrit=CFG("humidity_critical")
hwarn=CFG("humidity_warning")
tlow=CFG("acceptable_temp_low")
thigh=CFG("acceptable_temp_high")
tup[AXIS].axhline(y=CFG("target_temperatur"),ls=CFG("hline_line_style"),lw=CFG("hline_line_width"),color=CFG("acceptable_temp_color"))
tup[AXIS].axhline(y=hcrit,ls=CFG("hline_line_style"),lw=CFG("hline_line_width"),color=CFG("humidity_crit_color"))
tup[AXIS].axhspan(hwarn,hcrit,color=CFG("humidity_warning_color"),alpha=CFG("humidity_warning_alpha"))
tup[AXIS].axhspan(hcrit,ymax,color=CFG("humidity_crit_color"),alpha=CFG("humidity_crit_alpha"))
tup[AXIS].axhspan(tlow,thigh,color=CFG("acceptable_temp_color"),alpha=CFG("acceptable_temp_alpha"))
#### GRID ####
major_xticks = gen_xticks_from_timeseries(x)
minor_xticks = get_minor_xticks_from_major(major_xticks)
if CFG("raster"):
grid(tup,major_xticks,ymin,ymax)
#### XTICKS ####
tup[AXIS].set_xticks(major_xticks)
tup[AXIS].xaxis.set_major_formatter(ticker.FuncFormatter(xlabel_formater_callback))
tup[AXIS].xaxis.set_major_locator(ticker.FixedLocator(major_xticks, nbins=None))
tup[AXIS].xaxis.set_minor_locator(ticker.FixedLocator(minor_xticks, nbins=None))
tup[AXIS].xaxis.set_tick_params(which='minor',width=0.2,direction="out")
tup[AXIS].yaxis.set_major_locator(ticker.MultipleLocator(CFG("y_tick_interval")))
tup[AXIS].yaxis.set_minor_locator(ticker.MultipleLocator(1))
tup[AXIS].yaxis.set_tick_params(which='minor',width=0.2,direction="out")
tup[AXIS].tick_params(axis='x',which="major",labelsize=CFG("xticks_font_size"));
tup[AXIS].tick_params(axis='y',which="major",labelsize=CFG("yticks_font_size"));
## ROTATION XLABELS ##
rotation=CFG("xticks_label_degree")
if rotation > 0:
plt.xticks(rotation=rotation,ha='right')
## AXIS LABELS
ylabel_box = dict(boxstyle="square",facecolor='grey', alpha=0.4, edgecolor='black',lw=0.5)
xlabel_box = ylabel_box
label_size = 6
spacing=0.1
tup[AXIS].set_ylabel(CFG("y_label"),rotation='horizontal',size=label_size,bbox=ylabel_box)
tup[AXIS].yaxis.set_label_coords(0.055,0.95)
tup[AXIS].set_xlabel(CFG("x_label"),size=label_size,bbox=xlabel_box)
tup[AXIS].xaxis.set_label_coords(0.925,0.05)
## GENERAL LEGEND ##
legend_handle = tup[AXIS].legend(
loc=CFG("legend_location"),
edgecolor="inherit",
fancybox=False,
borderaxespad=spacing,
prop={'family': 'monospace','size':CFG("legend_font_size")}
)
legend_handle.get_frame().set_linewidth(0.2)
tup[AXIS].set_aspect(get_aspect_ratio(unix_x,ymin,ymax,major_xticks))
def get_aspect_ratio(ux,ymin,ymax,xticks):
ratio = 100
tmp = CFG("aspect_ratio")
if str(tmp) == "A4":
ratio = a4_aspect()
else:
ratio=tmp
magic_value = 3.25
return ratio * ( max(ux) - min(ux) ) / float(ymax - ymin + magic_value)
def a4_aspect():
return 1/math.sqrt(2)
def grid(tup,xticks,ymin,ymax):
lw = CFG("grid_line_width")
ls = CFG("grid_line_style")
color = CFG("grid_line_color")
hour_mul = 24
expected_vlines = len(list(filter(lambda xt: xt%3600 < 60,xticks)))
safety_first = 60*60 +10
step = xticks[1]-xticks[0]
if step < (24*3600)-safety_first:
if expected_vlines <= 6:
hour_mul = 1
elif expected_vlines <=12:
hour_mul = 2
elif expected_vlines <=24:
hour_mul = 4
for xt in xticks:
leck_mich = datetime.fromtimestamp(xt)
if leck_mich.hour == leck_mich.minute == leck_mich.second == 0:
tup[AXIS].axvline(xt,ls="-",lw=CFG("major_line_width"),color=color)
else:
tup[AXIS].axvline(xt,ls=ls,lw=lw,color=color)
## HLINES ##
y_interval = CFG("raster_hline_prefered_interval")
cur = ymin
while cur < ymax:
cur += y_interval
tup[AXIS].axhline(cur,ls=ls,lw=lw,color=color)
def find_step(step,x,total_xticks):
intervals = parse_possible_intervals()
start = min(x)
if CFG("always_allow_days_as_xticks") and step > timedelta(days=1)/2:
step = timedelta(days=round(step.days+1))
start = min(x).replace(hour=0,second=0,minute=0)
return (start,step)
min_delta_step = timedelta(days=1) # the actual step that has the lowest delta
min_delta = timedelta(days=1000) # the delta o thus step
for s in intervals:
delta = max(s,step)-min(s,step)
if delta < min_delta:
min_delta_step = s
min_delta = delta
step = min_delta_step
start = plot_timeutils.round_time_to_step(start,step)
warn_on_too_much_xticks(x,total_xticks,step)
return (start,step)
def parse_possible_intervals():
intervals = CFG("acceptable_x_intervals")
parsed_intervals = []
for s in intervals.split(','):
try:
st = int(s[:-1])
except ValueError:
raise ValueError("'acceptable_x_intervals' muss die Form 'Zahl[s(econds),m(minutes),h(ours),d(days)]' haben!")
except Exception:
raise ValueError("invalid intervals for x_labels %s [index out of bounds], did you write something like this ',,,,' ?]"%str(intervals))
if s.endswith("s"):
if 60 % st != 0:
raise ValueError("interval must fit to next bigger interval so basicly for hours 24%interval==0")
parsed_intervals += [timedelta(seconds=st)]
elif s.endswith("m"):
if 60 % st != 0:
raise ValueError("interval must fit to next bigger interval so basicly for hours 24%interval==0")
parsed_intervals += [timedelta(minutes=st)]
elif s.endswith("h"):
if 24 % st != 0:
raise ValueError("interval must fit to next bigger interval so basicly for hours 24%interval==0")
parsed_intervals += [timedelta(hours=st)]
elif s.endswith("d"):
parsed_intervals += [timedelta(days=st)]
else:
raise ValueError("invalide Zeitspezifizierer in %s (muss, s,m,h oder d sein)"%str(intervals))
return parsed_intervals
def warn_on_too_much_xticks(x,total_xticks,step):
if (max(x)-min(x))/step > 2*total_xticks:
print("Warnung: maximales xinterval zu niedrig eine sinnvolle Anzahl an xticks zu generieren (total x_ticks: %d"%total_xticks)
def get_minor_xticks_from_major(major):
mult = CFG("minor_xticks_per_major")
step = (major[1]-major[0])/mult
ret = []
for x in major:
if x == max(major):
break
ret += [x+ 0*step]
ret += [x+ 1*step]
ret += [x+ 2*step]
ret += [x+ 3*step]
ret += [x+ 4*step]
return ret
def gen_xticks_from_timeseries(x):
ticks=CFG("prefered_total_xticks")
xmin = min(x)
xmax = max(x)
delta = xmax-xmin
step = delta/ticks
cur,step = find_step(step,x,ticks)
xticks = []
xmax += step*CFG("add_x_labels_at_end")
while cur < xmax:
xticks += [plot_timeutils.unix(cur)]
cur+=step
return xticks
def xlabel_formater_callback(tick_val, tick_pos):
dt = datetime.fromtimestamp(tick_val)
tformat = CFG("timeformat_x_axis").replace('$','%')
return dt.strftime(tformat)

12
plot_imageutils.py Normal file
View File

@@ -0,0 +1,12 @@
#!/usr/bin/python3
from config_parse import CFG
from PIL import Image
import math
def check_and_rotate(path):
img = Image.open(path)
div=abs(float(img.size[1])/float(img.size[0])-a4_aspect())/a4_aspect()*100
print("Seitenverhältnisabweichung zu A4: %.2f"%div+r'%')
img.rotate(CFG("image_rotation"),expand=True).save(path.strip(".png")+"_rotated.png")
def a4_aspect():
return 1/math.sqrt(2)

68
plot_main.py Normal file
View File

@@ -0,0 +1,68 @@
#!/usr/bin/python3
import sys
from config_parse import CFG
from constants import *
from datetime import datetime, timedelta
from frontend_utils import open_file
from constants import *
import math
import matplotlib
matplotlib.use(CFG("use_gui_backend"))
import matplotlib.pyplot as plt
import matplotlib.dates
import matplotlib.ticker as ticker
import plot_graphutils
import plot_imageutils
import plot_timeutils
def plot(datapoints,path=None,date1=None,date2=None):
plotname = "" if CFG("name_of_plot") == "None" else CFG("name_of_plot")
tup = [None,None,plot_timeutils.between_dates,plotname]
if CFG("enable_multicore_support"):
thread = Process(target=__plot,args=(tup,datapoints,date1,date2))
thread.start()
else:
__plot(tup,datapoints,path,date1,date2)
def __plot(tup,datapoints,path,date1=None,date2=None):
NO_SERIES = True
x,y,ymin,ymax,unix_x,major_xticks = ( [] , [], -1 , -1 , [], [] )
lw = CFG("plot_line_width")
ls = CFG("plot_line_style")
tup[FIGURE],tup[AXIS] = plt.subplots(1, 1)
for g in datapoints.values():
#### GET AND CHECK TIMEFRAMES ####
x,y, = g.get_timeframe(tup[CALLBACK],date1,date2)
if len(x) <= 0 or len(y) <= 0:
print("Warning: Empty series of data '%s' (wrong start/end time?)"%g.name)
continue
else:
NO_SERIES = False
unix_x = list(map(plot_timeutils.unix,x))
ymin,ymax = plot_graphutils.getlimits_y(y)
#### GET LINE STYLES ####
legend_label = plot_graphutils.legend_box_contents(g.name,y)
tup[AXIS].plot(unix_x, y,ls=ls,lw=lw,marker="None", label=legend_label, color=g.color)
if NO_SERIES:
print("Error: no data, nothing to plot. cannot continue. exit.")
sys.exit(1)
## GRID ##
plot_graphutils.general_background_setup(tup,ymin,ymax,x)
## using unix_x relys on unix_x to be the same for all plots ##
if path == None:
path = open_file()
## TODO function for picpathn for picpath
pic_path = path + ".png"
tup[FIGURE].savefig(pic_path,dpi=CFG("outfile_resolution_in_dpi"), bbox_inches='tight',transparent=CFG("transparent_background"))
### do operations on the finished png ###
plot_imageutils.check_and_rotate(pic_path)

39
plot_timeutils.py Normal file
View File

@@ -0,0 +1,39 @@
#!/usr/bin/python3
from config_parse import CFG
from datetime import datetime, timedelta
def between_dates(t,date1,date2):
return t if (date1 == None or date1 < t) and (date2 == None or date2 > t) else None
def time_from_dbf(l,timeformat):
timeformat=None #dont need that here
offset_d = datetime(1970,1,1)-datetime(1900,1,1)
shit_epoch = l*24*60*60 #days to seconds
unix_epoch = datetime.fromtimestamp(shit_epoch)-offset_d
return (unix_epoch-timedelta(days=2)+timedelta(hours=CFG("add_hours_to_input"))).replace(microsecond=0)
def time_from_csv(l,timeformat):
return datetime.strptime(l,timeformat)
def unix(dt):
return dt.timestamp()
def round_time_to_step(start,step):
start += step / 2
discard = timedelta(days=0)
hround = int(step.seconds/3600)
mround = int(step.seconds/60)
if step >= timedelta(days=1):
discard = timedelta(days=start.day % step.days,hours=start.hour,minutes=start.minute,seconds=start.second)
elif step >= timedelta(hours=1):
if hround != 0:
discard = timedelta(hours=start.hour % hround,minutes=start.minute,seconds=start.second)
elif step >= timedelta(minutes=1):
if mround != 0:
discard = timedelta(minutes=start.minute % mround,seconds=start.second)
elif step >= timedelta(seconds=1):
discard = timedelta(seconds=start.second % step.seconds)
else:
raise ValueError("Rounding time failed, this actually should be impossible. wtf. ("+str(start)+","+str(step)+","+str(discard)+")")
start -= discard
return start

2
req.sh Normal file
View File

@@ -0,0 +1,2 @@
#!/bin/bash
python3 -m pip install matplotlib sys configparser datetime os dbfread multiprocessing tkinter PIL

0
test_cases/__init__.py Normal file
View File

22
test_cases/graphutils_test.py Normal file
View File

@@ -0,0 +1,22 @@
import unittest
from datetime import datetime
from datetime import timedelta
import random
import itertools
class Graphutils_Test(unittest.TestCase):
def test_get_y_limits(self):
import plot_graphutils as gu
y_axis_values = [[32.3, 60.3, 35.1, 34.9, 33.0, 32.0, 32.7, 32.4, 34.0, 32.7, 33.2, 32.7, 33.2, 32.4, 34.0, \
32.9, 33.4, 32.2, 30.8, 41.6, 34.7, 32.6, 35.1, 33.5, 32.5, 37.6, 32.6, 32.3, 31.3, 33.0, 34.0,\
32.7, 32.7, 32.4, 32.8, 34.0, 34.1, 32.5, 33.5, 33.8, 31.0, 32.8, 34.9],[6.4, 15.3, 7.9, 7.1,\
6.9, 6.4, 7.0, 6.4, 6.8, 5.8, 6.1, 6.6, 6.1, 6.8, 6.8, 5.9, 6.9, 6.3, 6.9, 12.2, 7.9, 6.5, 7.9,\
6.9, 6.5, 8.9, 6.8, 6.4, 6.4, 6.0, 6.7, 6.5, 7.0, 6.4, 6.7, 7.6, 6.9, 6.5, 7.2, 6.9, 4.9, 6.7,\
7.9],[24.0, 23.5, 24.4, 23.5, 24.3, 24.2, 24.5, 24.0, 23.6, 23.2, 23.2, 24.1, 23.2, 24.5, 23.6,\
23.2, 24.1, 24.0, 25.4, 26.3, 24.5, 24.0, 24.4, 24.0, 24.0, 24.4, 24.4, 24.0, 24.6, 23.2, 23.5,\
24.0, 24.5, 24.0, 24.1, 24.5, 23.7, 24.0, 24.4, 23.8, 23.0, 24.1, 24.5]]
yl_values = [(0,95),(0,95),(0,95)]
for y,yl in zip(y_axis_values,yl_values):
self.assertEqual(yl,gu.getlimits_y(y))

58
test_cases/timeutils_test.py Normal file
View File

@@ -0,0 +1,58 @@
import unittest
from datetime import datetime
from datetime import timedelta
import random
import itertools
class Timeutil_Test(unittest.TestCase):
DATES = []
STEPS = [ (timedelta(hours=1)),timedelta(hours=4),timedelta(days=1),timedelta(minutes=1),timedelta(minutes=3)]
def setUpClass():
random.seed("0")
for x in range(0,10000):
tmp = datetime(2018,1,1) + ( (random.random()-0.5) * timedelta(days=2*365) )
tmp = tmp.replace(microsecond=0)
Timeutil_Test.DATES += [ tmp ]
def test_between_dates(self):
import plot_timeutils as tu
d = Timeutil_Test.DATES
count = 0
while(count < len(d)-2):
t = d[count+0]
d1 = d[count+1]
d2 = d[count+2]
self.assertEqual(btw_wrapper(tu.between_dates(t,d1,d2), t), d1 < t < d2, "t: "+str(t)+", d1: "+str(d1)+", d2: "+str(d2) )
self.assertEqual(btw_wrapper(tu.between_dates(t,d1,None), t), d1 < t , "t: "+str(t)+", d1: "+str(d1)+", d2: "+str(d2) )
self.assertEqual(btw_wrapper(tu.between_dates(t,None,d1), t), d1 > t , "t: "+str(t)+", d1: "+str(d1)+", d2: "+str(d2) )
self.assertEqual(btw_wrapper(tu.between_dates(t,None,None),t), True , "t: "+str(t)+", d1: "+str(d1)+", d2: "+str(d2) )
count+=1
def test_parse_time_dbf(self):
import plot_timeutils as tu
ind = [43121.6821296,43121.6856018,43121.689074,43121.6925462,43121.6960185,43121.6994907,43121.7029629,43121.7064351,43121.7099074,43121.7133796,43121.7168518,43121.720324,43121.7237962,43121.7272685,43121.7307407,43121.7342129,43121.7376851,43121.7411574,43121.7446296,43121.7481018,43121.751574,43121.7550462,43121.7585185,43121.7619907,43121.7654629,43121.7689351,43121.7724074,43121.7758796,43121.7793518,43121.782824,43121.7862962,43121.7897685,43121.7932407,43121.7967129,43121.8001851,43121.8036574,43121.8071296,43121.8106018,43121.814074,43121.8175462,43121.8210185,43121.8244907,43121.8279629]
outd = ["2018-01-21 18:22:15","2018-01-21 18:27:15","2018-01-21 18:32:15","2018-01-21 18:37:15","2018-01-21 18:42:15","2018-01-21 18:47:15","2018-01-21 18:52:15","2018-01-21 18:57:15","2018-01-21 19:02:15","2018-01-21 19:07:15","2018-01-21 19:12:15","2018-01-21 19:17:15","2018-01-21 19:22:15","2018-01-21 19:27:15","2018-01-21 19:32:15","2018-01-21 19:37:15","2018-01-21 19:42:15","2018-01-21 19:47:15","2018-01-21 19:52:15","2018-01-21 19:57:15","2018-01-21 20:02:15","2018-01-21 20:07:15","2018-01-21 20:12:15","2018-01-21 20:17:15","2018-01-21 20:22:15","2018-01-21 20:27:15","2018-01-21 20:32:15","2018-01-21 20:37:15","2018-01-21 20:42:15","2018-01-21 20:47:15","2018-01-21 20:52:15","2018-01-21 20:57:15","2018-01-21 21:02:15","2018-01-21 21:07:15","2018-01-21 21:12:15","2018-01-21 21:17:15","2018-01-21 21:22:15","2018-01-21 21:27:15","2018-01-21 21:32:15","2018-01-21 21:37:15","2018-01-21 21:42:15","2018-01-21 21:47:15","2018-01-21 21:52:15"]
for i,o in zip(ind,outd):
self.assertEqual(str(tu.parse_time_dbf(i)),o)
def test_round_time_to_step(self):
import plot_timeutils as tu
for s in Timeutil_Test.STEPS:
for d in Timeutil_Test.DATES:
rounded = tu.round_time_to_step(d,s)
if s < timedelta(minutes=60):
self.assertEquals(rounded.minute * 60 % s.seconds ,0,'date: '+str(d)+' rounded: '+str(rounded)+' step: '+str(s))
elif s < timedelta(hours=24):
self.assertEquals(rounded.hour*60*60 % s.seconds ,0,'date: '+str(d)+' rounded: '+str(rounded)+' step: '+str(s))
elif s >= timedelta(days=1):
self.assertEquals(rounded.day % s.days ,0,'date: '+str(d)+' rounded: '+str(rounded)+' step: '+str(s))
else:
raise AssertionError(int(s.days),0,'date: '+str(d)+' rounded: '+str(rounded)+' step: '+str(s))
def btw_wrapper(inp,a):
if inp == a:
return True
return False

107
ths_config.txt Normal file
View File

@@ -0,0 +1,107 @@
# Ich bin ein Kommentar, das Programm ignoriert Zeilen die mit '#' beginnen.
# Die Zeile '[plot] darf NICHT gelöscht oder verändert werden!
[plot]
plot_humidity = True
plot_temperatur = True
plot_dewcels = False
show_avg = True
show_min = True
show_max = True
raster = True
draw_thresholds = True
interactive = True
language = DE
default_source_path = /home/ik15ydit/reps/random-code/ths/
default_target_dir = /home/ik15ydit/reps/random-code/ths/
output_filename = test.png
humidity_critical = 55
humidity_warning = 50
acceptable_temp_low = 18
acceptable_temp_high = 22
target_temperatur = 20
temperatur_plot_name = Innenlufttemperatur
humidity_plot_name = rel. Luftfeuchtigkeit
dewcels_plot_name = Taupunkt
y_label = Temp./r.L.
x_label = Datum/Uhrzeit
font = calibri
global_text_size = 7
xticks_font_size = 5
yticks_font_size = 5
timeformat_x_axis = '$d.$m, $H:$M'
acceptable_x_intervals = 1m,5m,10m,30m,1h,2h,4h,6h
image_rotation = 90
transparent_background = no
name_of_plot = None
aspect_ratio = A4
############# eye_candy #############
empty_space_above_plot = 10
yaxis_minnimum_hight = 95
yaxis_start_value = 0
# True: die Y-Achse beginnt auch bei xaxis_start_value wenn dadurch Werte nicht angezeit werden
# False: wenn ein Wert im plot kleiner xaxis_start_value ist beginnt die Y-Achse beim kleinsten Wert im Plot
yaxis_force_start_value = False
# ein höheres alpha für zu einer stärkeren Sättigung der Hintergrundfarbe (0 und es ist ganz weg)
humidity_crit_alpha = 0.35
humidity_warning_alpha = 0.35
acceptable_temp_alpha = 0.20
# Farben: cyan, yellow, green, red, blue, black, white, grey oder RGBA-Wert
humidity_crit_color = red
humidity_warning_color = yellow
acceptable_temp_color = blue
# Farbe der linie des graphen
humidity_color = red
temperatur_color = blue
dewcels_color = green
plot_line_width = 0.5
plot_line_style = solid
hline_draw_lines = True
# linestyles: https://matplotlib.org/devdocs/gallery/lines_bars_and_markers/line_styles_reference.html
hline_line_style = --
hline_line_width = 0.5
grid_line_style = :
grid_line_width = 0.15
grid_line_color = black
############# technical #############
enable_multicore_support = False
raster_alligment_auto = True
raster_hline_prefered_interval = 5
raster_minimum_hlines = 10
y_tick_interval = 5
outfile_resolution_in_dpi = 200
legend_location = upper right
terminate_on_missing_input_file = True
terminate_on_fail = True
add_hours_to_input = 1
prefered_total_xticks = 24
use_gui_backend = Agg
add_x_labels_at_end = 1
xticks_label_degree = 45
major_line_width = 0.5
legend_font_size = 5
minor_xticks_per_major = 5
terminate_on_warning = no
###### DEBUGGING ######
no_ask_date_input = no
input_filename = test.dbf
use_input_filename = no
debug_no_interactive = no
###### THINGS THERE IS REALLY NO REASON TO CHANGE ######
plot_temperatur_key = TEMP
plot_humidity_key = HUMIDITY
plot_dewcels_key = DEWCELS
always_allow_days_as_xticks = yes