modm_data.dl.stmicro

13class Document:
14    """
15    Smaller helper class to store all relevant information about a remote
16    STMicro PDF document and how to download it.
17    """
18
19    def __init__(self, data):
20        # Primary information from JSON data
21        self._title = data["title"]
22        self._description = data["localizedDescriptions"]["en"]
23        self._url = data["localizedLinks"]["en"]
24        self._version = data["version"]
25        self._update = data["latestUpdate"]
26        self._type = data.get("resourceType", "Unknown")
27
28        # Derived information
29        self._short_type = self._title[:2]
30        clean_version = self._version.replace(".0", "").replace(".", "_")
31        self.name = self._title + "-v" + clean_version
32        """The full name of the document including version."""
33        self.filename = self.name + ".pdf"
34        """The leaf filename of the document."""
35        self.url = "https://www.st.com" + self._url
36        """The full URL that the file was downloaded from."""
37
38    @property
39    def data(self) -> dict:
40        """
41        A dictionary uniquely identifying this document version in a similar
42        format to the one used by the STMicro homepage. This can be used to
43        keep track of which documents have already been downloaded and which
44        have been updated upstream.
45        """
46        return {
47            "title": self._title,
48            "localizedDescriptions": {"en": self._description},
49            "localizedLinks": {"en": self._url},
50            "version": self._version,
51            "latestUpdate": self._update,
52            "resourceType": self._type,
53        }
54
55    def store_pdf(self, path: str, overwrite: bool = False) -> bool:
56        """Download the PDF file to the path, optionally overwriting it."""
57        return download_file(self.url, path, overwrite=overwrite)
58
59    def __repr__(self) -> str:
60        return f"Doc({self._title} v{self._version.replace('.0', '')})"
61
62    def __eq__(self, other) -> bool:
63        if not isinstance(other, type(self)):
64            return False
65        return self.filename == other.filename
66
67    def __hash__(self) -> int:
68        return hash(self.filename)

117def load_remote_info(base_dir: Path, use_cached: bool = False) -> list[dict]:
118    info = base_dir / _remote_info
119    if use_cached and info.exists():
120        LOGGER.debug("Loading remote info from cache")
121        docs = json.loads(info.read_text())
122    else:
123        LOGGER.info("Downloading remote info")
124        docs = []
125        for urls in _json_urls.values():
126            for url in urls:
127                docs.extend(json.loads(download_data(url))["rows"])
128    missing = [d for d in docs if "title" not in d or "version" not in d]
129    if missing:
130        LOGGER.warning(
131            "Some documents are missing title or version information and will be ignored: %s",
132            str(missing),
133        )
134    return [d for d in docs if "title" in d and "version" in d]

144def load_local_info(base_dir: Path) -> list[dict]:
145    info = base_dir / _local_info
146    if info.exists():
147        LOGGER.debug("Loading local info from cache")
148        return json.loads(info.read_text())
149    return []

137def store_remote_info(base_dir: Path, docs: list[dict]):
138    info = base_dir / _remote_info
139    info.parent.mkdir(parents=True, exist_ok=True)
140    exit(1)
141    info.write_text(json.dumps(sorted(docs, key=lambda d: (d["title"], d["version"])), indent=4, sort_keys=True))

152def store_local_info(base_dir: Path, docs: list[dict]):
153    info = base_dir / _local_info
154    info.parent.mkdir(parents=True, exist_ok=True)
155    info.write_text(json.dumps(sorted(docs, key=lambda d: (d["title"], d["version"])), indent=4, sort_keys=True))

