Source code for cowidev.vax.incremental.isle_of_man

import json

import requests
import pandas as pd

from cowidev.utils.clean.dates import localdate
from cowidev.vax.utils.incremental import enrich_data, increment
from cowidev.vax.utils.base import CountryVaxBase


[docs]class IsleOfMan(CountryVaxBase): location: str = "Isle of Man" source_url: str = ( "https://wabi-west-europe-b-primary-api.analysis.windows.net/public/reports/querydata?synchronous=true" ) source_url_ref: str = "https://covid19.gov.im/general-information/covid-19-vaccination-statistics/" metrics_mapping: dict = { "First dose": "people_vaccinated", "Second dose": "people_fully_vaccinated", "Booster dose": "total_boosters", "Third dose (NOT booster)": "third_dose", }
[docs] def read(self) -> pd.Series: data = self.data_body df = json.loads(requests.post(self.source_url, headers=self.headers, data=data).content)["results"][0][ "result" ]["data"]["dsr"]["DS"][0]["PH"][0]["DM0"] return self._parse_data(df)
@property def headers(Self): return { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:85.0) Gecko/20100101 Firefox/85.0", "Accept": "application/json, text/plain, */*", "Accept-Language": "fr,fr-FR;q=0.8,en-US;q=0.5,en;q=0.3", "X-PowerBI-ResourceKey": "a01ccc19-80ba-4458-9849-66d7e1e300b7", "Content-Type": "application/json;charset=UTF-8", "Origin": "https://app.powerbi.com", "Referer": ( "https://app.powerbi.com/view?r=eyJrIjoiYTAxY2NjMTktODBiYS00NDU4LTk4NDktNjZkN2UxZTMwMGI3IiwidCI6IjM" "5YzAwODM2LWVkMTItNDhkYS05Yjk3LTU5NGQ4MDhmMDNlNSIsImMiOjl9" ), "Pragma": "no-cache", "Cache-Control": "no-cache", } @property def data_body(self): values = ["""[{{"Literal":{{"Value":"\'{}\'"}}}}]""".format(m) for m in self.metrics_mapping.keys()] values_2 = [ """[{{\\"Literal\\":{{\\"Value\\":\\"\'{}\'\\"}}}}]""".format(m) for m in self.metrics_mapping.keys() ] data = '{"version":"1.0.0","queries":[{"Query":{"Commands":[{"SemanticQueryDataShapeCommand":{"Query":{"Version":2,"From":[{"Name":"m","Entity":"Medway","Type":0}],"Select":[{"Column":{"Expression":{"SourceRef":{"Source":"m"}},"Property":"Dose schedule"},"Name":"Medway.Dose schedule"},{"Aggregation":{"Expression":{"Column":{"Expression":{"SourceRef":{"Source":"m"}},"Property":"Date of vaccination"}},"Function":5},"Name":"CountNonNull(Medway.Date of vaccination)"}],"Where":[{"Condition":{"In":{"Expressions":[{"Column":{"Expression":{"SourceRef":{"Source":"m"}},"Property":"Dose schedule"}}],"Values":[__VALUES__]}}}],"OrderBy":[{"Direction":2,"Expression":{"Aggregation":{"Expression":{"Column":{"Expression":{"SourceRef":{"Source":"m"}},"Property":"Date of vaccination"}},"Function":5}}}]},"Binding":{"Primary":{"Groupings":[{"Projections":[0,1]}]},"DataReduction":{"DataVolume":4,"Primary":{"Window":{"Count":1000}}},"Version":1}}}]},"CacheKey":"{\\"Commands\\":[{\\"SemanticQueryDataShapeCommand\\":{\\"Query\\":{\\"Version\\":2,\\"From\\":[{\\"Name\\":\\"m\\",\\"Entity\\":\\"Medway\\",\\"Type\\":0}],\\"Select\\":[{\\"Column\\":{\\"Expression\\":{\\"SourceRef\\":{\\"Source\\":\\"m\\"}},\\"Property\\":\\"Dose schedule\\"},\\"Name\\":\\"Medway.Dose schedule\\"},{\\"Aggregation\\":{\\"Expression\\":{\\"Column\\":{\\"Expression\\":{\\"SourceRef\\":{\\"Source\\":\\"m\\"}},\\"Property\\":\\"Date of vaccination\\"}},\\"Function\\":5},\\"Name\\":\\"CountNonNull(Medway.Date of vaccination)\\"}],\\"Where\\":[{\\"Condition\\":{\\"In\\":{\\"Expressions\\":[{\\"Column\\":{\\"Expression\\":{\\"SourceRef\\":{\\"Source\\":\\"m\\"}},\\"Property\\":\\"Dose schedule\\"}}],\\"Values\\":[__VALUES2__]}}}],\\"OrderBy\\":[{\\"Direction\\":2,\\"Expression\\":{\\"Aggregation\\":{\\"Expression\\":{\\"Column\\":{\\"Expression\\":{\\"SourceRef\\":{\\"Source\\":\\"m\\"}},\\"Property\\":\\"Date of vaccination\\"}},\\"Function\\":5}}}]},\\"Binding\\":{\\"Primary\\":{\\"Groupings\\":[{\\"Projections\\":[0,1]}]},\\"DataReduction\\":{\\"DataVolume\\":4,\\"Primary\\":{\\"Window\\":{\\"Count\\":1000}}},\\"Version\\":1}}}]}","QueryId":"","ApplicationContext":{"DatasetId":"819a1554-706f-4e7e-9f7d-ec4bf4a353e2","Sources":[{"ReportId":"a1d3f3f4-2b99-4dda-82af-e751394400c5"}]}}],"cancelQueries":[],"modelId":1616759}'.replace( "__VALUES__", ",".join(values) ).replace( "__VALUES2__", ",".join(values_2) ) return data
[docs] def _parse_data(self, df: pd.DataFrame) -> pd.Series: data = { value: [elem["C"][1] for elem in df if elem["C"][0] == key][0] for key, value in self.metrics_mapping.items() } return pd.Series(data=data)
[docs] def pipe_metrics(self, ds: pd.Series) -> pd.Series: ds["total_boosters"] += ds["third_dose"] ds = enrich_data( ds, "total_vaccinations", ds["people_vaccinated"] + ds["people_fully_vaccinated"] + ds["total_boosters"] ) return ds
[docs] def pipe_date(self, ds: pd.Series) -> pd.Series: date = localdate("Europe/Isle_of_Man") return enrich_data(ds, "date", date)
[docs] def pipe_location(self, ds: pd.Series) -> pd.Series: return enrich_data(ds, "location", self.location)
[docs] def pipe_vaccine(self, ds: pd.Series) -> pd.Series: return enrich_data(ds, "vaccine", "Moderna, Oxford/AstraZeneca, Pfizer/BioNTech")
[docs] def pipe_source(self, ds: pd.Series) -> pd.Series: return enrich_data( ds, "source_url", self.source_url_ref, )
[docs] def pipeline(self, ds: pd.Series) -> pd.Series: return ( ds.pipe(self.pipe_metrics) .pipe(self.pipe_date) .pipe(self.pipe_location) .pipe(self.pipe_vaccine) .pipe(self.pipe_source) )
[docs] def export(self): data = self.read().pipe(self.pipeline) increment( location=self.location, total_vaccinations=data["total_vaccinations"], people_vaccinated=data["people_vaccinated"], people_fully_vaccinated=data["people_fully_vaccinated"], total_boosters=data["total_boosters"], date=data["date"], source_url=data["source_url"], vaccine=data["vaccine"], ) self.force_monotonic()
[docs]def main(): IsleOfMan().export()