Nemcio/DemonEditor
1
0
Fork 0
mirror of https://github.com/DYefremov/DemonEditor.git synced 2025-12-22 16:39:42 +01:00
Code Issues Packages Projects Releases Activity

basic dvb-t and cable support for neutrino

Browse Source
This commit is contained in:
DYefremov
2021-08-23 14:34:56 +03:00
parent 56f1c75a41
commit 8383214e15
4 changed files with 285 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
app/eparser/neutrino/bouquets.py
View File

@@ -1,3 +1,31 @@
# -*- coding: utf-8 -*-
#
# The MIT License (MIT)
#
# Copyright (c) 2018-2021 Dmitriy Yefremov
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# Author: Dmitriy Yefremov
#
import os
from xml.dom.minidom import parse, Document

329
app/eparser/neutrino/services.py
View File

@@ -1,160 +1,247 @@
from xml.dom.minidom import parse, Document
# -*- coding: utf-8 -*-
#
# The MIT License (MIT)
#
# Copyright (c) 2018-2021 Dmitriy Yefremov
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# Author: Dmitriy Yefremov
#
import re
from collections import defaultdict
from xml.dom.minidom import Document, parseString
from app.commons import log
from ..ecommons import Service, POLARIZATION, FEC, SYSTEM, SERVICE_TYPE, PROVIDER
from app.eparser.ecommons import (Service, POLARIZATION, FEC, SYSTEM, SERVICE_TYPE, PROVIDER, T_SYSTEM, TrType,
SystemCable)
_FILE = "services.xml"
_TR_ATTR_NAMES = ("id", "on", "frq", "inv", "sr", "fec", "pol", "mod", "sys") # transponder attributes
_SRV_ATTR_NAMES = ("t", "s", "num", "f", "v", "a", "p", "pmt", "tx", "vt") # service attributes
SP = "_:::_"
KSP = "_::_"
def write_services(path, services):
doc = Document()
root = doc.createElement("zapit")
root.setAttribute("api", "4")
doc.appendChild(root)
comment = doc.createComment(" File was created in DemonEditor. Enjoy watching! ")
doc.appendChild(comment)
sats = {}
for srv in services:
flag = srv[0]
if flag in sats:
sats.get(flag).append(srv)
else:
srv_list = [srv]
sats[flag] = srv_list
for sat in sats:
tr_atr = sat.split(":")
sat_elem = doc.createElement("sat")
sat_elem.setAttribute("name", tr_atr[0])
sat_elem.setAttribute("position", tr_atr[1])
sat_elem.setAttribute("diseqc", tr_atr[2])
sat_elem.setAttribute("uncommited", tr_atr[3])
root.appendChild(sat_elem)
transponers = {}
for srv in sats.get(sat):
flag = srv[-1]
if flag in transponers:
transponers.get(flag).append(srv)
else:
srv_list = [srv]
transponers[flag] = srv_list
for tr in transponers:
tr_elem = doc.createElement("TS")
tr_atr = tr.split(":")
for i, value in enumerate(tr_atr):
if value == "None":
continue
tr_elem.setAttribute(_TR_ATTR_NAMES[i], value)
sat_elem.appendChild(tr_elem)
for srv in transponers.get(tr):
srv_elem = doc.createElement("S")
srv_elem.setAttribute("i", srv.ssid)
srv_elem.setAttribute("n", srv.service)
srv_attrs = srv.data_id.split(":")
api = srv_attrs.pop(0)
if api == "3":
root.setAttribute("api", "3") # !!!
for i, value in enumerate(srv_attrs):
if value == "None":
continue
srv_elem.setAttribute(_SRV_ATTR_NAMES[i], value)
tr_elem.appendChild(srv_elem)
doc.writexml(open(path + _FILE, "w"), addindent=" ", newl="\n", encoding="UTF-8")
doc.unlink()
NeutrinoServiceWriter(path, services).write()
def get_services(path):
return parse_services(path)
return NeutrinoServicesReader(path).get_services()
def parse_services(path):
""" Parsing services from xml"""
dom = parse(path + _FILE)
services = []
class NeutrinoServiceWriter:
def __init__(self, path, services):
self._path = path
self._services = services
self._api = "4"
self._doc = Document()
self._root = self._doc.createElement("zapit")
self._root.setAttribute("api", self._api)
self._doc.appendChild(self._root)
self._doc.appendChild(self._doc.createComment(" File was created in DemonEditor. Enjoy watching! "))
def write(self):
srvs = defaultdict(list)
for s in self._services:
srvs[s.transponder_type].append(s)
self.append_services(srvs.get(TrType.Satellite.value), "sat")
self.append_services(srvs.get(TrType.Terrestrial.value), "terrestrial")
self.append_services(srvs.get(TrType.Cable.value), "cable")
self._doc.writexml(open(self._path + _FILE, "w"), addindent=" ", newl="\n", encoding="UTF-8")
self._doc.unlink()
def append_services(self, services, s_type):
if not services:
return
sats = defaultdict(list)
for srv in services:
sats[srv[0]].append(srv)
for sat in sats:
sat_elem = self._doc.createElement(s_type)
attrs = self.get_attributes(sat)
for k, v in attrs.items():
sat_elem.setAttribute(k, v)
self._root.appendChild(sat_elem)
transponders = defaultdict(list)
for srv in sats.get(sat):
transponders[srv[-1]].append(srv)
for tr in transponders:
tr_elem = self._doc.createElement("TS")
for k, v in self.get_attributes(tr).items():
tr_elem.setAttribute(k, v)
sat_elem.appendChild(tr_elem)
for srv in transponders.get(tr):
srv_elem = self._doc.createElement("S")
s_attrs = self.get_attributes(srv.data_id)
api = s_attrs.pop("api", self._api)
if api != self._api:
self._root.setAttribute("api", api)
for k, v in s_attrs.items():
srv_elem.setAttribute(k, v)
tr_elem.appendChild(srv_elem)
@staticmethod
def get_attributes(data):
return {el[0]: el[1] for el in (e.split(KSP) for e in data.split(SP))}
class NeutrinoServicesReader:
def __init__(self, path):
self._path = path
self._attrs = None
self._tr = None
self._api = "4"
self._services = []
def get_services(self):
with open(self._path + _FILE, "rb") as f:
# Pre-processing is required to replace the '&' character.
dom = parseString(re.sub("&", "&", f.read().decode(encoding="utf-8", errors="ignore")))
for root in dom.getElementsByTagName("zapit"):
api = root.attributes["api"].value
if root.hasAttributes():
api = root.attributes["api"]
self._api = api.value if api else self._api
for elem in root.getElementsByTagName("sat"):
if elem.hasAttributes():
sat_name = elem.attributes["name"].value
sat_pos = elem.attributes["position"].value
diseqc = elem.attributes.get("diseqc")
diseqc = diseqc.value if diseqc else diseqc
uncommited = elem.attributes.get("uncommited")
uncommited = uncommited.value if uncommited else uncommited
sat = "{}:{}:{}:{}".format(sat_name, sat_pos, diseqc, uncommited)
sat_attrs = self.get_attributes(elem)
sat_attrs["name"] = re.sub("&", "&", sat_attrs.get("name", ""))
sat_pos = 0
try:
sat_pos = int(sat_attrs.get("position", "0"))
sat_pos = "{:0.1f}{}".format(abs(sat_pos / 10), "W" if sat_pos < 0 else "E")
except ValueError as e:
log("Neutrino parsing error [parse sat position]: {}".format(e))
sat = SP.join("{}{}{}".format(k, KSP, v) for k, v in sat_attrs.items())
for tr_elem in elem.getElementsByTagName("TS"):
if tr_elem.hasAttributes():
self.parse_sat_transponder(sat, sat_pos, tr_elem)
# Terrestrial DVB-T[2].
for elem in root.getElementsByTagName("terrestrial"):
if elem.hasAttributes():
terr_attrs = self.get_attributes(elem)
terr_attrs["name"] = re.sub("&amp;", "&", terr_attrs.get("name", ""))
terr = SP.join("{}{}{}".format(k, KSP, v) for k, v in terr_attrs.items())
for tr_elem in elem.getElementsByTagName("TS"):
if tr_elem.hasAttributes():
parse_transponder(api, sat, sat_pos, services, tr_elem)
self.parse_ct_transponder(terr, tr_elem, TrType.Terrestrial)
return services
# Cable.
for elem in root.getElementsByTagName("cable"):
if elem.hasAttributes():
cable_attrs = self.get_attributes(elem)
terr_attrs["name"] = re.sub("&amp;", "&", cable_attrs.get("name", ""))
cable = SP.join("{}{}{}".format(k, KSP, v) for k, v in cable_attrs.items())
for tr_elem in elem.getElementsByTagName("TS"):
if tr_elem.hasAttributes():
self.parse_ct_transponder(cable, tr_elem, TrType.Cable)
def parse_transponder(api, sat, sat_pos, services, tr_elem):
tr_id = tr_elem.attributes["id"].value
on = tr_elem.attributes["on"].value
freq = tr_elem.attributes["frq"].value
rate = tr_elem.attributes["sr"].value
inv = tr_elem.attributes["inv"].value
fec = tr_elem.attributes["fec"].value
pol = tr_elem.attributes["pol"].value
mod = tr_elem.attributes.get("mod")
mod = mod.value if mod else mod
sys = tr_elem.attributes.get("sys")
sys = sys.value if sys else sys
return self._services
tr = "{}:{}:{}:{}:{}:{}:{}:{}:{}".format(tr_id, on, freq, inv, rate, fec, pol, mod, sys)
tr_id = tr_id.lstrip("0")
pol = POLARIZATION.get(pol)
def parse_sat_transponder(self, sat, sat_pos, tr_elem):
tr_attr = self.get_attributes(tr_elem)
tr = SP.join("{}{}{}".format(k, KSP, v) for k, v in tr_attr.items())
tr_id = tr_attr.get("id", "0").lstrip("0")
on = tr_attr.get("on", "0")
freq = tr_attr.get("frq", "0")
rate = tr_attr.get("sr", "0")
fec = tr_attr.get("fec", "0")
pol = POLARIZATION.get(tr_attr.get("pol", "0"))
# Formatting displayed values.
try:
freq = "{}".format(int(freq) // 1000)
rate = "{}".format(int(rate) // 1000)
sat_pos = int(sat_pos)
sat_pos = "{:0.1f}{}".format(abs(sat_pos / 10), "W" if sat_pos < 0 else "E")
except ValueError as e:
log("Neutrino parsing error [parse_transponder]: {}".format(e))
for srv_elem in tr_elem.getElementsByTagName("S"):
if srv_elem.hasAttributes():
ssid = srv_elem.attributes["i"].value
name = srv_elem.attributes["n"].value
srv_type = srv_elem.attributes["t"].value
sys = srv_elem.attributes["s"].value
num = srv_elem.attributes.get("num")
num = num.value if num else num
f = srv_elem.attributes.get("f")
f = f.value if f else f
v, a, p, pmt, tx, vt = [None] * 6
# For v3 is possible so: '<S i="0001" n="name" t="1" s="0" num="770" f="4"/>' (equals v4 api)
if api == "3" and len(srv_elem.attributes) > 6:
v = srv_elem.attributes["v"].value
a = srv_elem.attributes["a"].value
p = srv_elem.attributes["p"].value
pmt = srv_elem.attributes["pmt"].value
tx = srv_elem.attributes["tx"].value
vt = srv_elem.attributes["vt"].value
data_id = "{}:{}:{}:{}:{}:{}:{}:{}:{}:{}:{}".format(api, srv_type, sys, num, f, v, a, p, pmt, tx, vt)
at = self.get_attributes(srv_elem)
at["api"] = self._api
ssid, name, s_type, sys = at.get("i", "0"), at.get("n", ""), at.get("t", "3"), at.get("s", "0")
name = re.sub("amp;", "", name)
at["n"] = name
data_id = SP.join("{}{}{}".format(k, KSP, v) for k, v in at.items())
fav_id = "{}:{}:{}".format(tr_id, on.lstrip("0"), ssid.lstrip("0"))
picon_id = "{}{}{}.png".format(tr_id, on, ssid)
prv, st, = PROVIDER.get(int(on, 16)), SERVICE_TYPE.get(str(int(srv_type, 16)), SERVICE_TYPE.get("-2"))
prv = PROVIDER.get(int(on, 16), "")
st = SERVICE_TYPE.get(str(int(s_type, 16)), SERVICE_TYPE.get("-2"))
srv = Service(sat, None, None, name, None, None, prv, st, None, picon_id, ssid, freq, rate, pol,
FEC.get(fec), SYSTEM.get(sys), sat_pos, data_id, fav_id, tr)
services.append(srv)
srv = Service(sat, TrType.Satellite.value, None, name, None, None, prv, st, None, picon_id, ssid, freq,
rate, pol, FEC.get(fec), SYSTEM.get(sys), sat_pos, data_id, fav_id, tr)
self._services.append(srv)
def parse_ct_transponder(self, terr, tr_elem, tr_type):
attrs = self.get_attributes(tr_elem)
tr = SP.join("{}{}{}".format(k, KSP, v) for k, v in attrs.items())
tr_id, on, freq = attrs.get("id", "0").lstrip("0"), attrs.get("on", "0"), attrs.get("frq", "0")
for srv_elem in tr_elem.getElementsByTagName("S"):
if srv_elem.hasAttributes():
s_at = self.get_attributes(srv_elem)
s_at["api"] = self._api
ssid, name, s_type, sys = s_at.get("i", "0"), s_at.get("n", ""), s_at.get("t", "3"), s_at.get("s", "0")
name = re.sub("amp;", "", name)
s_at["n"] = name
data_id = SP.join("{}{}{}".format(k, KSP, v) for k, v in s_at.items())
fav_id = "{}:{}:{}".format(tr_id, on.lstrip("0"), ssid.lstrip("0"))
picon_id = "{}{}{}.png".format(tr_id, on, ssid)
prv = PROVIDER.get(int(on, 16), "")
st = SERVICE_TYPE.get(str(int(s_type, 16)), SERVICE_TYPE.get("-2"))
if tr_type is TrType.Terrestrial:
sys = T_SYSTEM.get(sys)
pos = "T"
elif tr_type is TrType.Cable:
sys = SystemCable(sys).name
pos = "C"
else:
log("Parse transponder error: Not supported type [{}]".format(tr_type))
break
srv = Service(terr, tr_type.value, None, name, None, None, prv, st, None, picon_id, ssid,
freq, "0", None, None, sys, pos, data_id, fav_id, tr)
self._services.append(srv)
@staticmethod
def get_attributes(attr):
attrs = attr.attributes
return {t: attrs[t].value for t in attrs.keys()}
if __name__ == "__main__":

4
app/ui/main.py
View File

@@ -1762,15 +1762,11 @@ class Application(Gtk.Application):
yield True
self._fav_model.clear()
yield True
s_model = self._services_view.get_model()
self._services_view.set_model(None)
yield True
for index, itr in enumerate([row.iter for row in self._services_model]):
self._services_model.remove(itr)
if index % self.DEL_FACTOR == 0:
yield True
yield True
self._services_view.set_model(s_model)
self._blacklist.clear()
self._services.clear()
self._rows_buffer.clear()

54
app/ui/service_details_dialog.py
View File

@@ -35,6 +35,7 @@ from app.eparser.ecommons import (MODULATION, Inversion, ROLL_OFF, Pilot, Flag,
get_value_by_name, FEC_DEFAULT, PLS_MODE, SERVICE_TYPE, T_MODULATION, C_MODULATION,
TrType, SystemCable, T_SYSTEM, BANDWIDTH, TRANSMISSION_MODE, GUARD_INTERVAL, T_FEC,
HIERARCHY, A_MODULATION)
from app.eparser.neutrino.services import NeutrinoServiceWriter, SP, KSP
from app.settings import SettingsType
from .dialogs import show_dialog, DialogType, Action, get_builder
from .main_helper import get_base_model
@@ -52,8 +53,6 @@ class ServiceDetailsDialog:
_NEUTRINO_FAV_ID = "{:x}:{:x}:{:x}"
_NEUTRINO_TRANSPONDER_DATA = "{:04x}:{:04x}:{}:{}:{}:{}:{}:{}:{}"
_DIGIT_ENTRY_ELEMENTS = ("bitstream_entry", "pcm_entry", "video_pid_entry", "pcr_pid_entry", "srv_type_entry",
"ac3_pid_entry", "ac3plus_pid_entry", "acc_pid_entry", "he_acc_pid_entry",
"teletext_pid_entry", "pls_code_entry", "stream_id_entry", "tr_flag_entry",
@@ -223,7 +222,6 @@ class ServiceDetailsDialog:
self._package_entry.set_text(srv.package)
self._sid_entry.set_text(str(int(srv.ssid, 16)))
# Transponder
if self._s_type is SettingsType.ENIGMA_2:
self._tr_type = TrType(srv.transponder_type)
self._freq_entry.set_text(srv.freq)
self._rate_entry.set_text(srv.rate)
@@ -353,10 +351,12 @@ class ServiceDetailsDialog:
# ***************** Init Neutrino data *********************#
def init_neutrino_data(self, srv):
tr_data = srv.transponder.split(":")
self._transponder_id_entry.set_text(str(int(tr_data[0], 16)))
self._network_id_entry.set_text(str(int(tr_data[1], 16)))
self.select_active_text(self._invertion_combo_box, Inversion(tr_data[3]).name)
if self._tr_type is not TrType.Satellite:
return
tr_data = NeutrinoServiceWriter.get_attributes(srv.transponder)
self._transponder_id_entry.set_text(str(int(tr_data.get("id", "0"), 16)))
self._network_id_entry.set_text(str(int(tr_data.get("on", "0"), 16)))
self.select_active_text(self._invertion_combo_box, Inversion(tr_data.get("inv", "2")).name)
self.select_active_text(self._service_type_combo_box, srv.service_type)
self.update_reference_entry()
@@ -368,6 +368,7 @@ class ServiceDetailsDialog:
tr_grid.set_margin_bottom(5)
self._builder.get_object("tr_extra_expander").set_visible(False)
self._builder.get_object("srv_separator").set_visible(False)
self._package_entry.set_sensitive(False)
# ***************** Init Sat positions *********************#
@@ -407,6 +408,10 @@ class ServiceDetailsDialog:
self.save_data()
def save_data(self):
if self._s_type is SettingsType.NEUTRINO_MP and self._tr_type is not TrType.Satellite:
show_dialog(DialogType.ERROR, transient=self._dialog, text="Not implemented yet!")
return
if not self.is_data_correct():
show_dialog(DialogType.ERROR, self._dialog, "Error. Verify the data!")
return
@@ -538,9 +543,9 @@ class ServiceDetailsDialog:
if self._s_type is SettingsType.ENIGMA_2:
return self.get_enigma2_flags()
elif self._s_type is SettingsType.NEUTRINO_MP:
flags = self._old_service.flags_cas.split(":")
flags[1] = self.get_sat_position()
return ":".join(flags)
flags = NeutrinoServiceWriter.get_attributes(self._old_service.flags_cas)
flags["position"] = self.get_sat_position()
return SP.join("{}{}{}".format(k, KSP, v) for k, v in flags.items())
def get_enigma2_flags(self):
flags = ["p:{}".format(self._package_entry.get_text())]
@@ -594,10 +599,12 @@ class ServiceDetailsDialog:
fav_id = self._ENIGMA2_FAV_ID.format(ssid, tr_id, net_id, namespace)
return fav_id, data_id
elif self._s_type is SettingsType.NEUTRINO_MP:
data = NeutrinoServiceWriter.get_attributes(self._old_service.data_id)
data["n"] = self._name_entry.get_text()
data["t"] = "{:x}".format(int(service_type))
data["i"] = "{:04x}".format(ssid)
fav_id = self._NEUTRINO_FAV_ID.format(tr_id, net_id, ssid)
data_id = self._old_service.data_id.split(":")
data_id[1] = "{:x}".format(int(service_type))
return fav_id, ":".join(data_id)
return fav_id, SP.join("{}{}{}".format(k, KSP, v) for k, v in data.items())
# ***************** Transponder ********************* #
@@ -642,12 +649,19 @@ class ServiceDetailsDialog:
pls_code = self._pls_code_entry.get_text()
st_id = self._stream_id_entry.get_text()
pls = ":{}:{}:{}".format(st_id, pls_code, pls_mode) if pls_mode and pls_code and st_id else ""
return "{}:{}:{}:{}:{}{}".format(dvb_s_tr, flag, mod, roll_off, pilot, pls)
elif self._s_type is SettingsType.NEUTRINO_MP:
on_id, tr_id = int(self._network_id_entry.get_text()), int(self._transponder_id_entry.get_text())
mod = self.get_value_from_combobox_id(self._mod_combo_box, MODULATION) if sys == "DVB-S2" else None
srv_sys = None
return self._NEUTRINO_TRANSPONDER_DATA.format(tr_id, on_id, freq, inv, rate, fec, pol, mod, srv_sys)
tr_data = NeutrinoServiceWriter.get_attributes(self._old_service.transponder)
tr_data["frq"] = freq
tr_data["sr"] = rate
tr_data["pol"] = pol
tr_data["fec"] = fec
tr_data["on"] = "{:04x}".format(int(self._network_id_entry.get_text()))
tr_data["id"] = "{:04x}".format(int(self._transponder_id_entry.get_text()))
tr_data["inv"] = inv
return SP.join("{}{}{}".format(k, KSP, v) for k, v in tr_data.items())
def get_sat_position(self):
sat_pos = self._sat_pos_button.get_value() * (-1 if self._pos_side_box.get_active_id() == "W" else 1)
@@ -707,9 +721,9 @@ class ServiceDetailsDialog:
continue
if self._s_type is SettingsType.NEUTRINO_MP:
flags = srv[Column.SRV_CAS_FLAGS].split(":")
flags[1] = sat_pos
srv[Column.SRV_CAS_FLAGS] = ":".join(flags)
flags = NeutrinoServiceWriter.get_attributes(srv[Column.SRV_CAS_FLAGS])
flags["position"] = sat_pos
srv[Column.SRV_CAS_FLAGS] = SP.join("{}{}{}".format(k, KSP, v) for k, v in flags.items())
self._services[fav_id] = Service(*srv[:Column.SRV_TOOLTIP])
self._current_model.set_row(itr, srv)