28def download_cubemx(extraction_path: Path, with_download: bool = True, with_patch: bool = True) -> bool:
29    # First check STMUpdaterDefinitions.xml from this zip
30    update_url = "https://sw-center.st.com/packs/resource/utility/updaters.zip"
31    update_url2 = "https://www.ebuc23.com/s3/stm_test/software/utility/updaters.zip"  # noqa: F841
32    # Then Release="MX.6.2.0" maps to this: -win, -lin, -mac
33    cube_url = "https://sw-center.st.com/packs/resource/library/stm32cube_mx_v{}-lin.zip"
34    cube_url2 = "https://www.ebuc23.com/s3/stm_test/software/library/stm32cube_mx_v{}-lin.zip"  # noqa: F841
35
36    if with_download:
37        LOGGER.info("Downloading Update Info...")
38        # try:
39        #     urllib.request.urlopen(urllib.request.Request(update_url, headers=_hdr))
40        # except:
41        #     update_url = update_url2
42        #     cube_url = cube_url2
43        # LOGGER.debug(update_url)
44        # time.sleep(random.randrange(0,2))
45
46        z = zipfile.ZipFile(io.BytesIO(_dl(update_url)))
47        with io.TextIOWrapper(z.open("STMUpdaterDefinitions.xml"), encoding="utf-8") as defs:
48            version = re.search(r'Release="MX\.(.*?)"', defs.read())
49            version = version.group(1).replace(".", "")
50
51        shutil.rmtree(extraction_path / "mcu", ignore_errors=True)
52        shutil.rmtree(extraction_path / "plugins", ignore_errors=True)
53        extraction_path.mkdir(exist_ok=True, parents=True)
54
55        LOGGER.info("Downloading Database...")
56        LOGGER.debug(cube_url.format(version))
57        time.sleep(random.randrange(1, 6))
58
59        z = zipfile.ZipFile(io.BytesIO(_dl(cube_url.format(version))))
60        LOGGER.info("Extracting Database...")
61        for file in z.namelist():
62            if any(file.startswith(prefix) for prefix in ("MX/db/mcu", "MX/db/plugins")):
63                z.extract(file, extraction_path)
64
65        LOGGER.info("Moving Database...")
66        shutil.move(extraction_path / "MX/db/mcu", extraction_path / "mcu")
67        shutil.move(extraction_path / "MX/db/plugins", extraction_path / "plugins")
68        shutil.rmtree(extraction_path / "MX", ignore_errors=True)
69
70        LOGGER.info("Normalizing file endings...")
71        for file in Path(extraction_path).glob("**/*"):
72            if str(file).endswith(".xml"):
73                with file.open("r", newline=None, encoding="utf-8", errors="replace") as rfile:
74                    content = [line.rstrip() + "\n" for line in rfile.readlines()]
75                with file.open("w", encoding="utf-8") as wfile:
76                    wfile.writelines(content)
77
78    if with_patch:
79        LOGGER.info("Patching Database...")
80        from . import data
81
82        return pkg_apply_patch(data, "cubemx.patch", extraction_path)
83
84    return True

19def download_cubeprog(extraction_path: Path, with_download: bool = True) -> bool:
20    # com.st.stm32cube.ide.mcu.productdb.debug contains all the SVD files
21    # com.st.stm32cube.ide.mcu.externaltools.cubeprogrammer.linux64 contains the CubeProgrammer database
22    # com.st.stm32cube.common.mx contains the CubeMX database
23    base_url = "http://sw-center.st.com/stm32cubeide/updatesite1/"
24    cubeprog_id = "com.st.stm32cube.ide.mcu.externaltools.cubeprogrammer.linux64"
25
26    if with_download:
27        LOGGER.info("Downloading Update Info...")
28        with io.TextIOWrapper(
29            io.BytesIO(_dl(f"{base_url}/compositeContent.xml")), encoding="utf-8"
30        ) as compositeContent:
31            update_version = re.findall(r"child location='(\d+\.\d+\.\d+)'", compositeContent.read())
32            update_version = sorted(update_version, key=lambda v: tuple(map(int, v.split("."))), reverse=True)[0]
33
34        artifacts = lzma.LZMADecompressor().decompress(_dl(f"{base_url}/{update_version}/artifacts.xml.xz"))
35        version = re.search(f"id='{cubeprog_id}' version='(.*?)'", artifacts.decode("utf-8")).group(1)
36
37        download_url = f"{base_url}/{update_version}/plugins/{cubeprog_id}_{version}.jar"
38        shutil.rmtree(extraction_path / "Data_Base", ignore_errors=True)
39        extraction_path.mkdir(exist_ok=True, parents=True)
40
41        LOGGER.info("Downloading Database...")
42        LOGGER.debug(download_url)
43        time.sleep(random.randrange(1, 6))
44
45        z = zipfile.ZipFile(io.BytesIO(_dl(download_url)))
46        LOGGER.info("Extracting Database...")
47        for file in z.namelist():
48            if file.startswith("tools/Data_Base"):
49                z.extract(file, extraction_path)
50
51        LOGGER.info("Moving Database...")
52        shutil.move(extraction_path / "tools/Data_Base", extraction_path / "Data_Base")
53        shutil.rmtree(extraction_path / "tools", ignore_errors=True)
54
55        LOGGER.info("Normalizing file endings...")
56        for file in Path(extraction_path).glob("**/*"):
57            if str(file).endswith(".xml"):
58                with file.open("r", newline=None, encoding="utf-8", errors="replace") as rfile:
59                    content = [line.rstrip() + "\n" for line in rfile.readlines()]
60                with file.open("w", encoding="utf-8") as wfile:
61                    wfile.writelines(content)
62
63    return True